tp6 RabbitMQ

news/2024/9/1 2:58:02 标签: rabbitmq
1、composer 安装 AMQP 扩展
composer require php-amqplib/php-amqplib
2、RabbitMQ 配置

 在 config 目录下创建 rabbitmq.php 文件

<?php
return [
    'host'=>'',
    'port'=>'5672',
    'user'=>'',
    'password'=>'',
    'vhost'=>'',
    'exchange_name' => '',
    'queue_name' => '',
    'route_key' => '',
    'consumer_tag' => '',
];
3、生产者代码

app目录下创建Producer.php

<?php

namespace app;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Message\AMQPMessage;

class Producer
{

    private $connection;
    private $channel;
    private $mq_config;

    public function __construct()
    {
        $this->mq_config = config('rabbit_mq');
        $this->connection = new AMQPStreamConnection(
            $this->mq_config['host'],
            $this->mq_config['port'],
            $this->mq_config['user'],
            $this->mq_config['password'],
            'itcast'
        );
        //创建通道
        $this->channel = $this->connection->channel();
    }

    public function send($data)
    {
        /**
         * 创建队列(Queue)
         * name: hello         // 队列名称
         * passive: false      // 如果设置true存在则返回OK，否则就报错。设置false存在返回OK，不存在则自动创建
         * durable: true       // 是否持久化，设置false是存放到内存中的，RabbitMQ重启后会丢失;设置true,则代表是一个持久化的队列，服务重启后也会存在，因为服务会把持久化的queue存放到磁盘上当服务重启的时候，会重新加载之前被持久化的queue
         * exclusive: false    // 是否排他，指定该选项为true则队列只对当前连接有效，连接断开后自动删除
         * auto_delete: false // 是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除
         *
         */
        $this->channel->queue_declare($this->mq_config['queue_name'], false, true, false, false);

        /**
         * 创建交换机(Exchange)
         * name: vckai_exchange// 交换机名称
         * type: direct        // 交换机类型，分别为direct/fanout/topic，参考另外文章的Exchange Type说明。
         * passive: false      // 如果设置true存在则返回OK，否则就报错。设置false存在返回OK，不存在则自动创建
         * durable: false      // 是否持久化，设置false是存放到内存中的，RabbitMQ重启后会丢失
         * auto_delete: false  // 是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除
         */
        $this->channel->exchange_declare($this->mq_config['exchange_name'], AMQPExchangeType::DIRECT, false, true, false);

        // 绑定消息交换机和队列
        $this->channel->queue_bind($this->mq_config['queue_name'], $this->mq_config['exchange_name'],$this->mq_config['route_key']);

        $messageBody = json_encode($data);//将要发送数据变为json字符串

        /**
         * 创建AMQP消息类型
         * delivery_mode 消息是否持久化
         * AMQPMessage::DELIVERY_MODE_NON_PERSISTENT  不持久化
         * AMQPMessage::DELIVERY_MODE_PERSISTENT      持久化
         */
        $message = new AMQPMessage($messageBody, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));

        /**
         * 发送消息
         * msg: $message            // AMQP消息内容
         * exchange: vckai_exchange // 交换机名称
         * routing_key: hello       // 路由key
         */
        $this->channel->basic_publish($message, $this->mq_config['exchange_name'], $this->mq_config['route_key']);

        //关闭连接
        $this->stop();

    }

    //关闭进程
    public function stop()
    {
        $this->channel->close();
        $this->connection->close();
    }

}
4、消费者代码

app目录下创建Consumer.php

<?php

namespace app;

use app\index\controller\ApiCommunity;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use think\db\exception\PDOException;
use think\facade\Log;

class Consumer
{
    private $connection;
    private $channel;
    private $mq_config;

    public function __construct()
    {
        $this->mq_config = config('rabbit_mq');
        $this->connection = new AMQPStreamConnection(
            $this->mq_config['host'],
            $this->mq_config['port'],
            $this->mq_config['user'],
            $this->mq_config['password'],
            $this->mq_config['vhost']
        );
        //创建通道
        $this->channel = $this->connection->channel();
    }

    /**
     * @param $channel
     * @param $connection
     * 关闭进程
     */
    function shutdown($channel, $connection)
    {
        $channel->close();
        $connection->close();
    }

    /**
     * @param $message
     * 消息处理
     */
    function process_message($message)
    {
        //消息处理逻辑
        echo $message->body . "\n";

        if ($message->body !== 'quit') {
            $obj = json_decode($message->body);
            if (!isset($obj->id)) {
                Log::write("error data:" . $message->body, 2);
            } else {
                try {
                    Log::write("data:" . json_encode($message));
                    //消息处理
                } catch (\Think\Exception  $e) {
                    Log::write($e->getMessage(), 2);
                    Log::write(json_encode($message), 2);
                } catch (PDOException $pe) {
                    Log::write($pe->getMessage(), 2);
                    Log::write(json_encode($message), 2);
                }
            }
        }

        // 手动确认ack，确保消息已经处理
        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
        // Send a message with the string "quit" to cancel the consumer.
        if ($message->body === 'quit') {
            $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
        }
    }


