redis-queue怎么批量投递数据呢

小白更新于2023-05-12

比如一个表里有个3，5万个数据遍历投递很慢

foreach ($users as $data) {
        $queue = 'sms';
        Redis::send($queue, $data);
    }

这样很慢有啥解决方案吗

 918  3 5

3个回答

北月 2023-05-12

首先 workerman/redis-queue 队列组件是利用 Redis 的 List 和 Sorted Set 实现的，因此在客户端可以直接使用原生 Redis 的管道功能，文档地址：在非workerman环境向队列发送消息

自己单独封装一个函数去用即可：

function multi_redis_queue_send($redis, $queue, $multiData, $delay = 0) {
    $queue_waiting = '{redis-queue}-waiting'; //1.0.5版本之前为redis-queue-waiting
    $queue_delay = '{redis-queue}-delayed';//1.0.5版本之前为redis-queue-delayed
    // 创建管道
    $pipe = $redis->multi(Redis::PIPELINE);
    foreach ($multiData as $data) {
        $now = time();
        $package_str = json_encode([
            'id'       => rand(),
            'time'     => $now,
            'delay'    => 0,
            'attempts' => 0,
            'queue'    => $queue,
            'data'     => $data
        ]);
        if ($delay) {
            $pipe->zAdd($queue_delay, $now + $delay, $package_str);
        } else {
            $pipe->lPush($queue_waiting.$queue, $package_str);
        }
    }
    // 执行管道中的命令
    $results = $pipe->exec();
    // 需要注意的是，管道操作并不是原子操作，虽然管道中的所有命令一起执行，但是每个命令的执行结果仍然会返回给客户端。因此，在使用管道操作时，需要注意处理每个命令的执行结果。
    // 处理每个命令的执行结果
    foreach ($results as $result) {
        // ...
    }
}

在处理每个命令的执行结果时，需要注意以下几点：

如果命令执行成功，对应的执行结果是一个非负整数，表示执行成功的命令数量。
如果命令执行失败，对应的执行结果是一个RedisException异常对象，需要对异常进行处理。
在使用EXEC命令执行管道操作之前，不能执行其他Redis命令，否则会中断管道操作，导致管道中的命令没有被执行。

小白 2023-05-12

谢谢大佬耐心指点
小W 2023-05-12

mark

tanhongbin 2023-05-12

没错就是管道

小白 2023-05-12

谢谢大佬耐心指点

小白 2023-05-13

最后的解决方案分享一下还在测试呢，不知道还又什么问题，大佬们看下

function multi_redis_queue_send($queue, $multiData, $delay = 0) {

    Redis::pipeline(function ($pipe) use ($queue, $multiData, $delay) {
        $queue_waiting = '{redis-queue}-waiting'; //1.0.5版本之前为redis-queue-waiting
        $queue_delay = '{redis-queue}-delayed';//1.0.5版本之前为redis-queue-delayed
        foreach ($multiData as $data) {
            $now = time();
            $package_str = json_encode([
                'id'       => rand(),
                'time'     => $now,
                'delay'    => 0,
                'attempts' => 0,
                'queue'    => $queue,
                'data'     => $data
            ]);
            if ($delay) {
                $pipe->zAdd($queue_delay, $now + $delay, $package_str);
            } else {
                $pipe->lPush($queue_waiting.$queue, $package_str);
            }
        }
    });

}