PHP+RabbitMQ消息发布与订阅

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› PHP›PHP编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

我们有一个小说系统，每天会有很多作者发布新的小说内容，而读者因为个人爱好可能只订阅他喜欢的类型的小说，比如历史类、玄幻类小说。小说系统每天会根据用户的口味推送相关的小说更新消息，这就用到了消息发布和订阅系统。

本文将结合实例介绍PHP操作RabbitMQ实现消息发布和订阅功能，本文假设您已经安装好RabbitMQ，开放了对应的端口，且安装了php-amqplib。前置文章阅读：《在CentOS7系统安装与配置RabbitMQ》、《PHP与RabbitMQ消息队列》。

概念

我们先来了解几个概念：

交换器(Exchanges)

RabbitMQ消息传递模型的核心思想是，生产者不发送任何信息直接到队列。事实上，生产者甚至不知道消息是否会发送到任何队列。生产者只能向交换器发送消息（也叫交换机，默认交换器使用""空字符标记）。交换器需要知道如何处理接收的消息，将消息推入到指定的队列中，决定消息是否入列和抛弃。如下图，P表示消息发布者，X表示交换机，Q1和Q2表示不同的队列。

交换类型

交换机有几种类型：direct, topic, headers 和 fanout。

fanout：广播订阅，向所有的消费者发布消息。每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。fanout 类型转发消息是最快的。

direct：消息中的路由键（routing key）如果和 Binding 中的 binding key 一致，交换器就将消息发到对应的队列中。路由键与队列名完全匹配，如果一个队列绑定到交换机要求路由键为“dog”，则只转发 routing key 标记为“dog”的消息，不会转发“dogA”，也不会转发“dogB”等等。它是完全匹配、单播的模式。

topic：topic 交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。它同样也会识别两个通配符：符号“#”和符号“*”。#匹配0个或多个单词，*匹配不多不少一个单词。

headers类型的交换器基本不用，本文忽略。

举例：以下代码，发布者向名叫msg的交换器发布广播消息，全体消费者都能收到相同的消息。

$channel->exchange_declare('msg', 'fanout', false, false, false);

绑定(Bindings)

交换器和队列之间的对应关系称为绑定，可以理解为，队列对来自此交换器的消息感兴趣。

以下代码表示将队列绑定到名叫article的交换器上。

$channel->queue_bind($queue_name, 'article');

路由键

绑定可以采取额外的routing_key参数。避免混淆和$channel::basic_publish参数我们要叫它绑定key。这就是我们如何用键创建绑定的原因：

$routerKey = 'abc';
    $channel->queue_bind($queueName, $exchange, $routerKey);

消息发布

我们创建发布者文件publish_direct.php，指定交换机为article，类型为direct，我们只允许订阅了对应类型小说文章的消费者才可以消费对应的小说文章消息。我们将向消费者发布四个类型的小说文章消息：fantasy(玄幻)，military(军事)，history(历史)，romance(言情)。

以下代码模拟了发布者发布100条随机消息：

<?php
/**
 * @发布消息
 * @Author: Helloweba
 * @publish_direct.php
 */

require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$exchange = 'article';

$connection = new AMQPStreamConnection('192.168.0.100', 56720, 'helloweba', 'helloweba', 'test');
$channel = $connection->channel();

$channel->exchange_declare($exchange, 'direct', false, false, false);

for ($i = 0; $i < 100; $i++) { 
    $cate = ['fantasy', 'military', 'history', 'romance'];
    $key = array_rand($cate);

    $arr = [
        'id' => 'message_' . $i,
        'content' => 'helloweba '. $cate[$key]
    ];
    $data = json_encode($arr);
    $msg = new AMQPMessage($data);

    $channel->basic_publish($msg, $exchange, $cate[$key]);
    echo 'Send '.$cate[$key].' message: ' . $data . PHP_EOL;
}

$channel->close();
$connection->close();

消息订阅

现在我们建立订阅者文件subscribe_direct.php，指定交换机为article，路由键为fantasy，意为只订阅玄幻小说类消息，代码如下：

channel();
$channel->exchange_declare($exchange, 'direct', false, false, false);
list($queueName, ,) = $channel->queue_declare("", false, false, true, false);

$channel->queue_bind($queueName, $exchange, $routerKey);

echo " [*] Waiting for messages. To exit press CTRL+C" .PHP_EOL;
$callback = function ($msg) {
    //echo " Received message：", $msg->body, PHP_EOL;
    echo ' Received message：',$msg->delivery_info['routing_key'], ':', $msg->body, PHP_EOL;
    sleep(1);  //模拟耗时执行
};
$channel->basic_consume($queueName, '', false, true, false, false, $callback);

while ($channel->is_consuming()) {
    $channel->wait();
}

$channel->close();
$connection->close();

接着再新建订阅者文件subscribe_direct_2.php，复制粘贴subscribe_direct.php文件的代码，并将路由键改为history，意为只订阅历史小说类消息

模拟测试

好了，现在我们打开两个终端，分别执行两个订阅者程序：

php subscribe_direct.php //client1订阅玄幻小说类消息
php subscribe_direct_2.php //client2订阅历史小说类消息

再另开启一个终端，执行发布者程序：

php publish_direct.php

现在你应该可以看到如图效果：

client1，只订阅玄幻类（fantasy）消息:

client2，只订阅历史类（history）消息:

publish_direct.php：

<?php

/**
 * 发布消息
 * @Author: Helloweba
 * @Date:   2020-01-01 20:24:22
 * @Last Modified by:   Helloweba
 * @Last Modified time: 2020-01-05 20:29:46
 */

require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$exchange = 'article';

$connection = new AMQPStreamConnection('192.168.0.100', 56720, 'helloweba', 'helloweba', 'test');
$channel = $connection->channel();

