PHP实现RabbitMQ消息列队的示例代码

目录
  • 业务场景
  • 1、首先部署好thinkphp6框架
  • 2、安装workerman扩展
  • 3、生产者
  • 4、消费者
  • 5、整体测试

业务场景

项目公司是主php做开发的,框架为thinkphp。众所周知,php本身的运行效率存在一定的缺陷,所以如果有一个很复杂很耗时的业务时,必须开发一个常驻内存的程序。首先我想到了php的workerman与swoole,但是这里应上面的标题哈,想将耗时任务交给另一个服务器,同时列队处理。所以这里我想独立部署一个rabbitMQ服务器用于处理列队任务。

当rabbitMQ服务器我们准备好了,建立了一个持久化命名为ceshi的列队,如下:

项目上生产者和消费者的开发我这里全部采用tinkphp6+workerman,为便于管理。这里这么做也是因为发现workerman中对rabbitMQ的文档解释太少了!

所以开始踩坑!

1、首先部署好thinkphp6框架

过程去看thinkphp6手册

2、安装workerman扩展

过程去看thinkphp6手册

3、生产者

配置一个workerman类

创建的Send类代码如下:

<?php

namespace 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)
{

    }
}

上述都OK以后咱们可以项目路径下通过命令启动这个生产者:

php think worker:server

测试发送数据:

通过这个网站

连接【ws://127.0.0.1:2345】后发送数据!

前往rabbitMQ控制台

列队中有一条消息产生并且等待了!

这个时候你可能问,如果我发送数据不想通过ws发送而是接口发送怎么办?

笨思路呗:接口给内置服务器发消息->内置服务去发消息给rabbitMQ

将协议改为tcp

然后重新启动服务

然后去tp6创建一个路由接口

接口代码

<?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';
    }

}

执行结果:

说明接口成功的将数据发送给了本地内置的tcp服务。

同时,内置服务将收到的数据给了rabbitMQ服务列队中。

生产者完成。

4、消费者

同生产者一样新创建一个thinkphp6及安装workerman扩展,注意端口别和生产者冲突!这里我设置的是2346端口

创建的Receive类代码如下:

<?php

namespace 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
            );
        });

    }
}

都OK以后咱们可以项目路径下通过命令启动这个消费者:

php think worker:server

此时应该会自动消费掉rabbitMQ中等待的消息!

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

5、整体测试

接下来我用cmd来启动两个服务,然后用接口发送消息和消费测试!

至于具体怎么灵活应用自行开拓大脑哦~

比如php项目有些业务吃力,可以去做个java的消费端,让java来完成任务~

以上就是PHP实现RabbitMQ消息列队的示例代码的详细内容,更多关于PHP RabbitMQ消息列队的资料请关注我们其它相关文章!

时间: 2022-05-10

PHP+RabbitMQ实现消息队列的完整代码

前言 为什么使用RabbitMq而不是ActiveMq或者RocketMq? 首先,从业务上来讲,我并不要求消息的100%接受率,并且,我需要结合php开发,RabbitMq相较RocketMq,延迟较低(微妙级).至于ActiveMq,貌似问题较多.RabbitMq对各种语言的支持较好,所以选择RabbitMq. 先安装PHP对应的RabbitMQ,这里用的是 php_amqp 不同的扩展实现方式会有细微的差异. php扩展地址: http://pecl.php.net/package/amq

php+redis实现消息队列功能示例

