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

Thinkphp6使用RabbitMQ消息队列

Thinkphp6连接使用RabbitMQ(不止tp6,其他框架对应改下也一样),如何使用Docker部署RabbitMQ,在上一篇已经讲了->传送门<-。

部署环境

开始前先进入RabbitMQ的web管理界面,选择Queues菜单,点击底部的Add a new queue,新建一个test的队列。

安装thinkphp6框架

composer create-project topthink/think tp 

安装workerman扩展

composer require topthink/think-worker

安装rabbitmq扩展

composer require workerman/rabbitmq

代码编写

生产者

  • 在app目录下新建workerman目录,并在其下创建Send.php文件,$options数组中的host地址改成你的rabbitmq地址。
<?phpnamespace app\workerman;
use Bunny\Channel;
use Workerman\RabbitMQ\Client;
use think\worker\Server;
class Send extends Server
{//websocket地址,一会用于测试。protected $socket = 'websocket://127.0.0.1:2345';/*** 收到信息* @param $connection* @param $data*/public function onMessage($connection, $data){//websocket发送过来的消息$connection->send('我收到你的信息了:'.$data);//rabbitMQ配置$options = ['host'=>'127.0.0.1',//rabbitMQ IP'port'=>5672,//rabbitMQ 通讯端口'user'=>'admin',//rabbitMQ 账号'password'=>'123456'//rabbitMQ 密码];(new Client($options))->connect()->then(function (Client $client) {return $client->channel();})->then(function (Channel $channel) {/*** 创建队列(Queue)* name: ceshi         // 队列名称* passive: false      // 如果设置true存在则返回OK,否则就报错。设置false存在返回OK,不存在则自动创建* durable: true       // 是否持久化,设置false是存放到内存中RabbitMQ重启后会丢失,*                        设置true则代表是一个持久的队列,服务重启之后也会存在,因为服务会把持久化的Queue存放在硬盘上,当服务重启的时候,会重新加载之前被持久化的Queue* exclusive: false    // 是否排他,指定该选项为true则队列只对当前连接有效,连接断开后自动删除*  auto_delete: false // 是否自动删除,当最后一个消费者断开连接之后队列是否自动被删除*/return $channel->queueDeclare('ceshi', false, true, false, false)->then(function () use ($channel) {return $channel;});})->then(function (Channel $channel) use($data){echo "发送消息内容:".$data."\n";/*** 发送消息* body 发送的数据* headers 数据头,建议 ['content_type' => 'text/plain'],这样消费端是springboot注解接收直接是字符串类型* exchange 交换器名称* routingKey 路由key* mandatory* immediate* @return bool|PromiseInterface|int*/return $channel->publish($data, ['content_type' => 'text/plain'], '', 'ceshi')->then(function () use ($channel) {return $channel;});})->then(function (Channel $channel) {//echo " [x] Sent 'Hello World!'\n";$client = $channel->getClient();return $channel->close()->then(function () use ($client) {return $client;});})->then(function (Client $client) {$client->disconnect();});}/*** 当连接建立时触发的回调函数* @param $connection*/public function onConnect($connection){}/*** 当连接断开时触发的回调函数* @param $connection*/public function onClose($connection){}/*** 当客户端的连接上发生错误时触发* @param $connection* @param $code* @param $msg*/public function onError($connection, $code, $msg){echo "error $code $msg\n";}/*** 每个进程启动* @param $worker*/public function onWorkerStart($worker){}
}
  • 在config/worker_server.php中设置worker_class值为'app\workerman\Send'

  • 启动这个生产者

php think worker:server
方式1:通过tcp发送数据
  • 发送数据
    通过在线网址发送数据(websocket方式),->传送门<-
    输入【ws://127.0.0.1:2345】后点击发送数据!
    在这里插入图片描述
  • 前往rabbitMQ控制台查看
    在这里插入图片描述
    至此,生产这一步就走完了,那么如果我不想通过websocket方式,想用tcp方式生产数据怎么办?
方式2:通过tcp发送数据

