思路
利用 websocket 协议让客户端和服务器端保持有状态的长链接，保存链接上来的客户端 id。订阅发布者发布的消息针对已保存的客户端 id 进行广播消息。
websocket 服务
composer require hyperf/websocket-server
配置文件 [config/autoload/server.php]
<?php
return [
'mode' => swoole_process,
'servers' => [
[
'name' => 'http',
'type' => server::server_http,
'host' => '0.0.0.0',
'port' => 11111,
'sock_type' => swoole_sock_tcp,
'callbacks' => [
swooleevent::on_request => [hyperf\httpserver\server::class, 'onrequest'],
],
],
[
'name' => 'ws',
'type' => server::server_websocket,
'host' => '0.0.0.0',
'port' => 12222,
'sock_type' => swoole_sock_tcp,
'callbacks' => [
swooleevent::on_hand_shake => [hyperf\websocketserver\server::class, 'onhandshake'],
swooleevent::on_message => [hyperf\websocketserver\server::class, 'onmessage'],
swooleevent::on_close => [hyperf\websocketserver\server::class, 'onclose'],
],
],
],
websocket 服务器端代码示例
<?php
declare(strict_types=1);
/**
* this file is part of hyperf.
*
* @link https://www.hyperf.io
* @document https://doc.hyperf.io
* @contact group@hyperf.io
* @license https://github.com/hyperf-cloud/hyperf/blob/master/license
*/
namespace app\controller;
use hyperf\contract\oncloseinterface;
use hyperf\contract\onmessageinterface;
use hyperf\contract\onopeninterface;
use swoole\http\request;
use swoole\server;
use swoole\websocket\frame;
use swoole\websocket\server as websocketserver;
class websocketcontroller extends controller implements onmessageinterface, onopeninterface, oncloseinterface
{
/**
* 发送消息
* @param websocketserver $server
* @param frame $frame
*/
public function onmessage(websocketserver $server, frame $frame): void
{
//心跳刷新缓存
$redis = $this->container->get(\redis::class);
//获取所有的客户端id
$fdlist = $redis->smembers('websocket_sjd_1');
//如果当前客户端在客户端集合中,就刷新
if (in_array($frame->fd, $fdlist)) {
$redis->sadd('websocket_sjd_1', $frame->fd);
$redis->expire('websocket_sjd_1', 7200);
}
$server->push($frame->fd, 'recv: ' . $frame->data);
}
/**
* 客户端失去链接
* @param server $server
* @param int $fd
* @param int $reactorid
*/
public function onclose(server $server, int $fd, int $reactorid): void
{
//删掉客户端id
$redis = $this->container->get(\redis::class);
//移除集合中指定的value
$redis->srem('websocket_sjd_1', $fd);
var_dump('closed');
}
/**
* 客户端链接
* @param websocketserver $server
* @param request $request
*/
public function onopen(websocketserver $server, request $request): void
{
//保存客户端id
$redis = $this->container->get(\redis::class);
$res1 = $redis->sadd('websocket_sjd_1', $request->fd);
var_dump($res1);
$res = $redis->expire('websocket_sjd_1', 7200);
var_dump($res);
$server->push($request->fd, 'opened');
}
}
websocket 前端代码
function websockettest() {
if ("websocket" in window) {
console.log("您的浏览器支持 websocket!");
var num = 0
// 打开一个 web socket
var ws = new websocket("ws://127.0.0.1:12222");
ws.onopen = function () {
// web socket 已连接上，使用 send() 方法发送数据
//alert("数据发送中...");
//ws.send("发送数据");
};
window.setinterval(function () { //每隔5秒钟发送一次心跳，避免websocket连接因超时而自动断开
var ping = {"type": "ping"};
ws.send(json.stringify(ping));
}, 5000);
ws.onmessage = function (evt) {
var d = json.parse(evt.data);
console.log(d);
if (d.code == 300) {
$(".address").text(d.address)
}
if (d.code == 200) {
var v = d.data
console.log(v);
num++
var str = `<div class="item">
<p>${v.recordouttime}</p>
<p>${v.useroutname}</p>
<p>${v.useroutnum}</p>
<p>${v.dooroutname}</p>
</div>`
$(".tablehead").after(str)
if (num > 7) {
num--
$(".table .item:nth-last-child(1)").remove()
}
}
};
ws.error = function (e) {
console.log(e)
alert(e)
}
ws.onclose = function () {
// 关闭 websocket
alert("连接已关闭...");
};
} else {
alert("您的浏览器不支持 websocket!");
}
}
amqp 组件
composer require hyperf/amqp
配置文件 [config/autoload/amqp.php]
<?php
return [
'default' => [
'host' => 'localhost',
'port' => 5672,
'user' => 'guest',
'password' => 'guest',
'vhost' => '/',
'pool' => [
'min_connections' => 1,
'max_connections' => 10,
'connect_timeout' => 10.0,
'wait_timeout' => 3.0,
'heartbeat' => -1,
],
'params' => [
'insist' => false,
'login_method' => 'amqplain',
'login_response' => null,
'locale' => 'en_us',
'connection_timeout' => 3.0,
'read_write_timeout' => 6.0,
'context' => null,
'keepalive' => false,
'heartbeat' => 3,
],
],
];
mq 消费者代码
<?php
declare(strict_types=1);
namespace app\amqp\consumer;
use hyperf\amqp\annotation\consumer;
use hyperf\amqp\message\consumermessage;
use hyperf\amqp\result;
use hyperf\server\server;
use hyperf\server\serverfactory;
/**
* @consumer(exchange="hyperf", routingkey="hyperf", queue="hyperf", nums=1)
*/
class democonsumer extends consumermessage
{
/**
* rabbmitmq消费端代码
* @param $data
* @return string
*/
public function consume($data): string
{
print_r($data);
//获取集合中所有的value
$redis = $this->container->get(\redis::class);
$fdlist=$redis->smembers('websocket_sjd_1');
$server=$this->container->get(serverfactory::class)->getserver()->getserver();
foreach($fdlist as $key=>$v){
if(!empty($v)){
$server->push((int)$v, $data);
}
}
return result::ack;
}
}
控制器代码
/**
* test
* @return array
*/
public function test()
{
$data = array(
'code' => 200,
'data' => [
'useroutname' => 'ccflow',
'useroutnum' => '9999',
'recordouttime' => date("y-m-d h:i:s", time()),
'dooroutname' => '教师公寓',
]
);
$data = \guzzlehttp\json_encode($data);
$message = new demoproducer($data);
$producer = applicationcontext::getcontainer()->get(producer::class);
$result = $producer->produce($message);
var_dump($result);
$user = $this->request->input('user', 'hyperf');
$method = $this->request->getmethod();
return [
'method' => $method,
'message' => "{$user}.",
];
}
最终效果
黄山市民网：https://www.huangshanshimin.com/
