服务端及时推送消息给客户端

发表于 | 分类于

场景：

  1. 在商户后台, 若有用户下单, 实时通知商户[Ajax 轮询, 并不是实时]
  2. 后台批量导入产品, 每成功导入一个, 就通知前台, 显示导入成功

实现原理：

  1. 服务端 建立一个 Websocket Worker, 用于维护客户端长连接
  2. 服务端 Websocket Worker 启动后, 内部建立一个 Text Worker, 由于 Websocket Worker 与 Text Worker 是同一个进程, 方便共享客户端连接
  3. 某个独立的后台系统, 通过 Text 协议与 Text Worker 通讯
  4. Text Worker 操作 Websocket Worker 完成数据推送

执行流程：

  1. 启动 start.php, 运行命令：php start.php start -d
  2. 浏览器打开 client.html, 支持开启多个, 等于绑定UID, 并接收服务端推送
  3. 运行 send.php, 点击下单, client.html 会收到服务端推送的消息

服务端启动文件 start.php

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
<?php

use Workerman\Worker;
require_once './Workerman-master/Autoloader.php';
 
// 初始化一个Websocket Worker容器,监听1234端口
$worker = new Worker('websocket://0.0.0.0:1234');
 
// 注意这里进程数必须设置为1,否则会报端口占用错误(php 7可以设置进程数大于1,前提是$inner_text_worker->reusePort = true)
$worker->count = 1;
// worker进程启动后创建一个Text Worker以便打开一个内部通讯端口
$worker->onWorkerStart = function($worker)
{
    // 开启一个内部端口,方便内部系统推送数据,Text协议格式 文本 + 换行符
    $innerTextWorker = new Worker('text://0.0.0.0:5678');
    $innerTextWorker->onMessage = function($connection, $buffer)
    {
        $data = json_decode($buffer, true);
        $ret = sendToAll($data['order_num']);
        $connection->send($ret ? 'success' : 'fail');
    };
    // 启用监听5678端口
    $innerTextWorker->listen();
};
 
// 新增加一个属性,用来保存uid到connection的映射
$worker->uidConnections = [];
// 当有客户端发来消息时执行的回调函数
$worker->onMessage = function($connection, $data)
{
    global $worker;
    // 判断当前客户端是否已经验证,即是否设置了uid
    if (!isset($connection->uid)) {
        // 没验证的话把第一个包当做uid
        $connection->uid = $data;
        // 保存uid到connection的映射,这样可以方便的通过uid查找connection,实现针对特定uid推送数据
        $worker->uidConnections[$connection->uid][] = $connection;
 
        return;
    }
};
 
// 当有客户端连接断开时
$worker->onClose = function($connection)
{
    global $worker;
    if (isset($connection->uid)) {
        // 连接断开时删除映射
        unset($worker->uidConnections[$connection->uid]);
    }
};
 
// 向所有验证的用户推送数据
function sendToAll($message)
{
    global $worker;
    foreach($worker->uidConnections as $connections) {
        foreach ($connections as $connection) {
            $connection->send($message);
        }
    }
 
    return true;
}
 
// 针对uid推送数据
function sendToUid($uid, $message)
{
    global $worker;
    if (isset($worker->uidConnections[$uid])) {
        $connection = $worker->uidConnections[$uid];
        $connection->send($message);
 
        return true;
    }
 
    return false;
}
 
// 运行所有的worker
Worker::runAll();

前端绑定UID并接收推送的JS, client.html

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title></title>
</head>
<body>
<script>
    var ws = new WebSocket('ws://192.168.50.82:1234');
    ws.onopen = function() {
        var uid = 1;
        ws.send(uid);
    };
    ws.onmessage = function(e) {
        document.write(e.data + '<br />');
    };
</script>
</body>
</html>

后端推送消息的代码, send.php

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
<?php
if ($_SERVER['REQUEST_METHOD'] === 'POST') {
    /**
     * 用户下单,实时通知后台
     */
    // 1,建立socket连接到内部推送端口
    $client = stream_socket_client('tcp://127.0.0.1:5678', $errno, $errmsg, 1);
    // 2,推送的数据
    $data = ['order_num' => mt_rand(1, 9999)];
    // 3,发送数据,注意5678端口是Text协议的端口,Text协议需要在数据末尾加上换行符
    fwrite($client, json_encode($data) . "\n");
    // 4,读取推送结果
    echo '订单编号：' . $data['order_num'] . ', 推送结果：' . fread($client, 8192);
}
?>
<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title></title>
</head>
<body>
<form action="send.php" method="post">
    <input type="submit" value="下单" />
</form>
</body>
</html>
0%