服务器端（订阅）

swoole4.4.4版本demo

ini_set('error_reporting', E_ALL);
ini_set('display_errors', 'on');



class Server
{
    private $server;
    public function __construct()
    {
        $this->server=new Swoole\Server('0.0.0.0', 9501);
        $this->server->set([
            'worker_num' => 3, 'task_worker_num' => 3, 'daemonize' => false, 'max_request' => 10000, 'dispatch_mode' => 2,
        ]);
        $this->server->on("Start",[$this,'onStart']);
        $this->server->on("ManagerStart",[$this,'onManagerStart']);
        $this->server->on("WorkerStart",[$this,'onWorkerStart']);
        $this->server->on('Connect',[$this,'onConnect']);
        $this->server->on('Receive',[$this,'onReceive']);
        $this->server->on('Task', array($this, 'onTask'));
        $this->server->on('Finish', array($this, 'onFinish'));
        $this->server->on('Close',[$this,'onClose']);

        $this->server->start();
    }

    public function onStart(Swoole\Server $serv)
    {
        swoole_set_process_name('simple_route_master');
        echo "启动了".PHP_EOL;
        $scheduler = new Swoole\Coroutine\Scheduler;
        go(function () {
            $client = new Swoole\Coroutine\Redis();;
            $client->connect('192.168.138.138', 6379);
            if($client->subscribe(['test'])){
                go(function()use($client){
                    while ($msg = $client->recv()) {
                        list($type, $name, $message) = $msg;
                        var_dump($msg);
                        if ($type == 'subscribe') // 或psubscribe
                        {
                            // 频道订阅成功消息，订阅几个频道就有几条
                        } else if ($type == 'unsubscribe' && $message == 0) // 或punsubscribe
                        {
                            break; // 收到取消订阅消息，并且剩余订阅的频道数为0，不再接收，结束循环
                        }
                        else if ($type == 'message') // 若为psubscribe，此处为pmessage
                        {

                        }
                    }
                    echo "0".PHP_EOL;
                });

            }
            echo "00".PHP_EOL;
        });

        echo "000".PHP_EOL;

    }

    public function onManagerStart()
    {
        swoole_set_process_name('simple_route_manager');
    }

    public function onWorkerStart(Swoole\Server $serv,$workerId)
    {
        echo $workerId.PHP_EOL;
        swoole_set_process_name('simple_route_worker');
        if ($workerId == 0) {
            //必须声明在订阅前因为里面套了一个while循环 下面的代码就不执行了
            swoole_timer_tick(1000, function ($timer_id) use ($serv) {
                $connectionsNum = count($serv->connections);
//                var_dump($connectionsNum);
            });

        }

    }

    public function onConnect(Swoole\Server $serv, $fd, $from_id)
    {
        //        $serv->send( $fd, "Hello {$fd}!" );
        echo "client {$fd} Connect\n";

    }

    public function onReceive(Swoole\Server $serv,$fd, $from_id, $data){

    }

    public function onTask()
    {

    }

    public function onFinish()
    {

    }

    public function onClose()
    {
        echo "关闭了".PHP_EOL;
    }
}

$server=new Server();
echo 111111;

注：代码demo只能在swoole版本4.3以下的版本能生效4.3删除的异步模块

<?php
 $server = new swoole_websocket_server("0.0.0.0", 9533);
 
$server->on('workerStart', function ($server, $workerId) {
    if($workerId==0){
 
        $client = new swoole_redis;
        $client->on('message', function (swoole_redis $client, $result) use ($server) {
            if ($result[0] == 'message') {
                foreach($server->connections as $fd) {
                    $msg=['type'=>"push",'msg'=>$result[2],'fromUser'=>"系统",'toUser'=>0];
                    $server->push($fd, json_encode($msg));
                }
            }
        });
        $client->connect('127.0.0.1', 6379, function (swoole_redis $client, $result) {
            $client->subscribe('test');
        });
    }
 
});
 
$server->on('open', function ($server, $request) {
 
});
 
$server->on('message', function (swoole_websocket_server $server, $frame) {
    var_dump($frame);
    $server->push($frame->fd, "hello");
});
 
$server->on('close', function ($serv, $fd) {
 
});
 
$server->start();

服务器端(发布)

$redis = new Redis();
// 第一个参数为redis服务器的ip,第二个为端口
$res = $redis->connect('127.0.0.1', 6379);
// test为发布的频道名称,hello,world为发布的消息
$res = $redis->publish('test','hello,world');

前端socket链接

<?php  session_start(); $_SESSION['id']=1; ?>
<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>聊天室</title>
</head>
<body>
 
<input name="name"> 
<input type="button" value="连接" onclick="connect();"> 
<br/>
<input name="msg">
<input type="button" value="发言" onclick="send();">
<div id="show"></div>
 </body>
</html>
 
<script>
    var ws;
 
    /**
     * 连接
     */
    function connect() {
        if (ws) {
            return;
        }
 
        var name = document.getElementsByName("name")[0];
        if (!name.value) {
            alert('请输入名称！');
            name.focus();
            return;
        }
 
        ws = new WebSocket("ws://127.0.0.1:9533");
        ws.onopen = function () {
            //发送登录信息
            var data = '{"type":"login","name":"' + name.value + '"}';
            ws.send(data);
        };
        ws.onmessage = onmessage;
    }
 
    /**
     * 接收信息
     */
    function onmessage(e) {
        //var data = eval("(" + e.data + ")");
        var data = e.data;
        //显示
        var show = document.getElementById("show");
        show.innerHTML += data + "<br>";
    }
 
    /**
     * 发言
     */
    function send() {
        if (!ws) {
            alert('请先连接！');
            return;
        }
 
        var msg = document.getElementsByName("msg")[0];
        if (!msg.value) {
            alert('请输入内容！');
            msg.focus();
            return;
        }
 
        //发送
        var data = '{"type":"say","msg":"' + msg.value + '"}';
        ws.send(data);
        //清空
        msg.value = "";
    }
</script>