redis - 队列或其他方法来处理刻度数据？

Question

在我们的电子交易系统中，我们需要基于100多个合约的报价数据进行计算。

一条消息中未接收到合约的报价数据。一条消息仅包含一份契约(Contract)的报价数据。契约(Contract)的时间戳略有不同(有时差异很大，但让我们忽略这种情况)。

eg: (first column is timestamp. Second is contract name)
below 2 data has 1ms diff
10:34:03.235,10002007,510050C2006A03500  ,0.0546
10:34:03.236,10001909,510050C2003A02750  ,0.3888

below 2 data has 3ms diff
10:34:03.594,10002154,510300C2003M03700  ,0.4985
10:34:03.597,10002118,510300C2001M03700  ,0.4514

只有价格变动的人才能获得数据。因此，我无法计算契约(Contract)号以了解是否已收到此订单的所有数据。

但是另一方面，我们不想等到收到滴答声的所有数据后，因为有时数据可能会延迟很长时间，所以我们希望排除它们。

需要低延迟。因此，我认为我们将定义一个窗口(例如50毫秒)，并开始根据过去50毫秒内收到的任何数据进行计算。

处理此类用例的最佳方法是什么？

最初，我想使用redis流来维护一个小的队列，以便每当收到契约(Contract)的数据时，我都会将其推送到redis流。但是我无法确定在特定时间(例如50毫秒)过去后立即提取数据的最佳方法是什么。

我在考虑也许我应该使用其他一些技术？
任何建议表示赞赏。

Answer 1

使用XRANGE myStream - + COUNT 1获取第一个条目。

使用XREVRANGE myStream + - COUNT 1获取最后一个条目。
XINFO STREAM myStream也带来了第一个和最后一个条目，但是the docs说它是O(log N)。

假设您使用时间戳作为ID或字段，则可以计算时差。

如果您正在使用Redis Streams自动ID(XADD myStream * ...)，则ID的第一部分是UNIX时间戳(以毫秒为单位)。

假设以上内容，您可以使用Lua脚本自动进行检查:

EVAL "local first = redis.call('XRANGE', KEYS[1], '-', '+', 'COUNT', '1') local firstTime = {} if next(first) == nil then     return redis.error_reply('Stream is empty or key doesn`t exist') end for str in string.gmatch(first[1][1], '([^-]+)') do     table.insert(firstTime, tonumber(str)) end local last = redis.call('XREVRANGE', KEYS[1], '+', '-', 'COUNT', '1') local lastTime = {} for str in string.gmatch(last[1][1], '([^-]+)') do     table.insert(lastTime, tonumber(str)) end local ms = lastTime[1] - firstTime[1] if ms >= tonumber(ARGV[1]) then     return redis.call('XRANGE', KEYS[1], '-', '+') else     return redis.error_reply('Only '..ms..' ms') end" 1 myStream 50

参数为 numKeys(1 here) streamKey timeInMs(50 here): 1 myStream 50。

这里是Lua脚本的友好 View :

local first = redis.call('XRANGE', KEYS[1], '-', '+', 'COUNT', '1')
local firstTime = {}
if next(first) == nil then
    return redis.error_reply('Stream is empty or key doesn`t exist')
end
for str in string.gmatch(first[1][1], '([^-]+)') do
    table.insert(firstTime, tonumber(str))
end
local last = redis.call('XREVRANGE', KEYS[1], '+', '-', 'COUNT', '1')
local lastTime = {}
for str in string.gmatch(last[1][1], '([^-]+)') do
    table.insert(lastTime, tonumber(str))
end
local ms = lastTime[1] - firstTime[1]
if ms >= tonumber(ARGV[1]) then
    return redis.call('XRANGE', KEYS[1], '-', '+')
else
    return redis.error_reply('Only '..ms..' ms')
end

它返回:

(error) Stream is empty or key doesn`t exist

(error) Only 34 ms(如果我们没有经过所需的时间)

如果第一条消息和最后一条消息之间的所需时间已经过去，则为实际的条目列表。

确保检查 Introduction to Redis Streams以熟悉Redis Streams，并检查 EVAL command以了解Lua脚本。