当前位置: 首页 > news >正文

php:使用Ratchet类实现分布式websocket服务

一、前言

        最近需要做一个有关聊天的小程序,逻辑很简单,所以不打算用Swoole和workerman之类的,最后选择了Ratchet,因为简单易用,适合小型websocket服务。

二、问题

        但是目前我的项目是分布式环境,统一通过Nginx的反向代理分配到多个不同服务器,那么在其中一个服务器建立了WebSocket连接的用户如何给在另外一个服务器上建立了WebSocket连接的用户发送消息呢?这就涉及到了分布式websocket服务,但是我不希望太复杂,所以采用了消息队列的方式实现效果。

三、安装Ratchet类

直接使用composer安装,我用版本是"cboden/ratchet": "^0.2.8",比较老了。

composer require cboden/ratchet

四、创建websocket服务

因为php是同步阻塞型语言,通常每次请求都会从头到尾执行完成,在PHP中,所有的代码都按照顺序执行,直到脚本结束为止。但是我们要在一个进程中启动websocket服务还要监听消息队列,就需要用到ratchet中的事件循环机制,实现异步非阻塞通信效果。

<?php
//载入Ratchet类库
require_once APP_PATH.'vendor/autoload.php';
use Ratchet\Server\IoServer;
use Ratchet\WebSocket\WsServer;
use React\EventLoop\Factory as LoopFactory;
use React\Socket\Server as ReactSocket;set_time_limit(0);
ini_set('default_socket_timeout', -1);
/*** Websocket_Server*/
class ControllerWebsocket_Server
{public function indexAction(){try {$port = 8083;// 创建事件循环(使用该机制实现异步非阻塞通信)$loop = LoopFactory::create();// 创建 React Socket 服务器$socket = new ReactSocket($loop);$socket->listen($port, '0.0.0.0'); // 指定监听的端口和地址// 启动 WebSocket 服务器$server = new IoServer(new WsServer(new \ModelWebsocket_Handler($loop)),$socket,$loop);// 启动事件循环$loop->run();} catch (\Exception $e) {echo $e->getMessage();}}
}

其中ModelWebsocket_Handler是封装好的websocket操作类

<?php
//载入Ratchet类库
require_once APP_PATH.'vendor/autoload.php';
use Ratchet\MessageComponentInterface;
use Ratchet\ConnectionInterface;/*** websocket服务端-相关操作*/
class ModelWebsocket_Handler implements MessageComponentInterface {//数据缓存const REDIS_KEY_RESOURCE_DATA_MAP = 'h:websocket:resource:data:map';//客户端public $clients;public function __construct($loop) {$this->clients = new \SplObjectStorage();$this->subscribeMessage($loop);}/*** 连接建立时的逻辑*/public function onOpen(ConnectionInterface $conn) {$this->clients->attach($conn);echo "New connection! ({$conn->resourceId})\n";//获取连接请求的参数$params = [];$queryString = $conn->WebSocket->request->getQuery();parse_str($queryString, $params);//存储资源id相关数据$this->setResourceDataMap($conn->resourceId, $params);}/*** 收到消息时的逻辑*/public function onMessage(ConnectionInterface $from, $msg) {echo "Received message: {$msg}\n";foreach ($this->clients as $client) {if ($client === $from) {continue;}//发送消息$client->send($msg);}}/*** 连接关闭时的逻辑*/public function onClose(ConnectionInterface $conn) {$this->delResourceDataMap($conn->resourceId);$this->clients->detach($conn);echo "Connection {$conn->resourceId} has disconnected\n";}/*** 错误处理逻辑*/public function onError(ConnectionInterface $conn, \Exception $e) {echo "An error occurred: {$e->getMessage()}\n";$this->delResourceDataMap($conn->resourceId);$conn->close();}/*** 存储资源id相关数据* * @param  string  $resourceId* @param  array   $data* @return bool*/public function setResourceDataMap($resourceId, $data) {$redis = Comm_Redis::init(Comm_Redis::REDIS_TVDB, true);$rs = $redis->hSet(self::REDIS_KEY_RESOURCE_DATA_MAP, $resourceId, json_encode($data));return $rs;}/*** 获取资源id相关数据* * @param  string  $resourceId* @return array*/public function getResourceDataMap($resourceId) {$redis = Comm_Redis::init(true);$rs = $redis->hGet(self::REDIS_KEY_RESOURCE_DATA_MAP, $resourceId);return json_decode($rs, true) ?: [];}/*** 删除资源id相关数据* * @param  string  $resourceId* @return bool*/public function delResourceDataMap($resourceId) {$redis = Comm_Redis::init(true);$rs = $redis->hDel(self::REDIS_KEY_RESOURCE_DATA_MAP, $resourceId);return $rs;}/*** 订阅消息*/public function subscribeMessage($loop){$loop->addPeriodicTimer(1, function () {//在这里可以使用redis订阅消息、也可以使用kafka消费消息,然后再比对自身是否存在相应用户的连接,如果存在则发送,不存在则过滤,达到分布式webSocket服务的作用foreach ($this->clients as $client) {$client->send("测试");} });}
}

其中:subscribeMessage方法监听消息队列,收到消息之后比对自身是否存在相应用户的连接,如果存在则发送,不存在则过滤,达到分布式webSocket服务的作用。

当然如果你能直接找到用户所连接的服务器,并且可以直接推给相应的服务器,那更好,可以节省流量开销和一些额外的逻辑处理。

相关文章:

php:使用Ratchet类实现分布式websocket服务

一、前言 最近需要做一个有关聊天的小程序&#xff0c;逻辑很简单&#xff0c;所以不打算用Swoole和workerman之类的&#xff0c;最后选择了Ratchet&#xff0c;因为简单易用&#xff0c;适合小型websocket服务。 二、问题 但是目前我的项目是分布式环境&#xff0c;统一通过Ng…...

储能场站安全风险挑战

电化学储能目前最大的痛点问题就是安全问题&#xff0c;制约了储能行业的发展。 首先&#xff1a;锂作为最活泼的金属加上有机溶剂的电解液&#xff0c;安全性天生就差。基因不行。 其次储能系统的BMS对电池管理相对粗放&#xff0c;不足以保证锂电池的安全运行。 当前储能产业…...

Ubuntu系统为同一逻辑网口配置不同网段的IP

近期遇到一个问题&#xff1a;机载计算机的载版上有两个网口&#xff0c;但是这两个网口本质上是一个独立网口一个交换机&#xff0c;即对于机载计算机而言这两个物理网口是同一个逻辑网口。但是我需要将这两个网口分别连接到两个设备&#xff0c;并配置不同网段的IP&#xff0…...

MySQL出现Waiting for table metadata lock的原因以及解决方法(已亲测)

参考&#xff1a;MySQL出现Waiting for table metadata lock的原因以及解决方法 - digdeep - 博客园 当对表执行truncate\drop 操作时&#xff0c;会出现一直处于等待的状态&#xff0c;通过show processlist可以看到TableA停滞在Waiting for table metadata lock的状态。kill…...

学会Lambda,让程序Pythonic一点

Lambda是Python里的高阶用法&#xff0c;要把代码写得Pythonic&#xff0c;就需要了解这些高阶用法&#xff0c;想说自己是一名真正的Python程序员&#xff0c;先要把代码写得Pythonic。 今天聊下Lambda的用法&#xff0c;写篇简短的用法说明。 Lambda是匿名函数的意思&#…...

GDPU 信息安全 期末复习

文章目录 第一章 绪论✅ 单选题✅ 简答题6. 假定你是单位的安全主管&#xff0c;为了提高单位的网络安全性&#xff0c;在制定单位的安全保障方案时&#xff0c;有哪些措施&#xff08;包括技术和非技术的&#xff09;&#xff1f;9. 有人说只要我有足够多的钱&#xff0c;就可…...

Python 使用 Token 认证方案连接 Kubernetes (k8s) 的详细过程

在 Kubernetes 中&#xff0c;使用 Token 认证是一种常见的客户端身份验证方式&#xff0c;尤其适用于 ServiceAccount。以下是详细的步骤&#xff0c;包括如何查看 Token、获取 API 服务地址、配置远程连接&#xff0c;以及如何在 Python 中连接 k8s。 1. 获取 Token 首先&a…...

【C++】ReadFile概述,及实践使用时ReadFile的速率影响研究

ReadFile 函数概述 ReadFile 是 Windows API 函数&#xff0c;用于从文件或设备&#xff08;如串口、硬盘等&#xff09;中读取数据。它是同步和异步 I/O 操作的基础函数。 函数原型 BOOL ReadFile(_In_ HANDLE hFile, // 文件或设备句柄_Out_write…...

Mysql的UPDATE(更新数据)详解

MySQL的UPDATE语句是用于修改数据库表中已存在的记录&#xff0c;本文将详细介绍UPDATE语句的基本语法、高级用法、性能优化策略以及注意事项&#xff0c;帮助您更好地理解和应用这一重要的SQL命令。 1. 基本语法 单表更新 单表更新的基本语法如下&#xff1a; UPDATE [LOW…...

基于Java Springboot高校奖助学金系统

一、作品包含 源码数据库设计文档万字PPT全套环境和工具资源部署教程 二、项目技术 前端技术&#xff1a;Html、Css、Js、Vue、Element-ui 数据库&#xff1a;MySQL 后端技术&#xff1a;Java、Spring Boot、MyBatis 三、运行环境 开发工具&#xff1a;IDEA/eclipse 数据…...

如何在 Ubuntu 22.04 上安装带有 Nginx 的 ELK Stack

今天我们来聊聊如何在 Ubuntu 22.04 服务器上安装 ELK Stack&#xff0c;并集成 Nginx 作为 Web 服务器&#xff0c;同时使用 Let’s Encrypt Certbot 进行 SSL 认证。ELK Stack&#xff0c;包括 Elasticsearch、Logstash 和 Kibana&#xff0c;是一套强大的工具&#xff0c;用…...

Python爬虫:深入探索1688关键词接口获取之道

在数字化经济的浪潮中&#xff0c;数据的价值愈发凸显&#xff0c;尤其是在电商领域。对于电商平台而言&#xff0c;关键词不仅是搜索流量的入口&#xff0c;也是洞察市场趋势、优化营销策略的重要工具。1688作为中国领先的B2B电商平台&#xff0c;其关键词接口的获取对于商家来…...

Let‘s Encrypt SSL证书:acmessl.cn申请免费3个月证书

目录 一、CA机构 二、Lets Encrypt特点 三、申请SSL 一、CA机构 ‌Lets Encrypt‌是一个由非营利组织Internet Security Research Group (ISRG)运营的证书颁发机构&#xff08;CA&#xff09;&#xff0c;旨在通过自动化和开放的方式为全球网站提供免费、可靠的SSL/TLS证书。…...

JSON Web Token (JWT)的简单介绍、验证过程及令牌刷新思路

目录 一、JWT 1、什么是Jwt 2、为什么要使用Jwt 3、应用场景 4.Jwt的组成 4.1、Header 4.2、Payload 4.3、signature 二、Jwt验证过程 1、生成Jwt令牌 2、解析旧的Jwt 3、复制Jwt 4、Jwt有效时间测试 三、Jwt令牌刷新思路 1、配置JwtFilter过滤器 2、登录生成Jwt令…...

xxl-job入门

xxl-job , 定时任务 分布式 &#xff0c; 带来的问题的 解决方案 像之前 很多项目都用到定时任务&#xff0c; 但是如果要改为 分布式&#xff0c; 那么定时任务 就要用到 xxl-job 1.用户画像 拼多多&#xff0c;看了某个东西后&#xff0c;推荐类似东西&#xff0c; 做埋…...

100.【C语言】数据结构之二叉树的堆实现(顺序结构) 1

目录 1.顺序结构 2.示意图 ​编辑 从物理结构还原为逻辑结构的方法 3.父子节点编号的规律 4.顺序存储的前提条件 5.堆的简介 堆的定义 堆的两个重要性质 小根堆和大根堆 6.堆的插入 7.堆的实现及操作堆的函数 堆的结构体定义 堆初始化函数HeapInit 堆插入元素函…...

大模型 VS 大语言模型

最近很多朋友搞不懂大模型和大预言模型的区别&#xff0c;总是把大模型就认为是大语言模型。 今天就用这篇帖子做一个科普。 大模型 概念&#xff1a;大模型是指拥有超大规模参数&#xff08;通常在十亿个以上&#xff09;、复杂计算结构的机器学习模型。它通常能够处理海量数…...

Linux高阶——1117—TCP客户端服务端

目录 1、sock.h socket常用函数 网络初始化函数 首次响应函数 测试IO处理函数 获取时间函数 总代码 2、sock.c SOCKET() ACCEPT()——服务端使用这个函数等待客户端连接 CONNECT()——客户端使用这个函数连接服务端 BIND()——一般只有服务端使用 LISTEN()——服务端…...

【Qt】Qt 在main.cpp中使用tr()函数报错

1. 问题 Qt 在main.cpp中使用tr()报错。 error: tr was not declared in this scope2. 解决方法 main.cpp中注意如下&#xff1a; //添加头文件 #include <QObject>//添加QObject QObject::tr("Hello")3. 参考 Qt tr()函数不起效的小问题...

面向对象高级(5)接口

面向对象高级&#xff08;5&#xff09; 接口 接口就是规范&#xff0c;定义的是一组规则&#xff0c;体现了现实世界中“如果是...则必须能...”的思想。继承是一个"是不是"的is-a关系&#xff0c;而接口实现则是 "能不能"的has-a关系。 1、接口的定义格…...

Python爬虫实战:研究MechanicalSoup库相关技术

一、MechanicalSoup 库概述 1.1 库简介 MechanicalSoup 是一个 Python 库,专为自动化交互网站而设计。它结合了 requests 的 HTTP 请求能力和 BeautifulSoup 的 HTML 解析能力,提供了直观的 API,让我们可以像人类用户一样浏览网页、填写表单和提交请求。 1.2 主要功能特点…...

python打卡day49

知识点回顾&#xff1a; 通道注意力模块复习空间注意力模块CBAM的定义 作业&#xff1a;尝试对今天的模型检查参数数目&#xff0c;并用tensorboard查看训练过程 import torch import torch.nn as nn# 定义通道注意力 class ChannelAttention(nn.Module):def __init__(self,…...

3.3.1_1 检错编码(奇偶校验码)

从这节课开始&#xff0c;我们会探讨数据链路层的差错控制功能&#xff0c;差错控制功能的主要目标是要发现并且解决一个帧内部的位错误&#xff0c;我们需要使用特殊的编码技术去发现帧内部的位错误&#xff0c;当我们发现位错误之后&#xff0c;通常来说有两种解决方案。第一…...

Python爬虫(一):爬虫伪装

一、网站防爬机制概述 在当今互联网环境中&#xff0c;具有一定规模或盈利性质的网站几乎都实施了各种防爬措施。这些措施主要分为两大类&#xff1a; 身份验证机制&#xff1a;直接将未经授权的爬虫阻挡在外反爬技术体系&#xff1a;通过各种技术手段增加爬虫获取数据的难度…...

从零实现STL哈希容器:unordered_map/unordered_set封装详解

本篇文章是对C学习的STL哈希容器自主实现部分的学习分享 希望也能为你带来些帮助~ 那咱们废话不多说&#xff0c;直接开始吧&#xff01; 一、源码结构分析 1. SGISTL30实现剖析 // hash_set核心结构 template <class Value, class HashFcn, ...> class hash_set {ty…...

C# SqlSugar:依赖注入与仓储模式实践

C# SqlSugar&#xff1a;依赖注入与仓储模式实践 在 C# 的应用开发中&#xff0c;数据库操作是必不可少的环节。为了让数据访问层更加简洁、高效且易于维护&#xff0c;许多开发者会选择成熟的 ORM&#xff08;对象关系映射&#xff09;框架&#xff0c;SqlSugar 就是其中备受…...

VM虚拟机网络配置(ubuntu24桥接模式):配置静态IP

编辑-虚拟网络编辑器-更改设置 选择桥接模式&#xff0c;然后找到相应的网卡&#xff08;可以查看自己本机的网络连接&#xff09; windows连接的网络点击查看属性 编辑虚拟机设置更改网络配置&#xff0c;选择刚才配置的桥接模式 静态ip设置&#xff1a; 我用的ubuntu24桌…...

华为OD机考-机房布局

import java.util.*;public class DemoTest5 {public static void main(String[] args) {Scanner in new Scanner(System.in);// 注意 hasNext 和 hasNextLine 的区别while (in.hasNextLine()) { // 注意 while 处理多个 caseSystem.out.println(solve(in.nextLine()));}}priv…...

Git 3天2K星标:Datawhale 的 Happy-LLM 项目介绍(附教程)

引言 在人工智能飞速发展的今天&#xff0c;大语言模型&#xff08;Large Language Models, LLMs&#xff09;已成为技术领域的焦点。从智能写作到代码生成&#xff0c;LLM 的应用场景不断扩展&#xff0c;深刻改变了我们的工作和生活方式。然而&#xff0c;理解这些模型的内部…...

Vue 模板语句的数据来源

&#x1f9e9; Vue 模板语句的数据来源&#xff1a;全方位解析 Vue 模板&#xff08;<template> 部分&#xff09;中的表达式、指令绑定&#xff08;如 v-bind, v-on&#xff09;和插值&#xff08;{{ }}&#xff09;都在一个特定的作用域内求值。这个作用域由当前 组件…...