hyperf 并发问题，websocket延迟发送消息怎么使最后一条不等待很长时间

Question

hyperf 并发问题，我有一个列表内含100条数据，需要循环内部每一条数据，并延迟相应时间发送给对应的websocket客户端，怎么才能让最后一条不会因为延迟时间，而等待很久

Answer 1

引自免费微信小程序：皆我百晓生

为了解决Hyperf框架下WebSocket延迟发送消息时，最后一条消息不会因为累积的延迟时间而等待很久的问题，我们可以采用异步任务队列的方式来处理。这样，每条消息都会被独立地推送到队列中，由队列中的工作进程分别处理，从而避免了串行处理造成的延迟累积。

解决方案步骤：

1. 使用Hyperf的异步任务队列：Hyperf内置了基于Swoole的异步任务队列机制，可以将每个延迟发送的任务独立地推入队列中，由队列中的工作进程异步处理。

2. 定义任务类：创建一个任务类，用于封装需要发送的消息和其延迟时间。这个类应该继承自Hyperf\Contract\StdoutLoggerInterface或者实现相应的接口，以便于在队列中执行。

3. 发送任务到队列：遍历你的100条数据，对于每条数据，创建对应的任务实例并发送到队列中。

4. 配置队列工作进程：确保你的项目中已经正确配置了队列工作进程，以处理这些异步任务。

示例代码：

首先，你需要定义一个任务类，例如SendDelayedMessageTask：

<?php

declare(strict_types=1);

namespace App\Task;

use Hyperf\Contract\StdoutLoggerInterface;
use Hyperf\Utils\Context;
use Psr\Container\ContainerInterface;
use Psr\Log\LoggerInterface;

class SendDelayedMessageTask implements StdoutLoggerInterface
{
    protected $data;
    protected $delay;

    public function __construct($data, int $delay)
    {
        $this->data = $data;
        $this->delay = $delay;
    }

    public function handle(ContainerInterface $container)
    {
        /** @var LoggerInterface $logger */
        $logger = $container->get(LoggerInterface::class);
        $logger->info("Handling task for data: " . json_encode($this->data));

        // 使用sleep模拟延迟，实际应用中可以使用更精确的定时器
        sleep($this->delay);

        // 发送消息到WebSocket客户端
        // 这里使用$container获取WebSocket服务并发送消息
        $wsService = $container->get(\App\Service\WebSocketService::class);
        $wsService->sendMessage($this->data);
    }
}

然后，在你的业务逻辑中，遍历数据并发送任务到队列：

<?php

declare(strict_types=1);

use App\Task\SendDelayedMessageTask;
use Hyperf\Contract\StdoutLoggerInterface;
use Hyperf\Utils\Context;

class YourBusinessLogic
{
    public function sendMessages(array $messages): void
    {
        foreach ($messages as $index => $message) {
            $delay = $index; // 假设每个消息的延迟时间为它的索引值
            $task = new SendDelayedMessageTask($message, $delay);
            Context::getInstance()->get('queue')->push($task);
        }
    }
}

最后，确保你的main.php或相应的配置文件中开启了队列工作进程：

<?php

return [
    'mode' => SWOOLE_PROCESS,
    'settings' => [
        'worker_num' => swoole_cpu_num(), // 根据CPU核心数调整工作进程数量
    ],
];

这样，每条消息都会被独立地处理，最后一条消息也不会因为前面消息的延迟而等待很久。