//$channel->exchange_declare($exchange, 'fanout', false, false, false);
$channel->exchange_declare($exchange, 'direct', false, false, false);

for ($i = 0; $i < 100; $i++) { 
    $cate = ['fantasy', 'military', 'history', 'romance'];
    $key = array_rand($cate);

    $arr = [
        'id' => 'message_' . $i,
        'content' => 'helloweba '. $cate[$key]
    ];
    $data = json_encode($arr);
    $msg = new AMQPMessage($data);

    $channel->basic_publish($msg, $exchange, $cate[$key]);
    echo 'Send '.$cate[$key].' message: ' . $data . PHP_EOL;
}

$channel->close();
$connection->close();

subscribe_direct.php：

<?php

/**
 * 订阅消息
 * @Author: Helloweba
 * @Date:   2020-01-01 20:24:57
 * @Last Modified by:   Helloweba
 * @Last Modified time: 2020-01-05 20:17:16
 */

require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$exchange = 'article';
$routerKey = 'fantasy'; //只消费玄幻类消息

//$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$connection = new AMQPStreamConnection('localhost', 56720, 'helloweba', 'helloweba', 'test');
$channel = $connection->channel();
$channel->exchange_declare($exchange, 'direct', false, false, false);
list($queueName, ,) = $channel->queue_declare("", false, false, true, false);

$channel->queue_bind($queueName, $exchange, $routerKey);

echo " [*] Waiting for messages. To exit press CTRL+C" .PHP_EOL;
$callback = function ($msg) {
    //echo " Received message：", $msg->body, PHP_EOL;
    echo ' Received message：',$msg->delivery_info['routing_key'], ':', $msg->body, PHP_EOL;
    sleep(1);  //模拟耗时执行
};
$channel->basic_consume($queueName, '', false, true, false, false, $callback);

while ($channel->is_consuming()) {
    $channel->wait();
}

$channel->close();
$connection->close();

subscribe_direct_2.php：

<?php

/**
 * 订阅消息
 * @Author: Helloweba
 * @Date:   2020-01-01 20:24:57
 * @Last Modified by:   Helloweba
 * @Last Modified time: 2020-01-05 20:18:25
 */

require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$exchange = 'article';
$routerKey = 'history';

//$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$connection = new AMQPStreamConnection('localhost', 56720, 'helloweba', 'helloweba', 'test');
$channel = $connection->channel();
$channel->exchange_declare($exchange, 'direct', false, false, false);
list($queueName, ,) = $channel->queue_declare("", false, false, true, false);

$channel->queue_bind($queueName, $exchange, $routerKey);

//$channel->queue_bind($queue_name, $exchange);
echo " [*] Waiting for messages. To exit press CTRL+C" .PHP_EOL;
$callback = function ($msg) {
    echo ' Received message：',$msg->delivery_info['routing_key'], ':', $msg->body, PHP_EOL;
    sleep(1);  //模拟耗时执行
};
$channel->basic_consume($queueName, '', false, true, false, false, $callback);

while ($channel->is_consuming()) {
    $channel->wait();
}

$channel->close();
$connection->close();

转载 ; https://www.helloweba.net/php/626.html

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

php—cURL库基本用法总结发布时间：2022-07-10

PHP中字符串比较的常用方法发布时间：2022-07-10

CVE-2020-35163

1 dphi-official/Machine_Learning_Bootcamp

dphi-official/Machine_Learning_Bootcamp

阅读：1436|2022-08-18

2 delphi自定义窗口消息

假设你的消息这样定义： Const WM_YOURMSG=WM_USER+100; 你的过程定义

阅读：600|2022-07-18

3 juven/maven-bash-completion: Maven Bash

juven/maven-bash-completion: Maven Bash Auto Completion

阅读：859|2022-08-16

4 route101/mastoinker: Quick image view as

route101/mastoinker: Quick image view assistant for Mastodon client

阅读：971|2022-08-18

5 MATLAB地图工具箱学习总结（三）地图工具箱

MATLAB地图工具箱学习总结（三）地图工具箱的基本知识 MATLAB地图工具箱学习总结系列

阅读：883|2022-07-18

6 配置VLFeat-0.9.20+MATLAB2013a

在VLFeat官网上是这么介绍VLFeat的：VLFeat开源库实现了流行的计算机视觉算法，专门从

阅读：930|2022-07-18

7 CVE-2022-34819

A vulnerability has been identified in SIMATIC CP 1242-7 V2 (All versions), SIMA

阅读：752|2022-07-29

8 0e0w · GitHub

0e0w · GitHub

阅读：453|2022-08-15

9 tgreyuk/typedoc-plugin-markdown: A plugi

tgreyuk/typedoc-plugin-markdown: A plugin for TypeDoc that enables TypeScript AP

阅读：856|2022-08-18

10 CVE-2022-1325

A flaw was found in Clmg, where with the help of a maliciously crafted pandore o

阅读：1545|2022-09-18

客服电话

电子邮件

PHP+RabbitMQ消息发布与订阅

概念

交换器(Exchanges)

交换类型

绑定(Bindings)

路由键

消息发布

消息订阅

模拟测试

请发表评论

全部评论

上一篇：

下一篇：

krishnaik06/Machine-Learning-in-90-days

armancodv/building-energy-model-matlab:

美元符号为什么是“$”

FGRibreau/import-tweets-to-mastodon: How

Delphi与DirectX之DelphiX(14):TPictureCol

CVE-2020-35163

bradtraversy/iweather: Ionic 3 mobile we

mysensors/MySensors: MySensors library a

joaomh/curso-de-matlab

断牙刷新位置时间（断牙属性及刷新位置介绍

rugk/mastodon-simplified-federation: Sim

关于我们

产品与服务

解决方案

139-2527-9053