如何基于Hyperf实现RabbitMQ+WebSocket消息推送
介绍
基于 Hyperf+ WebSocket +RabbitMQ 实现的一个简单大屏幕的消息推送。
思路
利用 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}.",
];
}最终效果
推荐：《PHP教程》
以上是如何基于Hyperf实现RabbitMQ+WebSocket消息推送的详细内容。更多信息请关注PHP中文网其他相关文章！
热AI工具
Undresser.AI Undress
人工智能驱动的应用程序，用于创建逼真的裸体照片
AI Clothes Remover
用于从照片中去除衣服的在线人工智能工具。
Undress AI Tool
免费脱衣服图片
Clothoff.io
AI脱衣机
AI Hentai Generator
免费生成ai无尽的。
热门文章
热工具
记事本++7.3.1
好用且免费的代码编辑器
SublimeText3汉化版
中文版，非常好用
禅工作室 13.0.1
功能强大的PHP集成开发环境
Dreamweaver CS6
视觉化网页开发工具
SublimeText3 Mac版
神级代码编辑软件(SublimeText3)
热门话题
如何使用Hyperf框架进行配置管理
Oct 28, 2023 am 10:07 AM
Hyperf是一个优秀的PHP框架，其主要的特点是快速、灵活、可扩展，目前在业界有着广泛的应用。在使用Hyperf框架进行开发的过程中，我们常常会遇到需要配置管理的情况。本文将介绍如何使用Hyperf框架进行配置管理，并且提供具体的代码示例。一、配置文件的位置在使用Hyperf框架进行开发的时候，配置文件通常会放在config目录下，也可以在.env文件中进
如何使用Hyperf框架进行文件下载
Oct 21, 2023 am 08:23 AM
如何使用Hyperf框架进行文件下载引言：在使用Hyperf框架开发Web应用程序时，文件下载是一个常见的需求。本文将介绍如何使用Hyperf框架进行文件下载，包括具体的代码示例。一、准备工作在开始之前，确保你已经安装好了Hyperf框架并成功创建了一个Hyperf应用程序。二、创建文件下载控制器首先，我们需要创建一个控制器来处理文件下载的请求。打开终端，进
PHP Hyperf微服务开发指南：从入门到精通
Sep 12, 2023 am 10:31 AM
从2004年诞生至今，PHP一直是全球最流行的开发语言之一。随着互联网的快速发展和技术的不断创新，PHP的发展也日新月异。其中，微服务架构逐渐成为当今软件开发的热门趋势。本文将带你进入PHPHyperf微服务开发的世界，从入门到精通。一、什么是微服务架构？微服务架构是一种基于一组小型、独立部署的服务组件构建的系统架构。相比于传统的单体应用架构，微服务架构通
如何使用Hyperf框架进行请求限流
Oct 20, 2023 pm 01:58 PM
如何使用Hyperf框架进行请求限流引言：在现代互联网应用中，如何在高并发的情况下保证系统的稳定性是非常重要的。请求限流是常见的应对策略之一。本文将介绍如何使用Hyperf框架进行请求限流，并给出具体的代码示例。一、什么是请求限流请求限流是指在一段时间内限制系统的请求访问量，避免系统因为过多的请求而崩溃。通过合理的限流策略，可以提供更好的服务质量和稳定性。H
如何使用Hyperf框架进行数据分页
Oct 20, 2023 am 11:25 AM
如何使用Hyperf框架进行数据分页引言：数据分页在实际的Web开发中非常常见，通过分页可以让用户浏览大量数据时更加便捷。Hyperf是一个高性能的PHP框架，提供了一系列强大的特性和组件。本文将介绍如何使用Hyperf框架进行数据分页，并给出详细的代码示例。一、准备工作：在开始之前，需要确保已经正确安装和配置了Hyperf框架。可以通过Composer进行
如何使用Hyperf框架进行图片处理
Oct 24, 2023 pm 12:04 PM
如何使用Hyperf框架进行图片处理引言：随着移动互联网的快速发展，图片处理在现代Web开发中变得愈发重要。Hyperf是一款基于Swoole的高性能框架，它提供了丰富的组件和功能，包括图片处理。本文将介绍如何使用Hyperf框架进行图片处理，并提供具体的代码示例。一、安装Hyperf框架：在开始之前，我们先确保已经安装了Hyperf框架。可以通过Compo
如何使用Hyperf框架进行缓存管理
Oct 21, 2023 am 08:36 AM
如何使用Hyperf框架进行缓存管理缓存是提高应用性能的重要手段之一，而现代框架为我们提供了更加便捷的缓存管理工具。本文将介绍如何使用Hyperf框架进行缓存管理，并提供具体的代码示例。Hyperf框架是一款基于Swoole拓展开发的高性能框架，内置了丰富的组件和工具，其中包括强大的缓存管理功能。Hyperf框架支持多种缓存驱动，如Redis、Memcach
构建可扩展的微服务应用：探索PHP Hyperf的技术特点
Sep 11, 2023 pm 07:01 PM
近年来，微服务架构已成为构建现代应用程序的主流方式。它通过将一个大型应用拆分成小而自治的服务，从而提高了应用的可扩展性、可维护性和可部署性。在微服务架构中，每个服务都是独立开发、部署和运行的，它们之间通过轻量级的通信机制进行交互。在构建微服务应用时，选择一个适合的开发框架非常关键。PHPHyperf是一个基于Swoole高性能协程网络框架的微服务框架
