如何将 node.js 中的管道传输到 redis？

Posted 2023-04-18

【中文标题】如何将 node.js 中的管道传输到 redis？

【英文标题】：How to pipeline in node.js to redis?

【发布时间】：2014-02-20 09:31:46

【问题描述】：

我有很多数据要插入（SET \ INCR）到redis DB，所以我正在通过node.js寻找pipeline \ mass insertion。

我在 node.js 中找不到任何好的示例/API，所以任何帮助都会很棒！

【参考方案1】：

是的，我必须承认缺乏这方面的示例，但我设法创建了流，我在该流上批量发送了几个插入命令。

你应该为 redis 流安装模块：

npm install redis-stream

这就是你使用流的方式：

var redis = require('redis-stream'),
    client = new redis(6379, '127.0.0.1');

// Open stream
var stream = client.stream();

// Example of setting 10000 records
for(var record = 0; record < 10000; record++) 

    // Command is an array of arguments:
    var command = ['set', 'key' + record, 'value'];  

    // Send command to stream, but parse it before
    stream.redis.write( redis.parse(command) );


// Create event when stream is closed
stream.on('close', function () 
    console.log('Completed!');

    // Here you can create stream for reading results or similar
);

// Close the stream after batch insert
stream.end();

此外，您可以根据需要创建多个流，并随时打开/关闭它们。

在redis-stream node moduleredis-stream node module的node.js中有几个使用redis流的例子

【讨论】：

谢谢托尼！你知道它是否以及如何使用 Lua 脚本？

嗯我没试过，但我认为你可以在 redis 实例中加载脚本并使用通过管道发送的eval 或evalsha 命令运行它们。

我逐字运行了您的代码，但没有设置任何键。之后通过 redis-cli 调用“keys *”会产生一个空集。

@Jake 你是对的。感谢您的报告。这已经过时了。我现在检查了 redis-stream 的新版本 并且命令发生了变化，好吧，这个例子中只有一个： 而不是：stream.write( ... ) 在新版本中是：stream.redis.write( ... )

【参考方案2】：

在 node_redis 中，所有命令都是流水线的：

https://github.com/mranney/node_redis/issues/539#issuecomment-32203325

【讨论】：

谢谢！我真的尝试过使用 multi 和 eval & exec 但性能更差......

【参考方案3】：

您可能还想查看batch()。使用multi() 会更慢的原因是因为它是事务性的。如果某事失败，则不会执行任何操作。这可能是您想要的，但您可以在这里选择速度。

redis-stream 包似乎没有使用 Redis 的批量插入功能，因此它也比 Redis 网站继续与 redis-cli 讨论的批量插入要慢。

另一个想法是使用 redis-cli 并给它一个文件来流式传输，这个 NPM 包就是这样做的：https://github.com/almeida/redis-mass

不想先写入磁盘上的文件？本次回购：https://github.com/eugeneiiim/node-redis-pipe/blob/master/example.js

...也流式传输到 Redis，但不写入文件。它流式传输到一个衍生的进程并每隔一段时间刷新一次缓冲区。

在 Redis 网站的 mass insert (http://redis.io/topics/mass-insert) 下，您可以看到一个 Ruby 小示例。上面的 repo 基本上将其移植到 Node.js，然后将其直接流式传输到生成的 redis-cli 进程。

所以在 Node.js 中，我们有：

var redisPipe = spawn('redis-cli', ['--pipe']);

spawn() 返回一个对子进程的引用，您可以使用stdin 传递到该子进程。例如：redisPipe.stdin.write()。

您可以继续写入缓冲区，将其流式传输到子进程，然后不时清除它。这不会填满它，因此在内存上可能会比 node_redis 包（在其文档中字面上说数据保存在内存中）要好一些，尽管我没有深入研究它所以我不知道内存占用最终是什么。它可能会做同样的事情。

当然请记住，如果出现问题，一切都会失败。这就是创建 fluentd 之类的工具的目的（这是另一种选择：http://www.fluentd.org/plugins/all - 它有几个 Redis 插件）......但同样，这意味着您在某种程度上支持磁盘上的数据。我个人也使用 Embulk 来执行此操作（这需要磁盘上的文件），但它不支持批量插入，因此速度很慢。 30,000 条记录耗时近 2 小时。

流式处理方法（不受磁盘支持）的一个好处是，如果您正在从另一个数据源执行大量插入操作。假设数据源返回大量数据并且您的服务器没有硬盘空间来支持所有数据 - 您可以改为流式传输。同样，您也有失败的风险。

我发现自己处于这个位置，因为我正在构建一个 Docker 映像，该映像将在没有足够磁盘空间来容纳大型数据集的服务器上运行。当然，如果您可以将所有内容都放在服务器的硬盘上，那就容易多了……但如果您不能，流式传输到 redis-cli 可能是您唯一的选择。

如果您真的定期推送大量数据，老实说，我可能会推荐 fluentd。它具有许多出色的功能，可确保您的数据到达目的地，如果出现故障，它可以恢复。

所有这些 Node.js 方法的一个问题是，如果某些东西失败了，你要么全部丢失，要么必须重新插入。

【参考方案4】：

默认情况下，node_redis，Node.js 库在管道中发送命令，并自动选择将多少命令进入每个管道 [(https://github.com/NodeRedis/node-redis/issues/539#issuecomment-32203325)][1]。因此，您无需担心这一点。但是，其他 Redis默认情况下，客户端可能不使用管道；您需要查看客户端文档以了解如何利用管道。