    /**
     * @throws \ErrorException
     * 启动
     *
     * nohup php index.php index/Message_Consume/start &
     */
    public function start()
    {
        // 设置消费者（Consumer）客户端同时只处理一条队列
        // 这样是告诉RabbitMQ，再同一时刻，不要发送超过1条消息给一个消费者（Consumer），直到它已经处理了上一条消息并且作出了响应。这样，RabbitMQ就会把消息分发给下一个空闲的消费者（Consumer）。
        //消费者端要把自动确认autoAck设置为false，basic_qos才有效。
        //$this->channel->basic_qos(0, 1, false);

        // 同样是创建路由和队列，以及绑定路由队列，注意要跟producer（生产者）的一致
        // 这里其实可以不用设置，但是为了防止队列没有被创建所以做的容错处理
        $this->channel->queue_declare($this->mq_config['queue_name'], false, true, false, false);
        $this->channel->exchange_declare($this->mq_config['exchange_name'], AMQPExchangeType::DIRECT, false, true, false);
        $this->channel->queue_bind($this->mq_config['queue_name'], $this->mq_config['exchange_name'], $this->mq_config['route_key']);

        /**
         *
         * queue: queue_name    // 被消费的队列名称
         * consumer_tag: consumer_tag // 消费者客户端身份标识，用于区分多个客户端
         * no_local: false      // 这个功能属于AMQP的标准，但是RabbitMQ并没有做实现
         * no_ack: true         // 收到消息后，是否不需要回复确认即被认为被消费
         * exclusive: false     // 是否排他，即这个队列只能由一个消费者消费。适用于任务不允许进行并发处理的情况下
         * nowait: false        // 不返回执行结果，但是如果排他开启的话，则必须需要等待结果的，如果两个一起开就会报错
         * callback: $callback  // 回调逻辑处理函数
         *
         */
        $this->channel->basic_consume($this->mq_config['queue_name'], $this->mq_config['consumer_tag'], false, false, false, false, array($this, 'process_message'));

        register_shutdown_function(array($this, 'shutdown'), $this->channel, $this->connection);

        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }
    }
}
5、创建自定义命令
php think make:command Consumer

在项目跟目录执行以下命令，会自动生成 在 command 目录生成 Consumer 控制器 

<?php
declare (strict_types = 1);

namespace app\command;

use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;

class Consumer extends Command
{
    protected function configure()
    {
        // 指令配置
        $this->setName('consumer')
            ->setDescription('the consumer command');
    }

    protected function execute(Input $input, Output $output)
    {
        // 指令输出
        $output->writeln('consumer');
        $consumer = new \app\Consumer();
//        $consumer->process_message(11)
        $consumer->start();

    }
}

config/console.php 代码增加如下：

// 指令定义
'commands' => [
    'consumer' => 'app\command\Consumer',
],
6、命令

消费者命令

php think consumer

 生产者执行命令

$producer = new Producer();
$data = [
    'message' => "发送的消息内容"
];
$producer->send($data);


http://www.niftyadmin.cn/n/4937214.html

相关文章

返回参数规范

返回参数规范

项目目录 目录接口参考 项目目录结构设计&#xff0c;增加部分领域模型后缀强制定义&#xff0c;方便统一编码风格。 controller&#xff1a;请求处理 RestController module&#xff1a;按大业务区分&#xff0c;对多个业务对象数据聚合处理 Component manager&#xff1a;…
阅读更多...
PDF文件限制编辑怎么取消？

PDF文件限制编辑怎么取消？

PDF文件设置了限制编辑&#xff0c;想要取消PDF文件的限制编辑&#xff0c;很简单&#xff0c;打开PDF编辑器&#xff0c;点击工具栏中的文件&#xff0c;选择属性按钮&#xff0c;进入到熟悉感界面之后&#xff0c;点击安全&#xff0c;然后我们点击权限下拉框&#xff0c;选择…
阅读更多...
【Vue-Router】路由跳转

【Vue-Router】路由跳转

1. 路由标签 App.vue <template><h1>hello world</h1><div><router-link to"/">Login</router-link><router-link style"margin: 10px;" to"/reg">Reg</router-link></div><hr>&…
阅读更多...
【Go 基础篇】Go语言字符类型：解析字符的本质与应用

【Go 基础篇】Go语言字符类型：解析字符的本质与应用

介绍 字符类型是计算机编程中用于表示文本和字符的数据类型&#xff0c;是构建字符串的基本单位。在Go语言&#xff08;Golang&#xff09;中&#xff0c;字符类型具有独特的特点和表示方式&#xff0c;包括Unicode编码、字符字面值以及字符操作。本篇博客将深入探讨Go语言中的…
阅读更多...
metaRTC7 demo mac/ios编译指南

metaRTC7 demo mac/ios编译指南

概要 metaRTC7.0开始全面支持mac/ios操作系统&#xff0c;新版本7.0.023 mac os demo 包含有srs/zlm的推拉流演示。发布版自带了x64版第三方类库&#xff0c;arm版第三方类库还需开发者自己编译。 源码下载 下载文件metartc7.023.7z https://github.com/metartc/metaRTC/re…
阅读更多...
修改Linux中SSH的端口

修改Linux中SSH的端口

文章目录 修改Linux中SSH的端口Linux中默认的ssh端口关闭SELinux测试新端口 修改Linux中SSH的端口 Linux中默认的ssh端口 使用root用户操作 修改前先备份ssh_config cp /etc/ssh/sshd_config /etc/ssh/sshd_config_date "%Y%m%d%H%M%S"修改配置文件&#xff0c;找…
阅读更多...
RunLoop

RunLoop

1.CFRunLoopModeRef特征代表RunLoop对象内的运行模式(每个RunLoop对象内存中存在很多种运行模式,每个Mode运行模式下必然包含若干个有效的Source0/Source1/Timer/Observer数据序组) 2.RunLoop对象活跃(操作)启动时能且仅能选择某个Mode匹配currentMode(暗示Loop对象的操作运行必…
阅读更多...
最新 python 爱心代码？

最新 python 爱心代码？

python程序代码&#xff1a;heart.py from math import cos, pi import numpy as np import cv2 import os, globclass HeartSignal:def __init__(self, curve"heart", title"Love U", frame_num20, seed_points_num2000, seed_numNone, highlight_rate0.…
阅读更多...
最新文章

Copyright @ 2022~2023