如何使用 workerman 同时执行任务？

孙飞 2021-12-10

现在需要通过接口在 es 查数据，数据量较大，直接使用 PHP 查询，再处理后存库，时间较长。如何使用 workerman 开多个进程同时执行呢？每一次执行根据查询数据的 offset 和 limit 可以拿到对应的数据，这样就可以加速查询。

 1514  2 0

2个回答

冷丶秋秋秋秋秋 2021-12-10

可以考虑用redis-queue队列,开多个进程查询.
https://www.workerman.net/doc/workerman/components/workerman-redis-queue.html

孙飞 2021-12-10

感谢回答，我现在主要是es中数据太大，如果只用php单进程来读，然后保存，效率就会很低，很低。所以就想用多进程的方式，每次通过不同的offset来查询es中的数据，这样可以加快从es中拿到数据。对列这个方法好像没法加快从es中拿数据这一过程。请问您知道怎么解决吗？不胜感激
冷丶秋秋秋秋秋 2021-12-10

可以考虑用workerman来进行异步处理.
参考这个https://www.workerman.net/q/582
冷丶秋秋秋秋秋 2021-12-10

https://www.workerman.net/doc/workerman/faq/async-task.html
孙飞 2021-12-11

大佬，我现在把这个异步调好了，但是我这个是定时任务，每一次启动都都是会监听这个端口，但是这样第二次定时任务消息就发在第一次的代码中去了，这种怎么处理呢
冷丶秋秋秋秋秋 2021-12-11

你把这个异步服务作为任务执行端，定时任务作为任务发布端。定时任务执行的时候把需要处理的任务发给执行端，让执行端慢慢执行，执行完了之后可以用回调的方式异步通知你
孙飞 2021-12-12

我昨天完成了这个实现，然后我异步执行的时候，数据太多，执行很耗时，我就想着使用 redis-queue 来做，但是数据放进 redis-queue，然后那边拿到的数据会重复，就很难受了
yongdao35 2021-12-12

是不是投送重复了，投递消息的时候记录日志，看下投递了几次。如果用了websocket推送，看下是不是建立了多个连接导致推送了多次
孙飞 2021-12-12

推送代码：
$text_worker = new Worker('text://0.0.0.0:56178');
$m = $data['talentId']%10;
$this->redis_queue_send($redis, 'User:'.$m , 1);
接收代码：总共有10个队列
$client->subscribe('User:0', function($data){
print_r($data);
});
...
$client->subscribe('User:9', function($data){
print_r($data);
});

是上边网消息队列中添加消息的时候有问题嘛
冷丶秋秋秋秋秋 2021-12-12

你投递的时候开了几个进程?

six 2021-12-10

开多个进程，每个进程按照进程id取不同的from,offset。例如

$worker = new Worker();
$worker->count = 10;
$worker->onWorkerStart = function($worker) {
    // 进程数为10的话，进程id会是从0到9
    $from = $woker->id * $worker->count;
    $offset = $worker->count;
    while($result = your_es_search($from, $offset)) {
        $from = $from + $worker->count*$offset;
    }
};

大致思路是这样，代码不一定对，你可以试下