本文实例讲述了php+redis实现消息队列功能.分享给大家供大家参考,具体如下: 个人理解在项目中使用消息队列一般是有如下几个原因: 把瞬间服务器的请求处理换成异步处理,缓解服务器的压力 实现数据顺序排列获取 redis实现消息队列步骤如下: 1).redis函数rpush,lpop 2).建议定时任务入队列 3)创建定时任务出队列 文件:demo.php插入数据到redis队列 <?php $redis = new Redis(); $redis->connect('127.0.0.1',

使用PHP访问RabbitMQ消息队列的方法示例

本文实例讲述了使用PHP访问RabbitMQ消息队列的方法.分享给大家供大家参考,具体如下: 扩展安装 PHP访问RabbitMQ实际使用的是AMQP协议,所以我们只要安装epel库中的php-pecl-amqp这个包即可 rpm -ivh http://mirror.neu.edu.cn/fedora/epel/6/x86_64/epel-release-6-8.noarch.rpm yum install php-pecl-amqp 交换建立 <?php $connection = new

php+redis消息队列实现抢购功能

本文实例为大家分享了php+redis消息队列实现抢购的具体代码,供大家参考,具体内容如下 实现功能: 1. 基于redis队列,防止高并发的超卖 2. 基于mysql的事务加排它锁,防止高并发的超卖 基于redis队列工作流程: 1. 管理员根据goods表中的库存,创建redis商品库存队列 2. 客户端访问秒杀API 3. web服务器先从redis的商品库存队列中查询剩余库存重点内容 4. redis队列中有剩余,则在mysql中创建订单,去库存,抢购成功 5. redis队列中没有剩余

PHP基于rabbitmq操作类的生产者和消费者功能示例

本文实例讲述了PHP基于rabbitmq操作类的生产者和消费者功能.分享给大家供大家参考,具体如下: 注意事项: 1.accept.php消费者代码需要在命令行执行 2.'username'=>'asdf','password'=>'123456' 改成自己的帐号和密码 RabbitMQCommand.php操作类代码 <?php /* * amqp协议操作类,可以访问rabbitMQ * 需先安装php_amqp扩展 */ class RabbitMQCommand{ public $

JavaScript基于DOM操作实现简单的数学运算功能示例

本文实例讲述了JavaScript基于DOM操作实现简单的数学运算功能.分享给大家供大家参考,具体如下: <!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd"> <html xmlns="http://www.w3.org/1999/xhtml"

基于C#实现的多生产者多消费者同步问题实例

本文实例讲述了基于C#实现的多生产者多消费者同步问题,分享给大家供大家参考之用.具体代码如下: // 多个生产者和多个消费者,能生产n个产品的情况 using System; using System.Threading; public class HoldIntegerSynchronized{ private int[] buffer; //缓冲区 private int occupiedBufferCount = 0; private int readPosition = 0 , writ

python基于socket实现的UDP及TCP通讯功能示例

本文实例讲述了python基于socket实现的UDP及TCP通讯功能.分享给大家供大家参考,具体如下: Server: import socket address = ('127.0.0.1', 31500) s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.bind(address) while True: data, addr = s.recvfrom(2048) if not data: print "client has ex

基于代数方程库Algebra.js解二元一次方程功能示例

本文实例讲述了基于代数方程库Algebra.js解二元一次方程功能.分享给大家供大家参考,具体如下: 假设二元一次方程如下: x + y = 11 x - y = 5 解方程如下: <!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8" /> <meta http-equiv="X-UA-Compatible" cont

C#实现String类型和json之间的相互转换功能示例

本文实例讲述了C#实现String类型和json之间的相互转换功能.分享给大家供大家参考,具体如下: ////Donet2.0 需要添加引用 // 从一个对象信息生成Json串 public static string ObjectToJson(object obj) { return JavaScriptConvert.SerializeObject(obj); } // 从一个Json串生成对象信息 public static object JsonToObject(string jsonS

JS基于面向对象实现的多个倒计时器功能示例

本文实例讲述了JS基于面向对象实现的多个倒计时器功能.分享给大家供大家参考,具体如下: 运行效果图如下: 实现代码如下: 代码 <!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd"> <html xmlns="http://www.w3.org/1999/xhtml"> &l

jQuery基于ajax方式实现用户名存在性检查功能示例

本文实例讲述了jQuery基于ajax方式实现用户名存在性检查功能.分享给大家供大家参考,具体如下: 对于拥有会员功能的网站,尤其是会员登录后可以留言或评论的网站,一般要求不能有两个或两个以上相同的用户名存在.因此,在用户注册的时就需要对用户名是否已经被注册进行检查防止出现相同的用户名.下面是我实现这种功能的一种解决方案. 1.方案原理:利用ajax的异步请求不刷新正在注册的页面向后端发送请求,后端对请求数据进行处理返回用户名是否已经存在的结果. 2.方案详情 (1)html代码部分,运用了in

Python基于TCP实现会聊天的小机器人功能示例

本文实例讲述了Python基于TCP实现会聊天的小机器人功能.分享给大家供大家参考,具体如下: 一 代码 1.服务端程序 import socket words ={'how are you?':'Fine,thank you.', 'how old are you?':'38', 'what is your name?':'Dong FuGuo', "what's your name?":'Dong FuGuo', 'where do you work?':'SDIBT', 'bye

jQuery基于闭包实现的显示与隐藏div功能示例

本文实例讲述了jQuery基于闭包实现的显示与隐藏div功能.分享给大家供大家参考,具体如下: <div class="binds"> <div class="phonebind"> <h3>手机绑定</h3> <p>当前手机号码:<span id="oldPhone">$!{user.phone}</span><input type="butto