发表新帖

new process从redis订阅消息特殊情况无法正常广播

### 问题描述

当我打开多个浏览器页面 比如 5个，建立了5个链接，这时候往redis发布数据，是能正常广播的，但是如果我关闭一个浏览器页面，有一个
client-3 is closed 以后，再去redis 发布数据，所有其他页面都收不到广播了

第47行 var_dump(count($ws->connections) . “ conns”);
这地方还能正常打印4 conns
但是 48行的foreach 里面的2个分支 都没有打印东西，不知道怎么回事了。

### Swoole版本，PHP版本，以及操作系统版本信息

swoole 4.5.7
php7.2
centos7

### 相关代码
```php
/**
 * 第一版本正常运行的
 */
$ws = new Swoole\WebSocket\Server('0.0.0.0', 9501);

//Co::set(['hook_flags'=> SWOOLE_HOOK_ALL]);

Swoole\Runtime::enableCoroutine(SWOOLE_HOOK_ALL | SWOOLE_HOOK_CURL);

$ws->set(array(
    'debug_mode' => 1,
    'worker_num' => 1,
    'max_request' => 10000,
    'daemonize' => 1, //以守护进程执行
    'dispatch_mode' => 2,
    'task_worker_num' => 2, //task进程的数量
    'max_wait_time' => 60,
    'reload_async' => true,
    'heartbeat_check_interval' => 60,
    'heartbeat_idle_time' => 120,
    'pid_file' => '/data/nginx/logs/swoole.pid',
    "log_file" => "/data/nginx/logs/swoole.log",//日志
//    'ssl_cert_file' => __DIR__ . '/../config/fullchain.pem',
//    'ssl_key_file' => __DIR__ . '/../config/privkey.pem'
));

$newprocess = new Swoole\Process(function ($process) use ($ws) {
    Co\run(function () use ($ws) {
        $redis = new Swoole\Coroutine\Redis();
        $redis->connect('127.0.0.1', 6379, true);
        $redis->auth("123456");
        if ($redis->subscribe(['msg'])) // 或者使用psubscribe
        {
            while ($msg = $redis->recv()) {
                // msg是一个数组, 包含以下信息
                // $type # 返回值的类型：显示订阅成功
                // $name # 订阅的频道名字 或 来源频道名字
                // $info  # 目前已订阅的频道数量 或 信息内容
                list($type, $name, $info) = $msg;
                if ($type == 'subscribe') { // 或psubscribe
                    // 频道订阅成功消息，订阅几个频道就有几条
                } else if ($type == 'unsubscribe' && $info == 0) { // 或punsubscribe
                    break; // 收到取消订阅消息，并且剩余订阅的频道数为0，不再接收，结束循环
                } else if ($type == 'message') {  // 若为psubscribe，此处为pmessage
//                    var_dump($name); // 打印来源频道名字
                    var_dump($info); // 打印消息
                    var_dump(count($ws->connections) . " conns");
                    foreach ($ws->connections as $fd) {
                        var_dump($fd . " fd");
                        // 需要先判断是否是正确的websocket连接，否则有可能会push失败
                        if ($ws->isEstablished($fd)) {
                            //向 WebSocket 客户端连接推送数据，长度最大不得超过 2M。
                            $ws->push($fd, "hello, welcome $info\n");
                        } else {
                            var_dump($fd . " fd error");
                        }
                    }
                    // balabalaba.... // 处理消息
//                    if ($need_unsubscribe) { // 某个情况下需要退订
//                        $redis->unsubscribe(); // 继续recv等待退订完成
//                    }
                }
            }
        }
        Co::sleep(1);
    });
}, false, 2, 0);
$ws->addProcess($newprocess);

//监听WebSocket连接打开事件
$ws->on('open', function ($ws, $request) {
//    $ws->connections[] = $request->fd;
    var_dump($request->fd);
    $ws->push($request->fd, "open\n");
});

//监听WebSocket消息事件
$ws->on('message', function ($ws, $frame) {
    echo "Message: {$frame->data}\n";
    $ws->push($frame->fd, "server: {$frame->data}");
});

//监听WebSocket连接关闭事件
$ws->on('close', function ($ws, $fd) {
    echo "client-{$fd} is closed\n";
});

//workstart事件
$ws->on('workerstart', function ($ws, $work_id) {

});

//server task
$ws->on('task', function ($ws, $taskid, $from_id, $data) {
    global $table;
    foreach ($table as $row) {
        $ws->send($row['fd'], "{$row['fd']} i am broadcast");
    }

$ws->finish("$data -> OK");
});

//处理异步任务的结果
$ws->on('finish', function ($ws, $task_id, $data) {
    echo "AsyncTask[$task_id] Finish: $data" . PHP_EOL;
});

$ws->start();

```

### 你期待的结果是什么？实际看到的错误信息又是什么？

希望一个客户端关闭，其他不受影响

发布于4年前 · 1 次浏览 · 来自提问

phpmianshi