node.js - Redis 流 : How to manage perpetual subscription and BLOCK behaviour?

Question

我在 Express 应用程序中使用 Redis。我的应用程序既是流的发布者又是流的消费者，使用单个 redis 连接 (redis.createClient)。我对管理永久订阅(使用 xreadgroup)的最佳方式有疑问。目前我正在这样做:

    const readStream = () => xreadgroup('GROUP' appId, consumerId, 'BLOCK', 1, 'COUNT', 1, 'STREAMS' key, '>')
        .then(handleData)
        .then(() => setImmeadiate(readStream));

xreadgroup 只是 node-redis 的 xreadgroup 的 promisified 版本。

我的问题 - BLOCK 的正确用法是什么？如果我无限期地或长时间地阻止，那么我的客户端将无法发布任何消息(使用 xadd)，直到它被解除阻止，或者阻止超时。因为我必须使用某种循环/递归来继续读取事件，所以 BLOCK 似乎是完全没有必要的；我可以把它关掉吗？这是预期的用法吗？

同样，使用 setImmeadiate 是否合适，还是首选 process.nextTick 或异步循环？

node-redis 中的文档非常少，少数 examples阻塞后只读一次消息，不要在同一个客户端上生产/消费。

Answer 1

我不是这方面的专家，但我想分享一些可能有帮助的想法。

1. 我不确定 node-redis 是否可以“堆叠”多个命令，这意味着 - 它是否能够在等待 XREADGROUP 完成时触发新命令？

2. 从您的描述来看，情况似乎就是这样。在这种情况下，我建议您创建一个专用连接来调用 XREADGROUP - 这样您就可以在不阻塞彼此的情况下发布和收听。

3. 您不需要使用BLOCK；但是如果您的目标是监听所有事件并等待那些尚未发布的事件，那么使用它可能是明智的，并且会在减少对 redis 的调用的同时为您提供更好的性能。

4. setImmediate 可能不错，尤其是使用 BLOCK 时。如果您不使用它，那么在调用之间添加一点超时可能会很好 - 没有 BLOCK 调用几乎会立即回答返回。你可以查看this了解更多详情。

友情提醒:不要忘记确认您的消息或使用 NOACK(根据您的用例可能没问题):

consumer groups require explicit acknowledgment of the messages successfully processed by the consumer, via the XACK command.

The NOACK subcommand can be used to avoid adding the message to the PEL in cases where reliability is not a requirement and the occasional message loss is acceptable.

来源:https://redis.io/commands/xreadgroup