接口给内置服务器发消息->内置服务去发消息给rabbitMQ

  • 将Send.php中websocket:127.0.0.1改成tcp:127.0.0.1
  • 重启服务
  • 把controller目录中Index.php修改为以下内容
<?php
namespace app\controller;use app\BaseController;class Index extends BaseController
{public function index(string $msg){//连接本地tcp服务$client = stream_socket_client('tcp://127.0.0.1:2345', $errno, $errmsg, 1);//发送字符串fwrite($client, $msg."\n");//断开服务fclose($client);return 'OK';}}
  • 用Postman访问对应接口就好,也会有数据进入队列

消费者

同生产者一样新创建一个thinkphp6项目,注意端口别和生产者冲突!这里我设置的是2346端口

  • 在app目录下新建workerman目录,并在其下创建Receive.php文件,$options数组中的host地址改成你的rabbitmq地址。
<?phpnamespace app\workerman;
use Bunny\Channel;
use Bunny\Message;
use Workerman\RabbitMQ\Client;
use think\worker\Server;
class Receive extends Server
{protected $socket = 'tcp://127.0.0.1:2346';/*** 收到信息* @param $connection* @param $data*/public function onMessage($connection, $data){}/*** 当连接建立时触发的回调函数* @param $connection*/public function onConnect($connection){}/*** 当连接断开时触发的回调函数* @param $connection*/public function onClose($connection){}/*** 当客户端的连接上发生错误时触发* @param $connection* @param $code* @param $msg*/public function onError($connection, $code, $msg){echo "error $code $msg\n";}/*** 每个进程启动* @param $worker*/public function onWorkerStart($worker){//rabbitMQ配置$options = ['host'=>'127.0.0.1',//rabbitMQ IP'port'=>5672,//rabbitMQ 通讯端口'user'=>'admin',//rabbitMQ 账号'password'=>'123456'//rabbitMQ 密码];(new Client($options))->connect()->then(function (Client $client) {return $client->channel();})->then(function (Channel $channel) {/*** 创建队列(Queue)* name: ceshi         // 队列名称* passive: false      // 如果设置true存在则返回OK,否则就报错。设置false存在返回OK,不存在则自动创建* durable: true       // 是否持久化,设置false是存放到内存中RabbitMQ重启后会丢失,*                        设置true则代表是一个持久的队列,服务重启之后也会存在,因为服务会把持久化的Queue存放在硬盘上,当服务重启的时候,会重新加载之前被持久化的Queue* exclusive: false    // 是否排他,指定该选项为true则队列只对当前连接有效,连接断开后自动删除*  auto_delete: false // 是否自动删除,当最后一个消费者断开连接之后队列是否自动被删除*/return $channel->queueDeclare('ceshi', false, true, false, false)->then(function () use ($channel) {return $channel;});})->then(function (Channel $channel) {echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";$channel->consume(function (Message $message, Channel $channel, Client $client) {echo "接收消息内容:", $message->content, "\n";},'ceshi','',false,true);});}
}
  • 在config/worker_server.php中设置worker_class值为'app\workerman\Receive',并将端口改为2346

  • 启动这个消费者

php think worker:server

到这里消费者也就结束啦!

使用

接下来我用cmd来启动两个服务,然后用接口发送消息和消费测试!
在这里插入图片描述




部分参考自:
https://www.workerman.net/doc/workerman/components/workerman-rabbitmq.html
https://blog.csdn.net/weixin_47723549/article/details/124493059

相关文章:

Thinkphp6使用RabbitMQ消息队列

Thinkphp6连接使用RabbitMQ&#xff08;不止tp6&#xff0c;其他框架对应改下也一样&#xff09;&#xff0c;如何使用Docker部署RabbitMQ&#xff0c;在上一篇已经讲了->传送门<-。 部署环境 开始前先进入RabbitMQ的web管理界面&#xff0c;选择Queues菜单&#xff0c;点…...

小成本互联网创业怎么做?低成本创业的方法分享

多数人都会有想法创业&#xff0c;尤其是在互联网上面创业&#xff0c;很多人看到了商机&#xff0c;但是因为成本的原因又放弃了&#xff0c;实际上&#xff0c;小成本也可以互联网创业&#xff01;那么&#xff0c;小成本互联网创业怎么做&#xff1f;低成本创业的方法在这里…...

六、栈、栈的相关问题

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 目录 前言 一、栈 1.栈概述 2.栈的实现 2.1 栈的API 2.2 栈的实现 二、栈的括号匹配问题 1.问题描述 2.代码实现 三、逆波兰表达式求值问题 1.问题描述 2.代码 总结 前言 提…...

Java安全停止线程

Thread 类虽提供了一个 stop() 方法&#xff08;已经被废弃&#xff09;&#xff0c;但由于 stop() 方法强制终止一个正在执行的线程&#xff0c;可能会造成数据不一致的问题&#xff0c;所以在生产环境中最好不要使用。 场景&#xff1a; 由于一些操作需要轮询处理&#xff…...

12 readdir 函数

前言 在之前 ls 命令 中我们可以看到, ls 命令的执行也是依赖于 opendir, readdir, stat, lstat 等相关操作系统提供的相关系统调用来处理业务 因此 我们这里来进一步看一下 更细节的这些 系统调用 我们这里关注的是 readdir 这个函数, 入口系统调用是 getdents 如下调试…...

Windows环境搭建Android开发环境-Android Studio/Git/JDK

Windows环境搭建Android开发环境-Android Studio/Git/JDK 因为休假回来后公司的开发环境由Ubuntu变为了Windows&#xff0c;所以需要重新配置一下开发环境。 工作多年第一次使用Windows环境进行开发工作&#xff0c;作次记录下来。 一、 Git安装 1.1git 标题软件下载 网址&…...

全国爱耳日丨听力受损严重有哪些解决办法

——【科学爱耳护耳&#xff0c;实现主动健康】随着数码电子设备使用越来越方便、日常使用时间越来越长&#xff0c;听力障碍、患上耳道疾病一系列问题也接踵而至&#xff0c;在当下我们必须重视听力健康&#xff0c;采取更科学的听音方式&#xff0c;保护听力健康&#xff0c;…...

【抽水蓄能电站】基于粒子群优化算法的抽水蓄能电站的最佳调度方案研究(Matlab代码实现)

&#x1f468;‍&#x1f393;个人主页&#xff1a;研学社的博客&#x1f4a5;&#x1f4a5;&#x1f49e;&#x1f49e;欢迎来到本博客❤️❤️&#x1f4a5;&#x1f4a5;&#x1f3c6;博主优势&#xff1a;&#x1f31e;&#x1f31e;&#x1f31e;博客内容尽量做到思维缜密…...

【异常】因多租户字段缺少导致Error updating database. Column ‘tenant_id‘ cannot be null

一、报错内容 org.springframework.dao.DataIntegrityViolationException: ### Error updating database. Cause: java.sql.SQLIntegrityConstraintViolationException: Column tenant_id cannot be null ### The error may exist in com/xxx/cloud/mall/admin/mapper/Goods…...

类和对象(上)

文章目录 面向对象的初步认知类的实例化this引用对象的构造及初始化封装static成员代码块内部类 对象的打印一、面向对象的初步认知 Java是一门纯面向对象的语言(Object Oriented Program&#xff0c;简称OOP)&#xff0c;在面向对象的世界里&#xff0c;一切皆为对象。在java中…...

Java经典面试题——谈谈 Java 反射机制,动态代理是基于什么原理?

典型回答 反射机制是 Java 语言提供的一种基本功能&#xff0c;赋予程序在运行时 自省&#xff08;introspect&#xff0c;官方用语&#xff09;的能力。通过反射我们可以直接操作类或者对象&#xff0c;比如获取某个对象的类定义&#xff0c;获取类声明的属性和方法&#xff…...

19 客户端服务订阅机制的核心流程

Nacos客户端服务订阅机制的核心流程 说起Nacos的服务订阅机制&#xff0c;大家会觉得比较难理解&#xff0c;那我们就来详细分析一下&#xff0c;那我们先从Nacos订阅的概述说起 Nacos订阅概述 Nacos的订阅机制&#xff0c;如果用一句话来描述就是&#xff1a;Nacos客户端通…...

教师论文|科技专著管理系统

技术&#xff1a;Java、JSP等摘要&#xff1a;随着计算机和互联网技术的发展&#xff0c;社会的信息化程度越来越高&#xff0c;各行各业只有适应这种发展趋势&#xff0c;才能增强自己的适应能力和竞争能力&#xff0c;不断发展壮大。大学作为教育的基地&#xff0c;是社会进步…...

骨传导耳机是什么意思,骨传导耳机的好处具体有哪些

​在这个全民都是手机的时代&#xff0c;各种蓝牙耳机&#xff0c;入耳式耳机&#xff0c;真无线耳机等各种款式琳琅满目。而骨传导耳机是一种全新的科技产物&#xff0c;顾名思义就是通过头骨振动将声音传至外耳内的耳机。由于无需入耳&#xff0c;不会对耳朵造成任何影响。那…...

elasticsearch—使用汇总

文档结构1、概念简介2、使用创景3、核心组件4、环境部署5、操作实践官方网站&#xff1a;https://www.elastic.co/cn/elasticsearch/ 官方手册&#xff1a;https://www.elastic.co/guide/en/elasticsearch/reference/8.6/getting-started.html 参考教程&#xff1a; A&#xff…...

聊一聊代码重构——我们为什么要代码重构

代码重构 事情的起因是在去年下半年&#xff0c;我们终于无法承受往年的历史包袱而决定开始进行代码重构。 在以前我们尝试过进行代码重构但是从来没有系统性的考虑过如何重构。在对代码重构的过程中很多经验都是来自《重构&#xff1a;改善既有代码的设计》这本书&#xff0c;…...

【Python学习笔记】第二十九节 Python2 和Python3发生了哪些变化

Python 版本分为两大流派&#xff0c;一个是 Python 2.x 版本&#xff0c;另外一个是 Python 3.x 版本&#xff0c;Python 官方同时提供了对这两个版本的支持和维护。2020 年 1 月 1 日&#xff0c;Python 官方终止了对 Python 2.7 版本&#xff08;最后一个 Python 2.x 版本&a…...

[oeasy]python0099_雅达利大崩溃_IBM的开放架构_兼容机_oem

雅达利大崩溃 回忆上次内容 个人计算机浪潮已经来临 苹果公司迅速发展微软公司脱离mits准备做纯软件公司IBM用大型机思路制作的5100惨败 Commodore 64 既做计算机又做游戏机 计算机行业和游戏行业 跟随着底层技术不断迭代已经进入了战乱纷纷的年代最终又会如何呢&#xff1f…...

学术论文投稿之同行评审过程中可能会遭遇哪些偏见?

同行评审过程的顺利进行&#xff0c;在很大程度上取决于学术界的积极参与和相互信任&#xff0c;以及需要参与各方都以负责任的态度行事。作为审稿专家&#xff0c;向作者提供公正、客观的评价是至关重要的。同行评审过程中&#xff0c;若有任何偏离客观性的行为&#xff0c;均…...

Python写一个自动发送直播弹幕的工具,非常简单

哈喽大家好&#xff0c;今天给大家用Python整一个可以在直播间自动发弹幕的工具&#xff0c;来为喜欢的主播疯狂扣6 &#xff01; 事情原由昨晚回家&#xff0c;表弟在看LOL直播&#xff0c;看得我气不打一处来&#xff0c;差点就想锤他。 身为程序员的表弟&#xff0c;看直…...

探索前沿技术趋势:2024年最值得关注的创新应用场景

1. 生成式AI的爆发式应用 2024年最让人兴奋的技术趋势&#xff0c;莫过于生成式AI从实验室走向千家万户。我最近测试了十几个主流AI创作工具&#xff0c;发现它们已经能完成许多过去认为"只有人类能做到"的任务。比如用Midjourney生成产品设计图&#xff0c;只需要简…...

Easy-Scraper:革新性HTML数据提取库的技术突破与实战应用

Easy-Scraper&#xff1a;革新性HTML数据提取库的技术突破与实战应用 【免费下载链接】easy-scraper Easy scraping library 项目地址: https://gitcode.com/gh_mirrors/ea/easy-scraper 在数据驱动决策的时代&#xff0c;网页数据采集已成为企业获取市场情报、科研机构…...

5个核心功能提升音频处理效率:AsrTools语音转文字工具用户指南

5个核心功能提升音频处理效率&#xff1a;AsrTools语音转文字工具用户指南 【免费下载链接】AsrTools ✨ AsrTools: Smart Voice-to-Text Tool | Efficient Batch Processing | User-Friendly Interface | No GPU Required | Supports SRT/TXT Output | Turn your audio into a…...

NXP S32K3xx之HSE密钥管理与安全服务实战

1. HSE密钥管理基础&#xff1a;从零开始理解安全引擎 第一次接触NXP S32K3xx的HSE模块时&#xff0c;我被各种密钥术语搞得晕头转向。经过几个实际项目的打磨&#xff0c;现在我可以负责任地告诉你&#xff1a;理解HSE密钥管理就像学习一门新语言&#xff0c;掌握基础词汇后就…...

ESP32硬件定时器虚拟化:16路ISR定时器实现原理与工程实践

1. ESP32_New_TimerInterrupt 库深度解析&#xff1a;16路高精度硬件定时器中断的工程实践1.1 为什么嵌入式系统迫切需要此库在ESP32系列微控制器的实际工程开发中&#xff0c;硬件定时器资源极其稀缺且关键。标准ESP32芯片仅配备两组定时器组&#xff08;Timer Group 0/1&…...

数据稠密计算的算法优化:从理论到实践

数据稠密计算的算法优化&#xff1a;从理论到实践 引言 作为一名在数据深渊里捞了十几年 Bug 的女码农&#xff0c;我见过太多因为算法选择不当导致的性能问题。在数据稠密计算中&#xff0c;算法的选择和优化是提升计算性能的关键因素之一。今天&#xff0c;我们来聊聊数据稠密…...

软件测试学习第一期

&#x1f3ac; 博客主页&#xff1a;博主链接 &#x1f3a5; 本文由 M malloc 原创&#xff0c;首发于 CSDN&#x1f649; &#x1f384; 学习专栏推荐&#xff1a;LeetCode刷题集&#xff01; &#x1f3c5; 欢迎点赞 &#x1f44d; 收藏 ⭐留言 &#x1f4dd; 如有错误敬请指…...

做了5年GEO优化,我敢说90%的企业都没看懂GEO的真实成本

很多人来问我 GEO 是什么意思&#xff0c;大多是听别人说这是 AI 时代的获客新路子&#xff0c;能比传统推广省好几倍的钱&#xff0c;还能让 AI 优先推荐自己家。但我每次都先不说那些好听的好处&#xff0c;先给大家算清楚&#xff0c;做 GEO 这件事里&#xff0c;那些 90% 的…...

R语言新手必看:clusterProfiler功能富集分析从安装到实战(附常见报错解决方案)

R语言实战&#xff1a;clusterProfiler功能富集分析全流程指南 第一次接触功能富集分析时&#xff0c;我被那些密密麻麻的基因列表和复杂的生物学术语搞得晕头转向。直到发现了clusterProfiler这个神器&#xff0c;它就像生物信息学分析中的瑞士军刀&#xff0c;把复杂的富集过…...

【国家级等保2.0合规必读】:Python扩展模块安全开发规范(含12项强制检查项+自动化检测脚本)

第一章&#xff1a;Python扩展模块安全开发概述Python 扩展模块&#xff08;C/C 编写的 .so/.dll 文件&#xff09;是提升性能、复用底层库或与系统交互的关键手段&#xff0c;但其直接操作内存、绕过 Python 运行时保护机制的特性&#xff0c;也使其成为安全风险的高发区。开发…...