php使用RabbitMQ详解_php rabbitmq文档

0，rabbitmq安装，参考： https://blog.csdn.net/gaokcl/article/details/82804519

RabbitMQ的四种ExChange https://blog.csdn.net/dandanzmc/article/details/52262850

信道、交换器和路由键、队列概念 参考：https://www.cnblogs.com/zhangxue/p/7699698.html

交换机： http://rabbitmq.mr-ping.com/AMQP/AMQP_0-9-1_Model_Explained.html

使用php-amqplib连接rabbitMQ 学习笔记及总结 https://www.cnblogs.com/oyxp/p/7376733.html

RabbitMQ基础知识 https://www.cnblogs.com/dwlsxj/p/RabbitMQ.html

1， .env文件，方便管理配置：

# mysql config

DB_HOST     = 127.0.0.1
DB_PORT     = 3306
DB_USER     = "root"
DB_PASS     = ""
DB_NAME     = ""
DB_PREFIX   = ""
DB_CHARSET  = "utf8"

# rabbitMQ config

RABBIT_HOST = 127.0.0.1
RABBIT_PORT = 5672
RABBIT_USER = "guest"    #可以修改
RABBIT_PASS = "guest"    #可以修改
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

2，rabbitmq端口号说明

默认端口说明

- client端通信端口:  5672

-  管理端口: 15672     # 打开RabbitMQ的管理中心 http://127.0.0.1:15672/ 

-  server间内部通信端口: 25672

-  erlang发现端口: 4369
1
2
3
4
5
6
7
8
9

一，方法一（偏原生）：

主要分为两个部分：

第一：发送者（publisher）

第二：消费者（consumer）

---

（一）生产者 (创建一个rabbit_publisher.php的文件)

创建连接–>创建channel–>创建交换机对象–>发送消息

$conn_args = array(
    'host' => '127.0.0.1',
    'port' => '5672',
    'login' => 'guest',
    'password' => 'guest',
    'vhost'=>'/'
);

//创建连接和channel
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
    die("Cannot connect to the broker!\n");
}
$channel = new AMQPChannel($conn);

//创建交换机
$e_name = 'e_linvo'; //交换机名
$ex = new AMQPExchange($channel);
$ex->setName($e_name);
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型
$ex->setFlags(AMQP_DURABLE); //持久化
echo "Exchange Status:".$ex->declare()."\n";


echo "Send Message:".$ex->publish("TEST MESSAGE，key_1 by xust" . date('H:i:s', time()), 'key_1')."\n";
echo "Send Message:".$ex->publish("TEST MESSAGE，key_2 by xust" . date('H:i:s', time()), 'key_2')."\n";
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

（二）消费者(创建一个rabbit_consumer.php的文件)

创建连接–>创建channel–>创建交换机–>创建队列–>绑定交换机/队列/路由键–>接收消息

$conn_args = array(
    'host' => '127.0.0.1',
    'port' => '5672',
    'login' => 'guest',
    'password' => 'guest',
    'vhost'=>'/'
);

$e_name = 'e_linvo'; //交换机名
$q_name = 'q_linvo'; //队列名
$k_route = 'key_2'; //路由key

//创建连接和channel
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
    die("Cannot connect to the broker!\n");
}
$channel = new AMQPChannel($conn);

//创建交换机
$ex = new AMQPExchange($channel);
$ex->setName($e_name);
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型
$ex->setFlags(AMQP_DURABLE); //持久化
echo "Exchange Status:".$ex->declare()."\n";

//创建队列
$q = new AMQPQueue($channel);
$q->setName($q_name);
$q->setFlags(AMQP_DURABLE); //持久化

//绑定交换机与队列，并指定路由键
echo 'Queue Bind: '.$q->bind($e_name, 'key_2')."\n";　

//阻塞模式接收消息
echo "Message:\n";
$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答

$conn->disconnect();

/**
 * 消费回调函数
 * 处理消息
 */
function processMessage($envelope, $queue) {
    var_dump($envelope->getRoutingKey);
    $msg = $envelope->getBody();
    echo $msg."\n"; //处理消息
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

二，方法二（使用composer安装php-amqplib）：

1、使用composer安装php-amqplib

在你的项目中添加一个 composer.json文件：
  {
     "require": {
      "php-amqplib/php-amqplib": "2.6.*"
     }
 }
只要你已经安装Composer功能，你可以运行以下：
   $ composer install
已经存在的项目则执行 
   $ composer update
这时在verdor目录就已经下载完毕

具体可以参考官方文档：https://github.com/php-amqplib/php-amqplib
1
2
3
4
5
6
7
8
9
10
11
12
13

2、使用php-amqplib

英文文档：http://www.rabbitmq.com/getstarted.html
中文文档：https://rabbitmq.shujuwajue.com/tutorials_with_php/[2]Work_Queues.md.html

总结：官方文档看起来有点乱，总结了一下，基本都是互通的，以下为简单基本步骤，仅供参考，NO BB。

生产者：
1、创建连接

主要参数说明：
$host:  RabbitMQ服务器主机IP地址
$port:  RabbitMQ服务器端口
$user:  连接RabbitMQ服务器的用户名
$password:  连接RabbitMQ服务器的用户密码
$vhost:   连接RabbitMQ服务器的vhost（服务器可以有多个vhost，虚拟主机，类似nginx的vhost）

$connection =  new AMQPStreamConnection($host,$port,$user,$password,$vhost);

 
2、获取信道  
// $channel_id 信道id，不传则获取$channel[“”]信道，再无则循环$this->channle数组，下标从1到最大信道数找第一个不是AMQPChannel对象的下标，实例化并返回AMQPChannel对象，无则抛出异常No free channel ids

$channel = $connection->channel($channel_id);


3、在信道里创建交换器

 # $exhcange_name 交换器名字
 # $type 交换器类型：
’’  默认交换机 匿名交换器 未显示声明类型都是该类型
fanout  扇形交换器 会发送消息到它所知道的所有队列，每个消费者获取的消息都是一致的
headers 头部交换器
direct 直连交换器，该交换机将会对绑定键（binding key）和路由键（routing key）进行精确匹配
topic 话题交换器 该交换机会对路由键正则匹配，必须是*(一个单词)、#(多个单词，以.分割) 、      user.key .abc.* 类型的key

rpc 
#$passive  false
#durable false
#auto_detlete false 

$channel->exchange_declare($exhcange_name,$type,$passive,$durable,$auto_delete);
//常用设置 $passive=>false  $durable=>false $auto_delete->false


4、创建要发送的信息 ，可以创建多个消息
   #$data  string类型 要发送的消息
   #$properties array类型 设置的属性，比如设置该消息持久化[‘delivery_mode’=>2]

$msg = new AMQPMessage($data,$properties)

5、发送消息

#$msg object AMQPMessage对象
#$exchange string 交换机名字  
#$routing_key string 路由键 如果交换机类型
  fanout： 该值会被忽略，因为该类型的交换机会把所有它知道的队列发消息，无差别区别
direct  只有精确匹配该路由键的队列，才会发送消息到该队列
topic   只有正则匹配到的路由键的队列，才会发送到该队列
$channel->basic_publish($msg,$exchange,$routing_key);

6、关闭信道和链接

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









消费者：
1、创建连接

主要参数说明：
$host:  RabbitMQ服务器主机IP地址
$port:  RabbitMQ服务器端口
$user:  连接RabbitMQ服务器的用户名
$password:  连接RabbitMQ服务器的用户密码
$vhost:   连接RabbitMQ服务器的vhost（服务器可以有多个vhost，虚拟主机，类似nginx的vhost）

$connection =  new AMQPStreamConnection($host,$port,$user,$password,$vhost);

 
2、获取信道  
// $channel_id 信道id，不传则获取$channel[“”]信道，再无则循环$this->channle数组，下标从1到最大信道数找第一个不是AMQPChannel对象的下标，实例化并返回AMQPChannel对象，无则抛出异常No free channel ids

$channel = $connection->channel($channel_id);



3、在信道里创建交换器，未显式声明交换机都是使用匿名交换机

 # $exhcange_name 交换器名字
 # $type 交换器类型：
’’  默认交换机 匿名交换器 未显示声明类型都是该类型
fanout  扇形交换器 会发送消息到它所知道的所有队列，每个消费者获取的消息都是一致的
headers 头部交换器
direct 直连交换器，该交换机将会对绑定键（binding key）和路由键（routing key）进行精确匹配
topic 话题交换器 该交换机会对路由键正则匹配，必须是*(一个单词)、#(多个单词，以.分割) 、      user.key .abc.* 类型的key

rpc 
#$passive  false
#durable false
#auto_detlete false 

$channel->exchange_declare($exhcange_name,$type,$passive,$durable,$auto_delete);
//常用设置 $passive=>false  $durable=>false $auto_delete->false




4、声明消费者队列

（1）    非持久化队列,RabbitMQ退出或者崩溃时，该队列就不存在

  list($queue_name, ,) = $channel->queue_declare("", false, false, false, false)
   
（2）    持久化队列（需要显示声明，第三个参数要设置为true），保存到磁盘，但不一定完全保证不丢失信息，因为保存总是要有时间的。

 list($queue_name, ,) = $channel->queue_declare("ex_queue", false, false, true, false)



5、绑定交换机，当未显示绑定交换机时，默认是绑定匿名交换机

   #绑定：交换机与队列的关系，如下面这句代码意思是，$queue_name队列对logs交换机数据感兴趣，该队列就消费该交换机传过来的数据：这个队列（queue）对这个交换机（exchange）的消息感兴趣. $binding_key默认为空，表示对该交换机所有消息感兴趣，如果值不为空，则该队列只对该类型的消息感兴趣（除了fanout交换机以外）
   $channel->queue_bind($queue_name, 'logs', $binding_key);


6、消费消息

#该代码表示使用basic.qos方法，并设置prefetch_count=1。这样是告诉RabbitMQ，再同一时刻，不要发送超过1条消息给一个工作者（worker），直到它已经处理了上一条消息并且作出了响应。这样，RabbitMQ就会把消息分发给下一个空闲的工作者（worker），轮询、负载均衡配置

#$channel->basic_qos(null, 1, null);


#第四个参数 no_ack = false 时，表示进行ack应答，确保消息已经处理
#$callback 表示回调函数，传入消息参数
$channel->basic_consume('ex_queue', '', false, false, false, false, $callback);

$callback = function($msg){
  echo " [x] Received ", $msg->body, "\n";
  sleep(substr_count($msg->body, '.'));
  echo " [x] Done", "\n";

#当no_ack=false时， 需要写下行代码，否则可能出现内存不足情况#$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};

#监听消息，一有消息，立马就处理
while(count($channel->callbacks)) {
    $channel->wait();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154

– end

---

下面是对消费者的详细说明（我自己备注，可以不看）：

<?php
defined('DIR_ROOT') OR die('不能直接执行该文件' . PHP_EOL);

use PhpAmqpLib\Connection\AMQPStreamConnection;

// use PhpAmqpLib\Message\AMQPMessage;

// 1,消费者，创建连接
$connection = new AMQPStreamConnection(
    getenv('RABBIT_HOST'), //  .env文件
    getenv('RABBIT_PORT'), //  .env文件
    getenv('RABBIT_USER'), //  .env文件
    getenv('RABBIT_PASS')  //  .env文件
);

/**
 * 2,获取信道
 * $channel_id ==> 信道id，不传则获取$channel[""]信道，每个信道相互独立
 * 再无 => 则循环$this->channle数组，下标从1到最大信道数找第一个不是AMQPChannel对象的下标，实例化并返回AMQPChannel对象，
 * 无则抛出异常No free channel ids
 */
$channel = $connection->channel();

// 3，声明交换机
// 在信道里创建交换机，未显式声明交换机都是使用匿名交换机
$channel->exchange_declare(
    $exchange_name, // 交换机名字
    'topic', // 交换机类型，主题交换机，集合对应关系：N：1
    false,
    true, // 交换机将在服务器重启后继续存在
    false // 通道关闭的时候，交换机不会被删除
);
// 4，声明消费者队列
$channel->queue_declare(
    $queue_name, // 队列名称
    false,
    true, // 队列是否持久化， 持久化队列为true
    false, // 当前连接不在时，队列是否自动删除
    false // 没有consumer时，队列是否自动删除
);

/**
 * 5，绑定队列到交换机，当未显示绑定交换机时，默认是绑定匿名交换机
 * 绑定：交换机与队列的关系，如下面这句代码意思是，$queue_name队列对$exchange_name交换机数据感兴趣，
 * 该队列就消费该交换机传过来的数据：这个队列（queue）对这个交换机（exchange）的消息感兴趣. $routing_key默认为空，表示对该交换机所有消息感兴趣，
 * 如果值不为空，则该队列只对该类型的消息感兴趣（除了fanout交换机以外）
 */
$channel->queue_bind(
    $queue_name, // 队列名
    $exchange_name, // 交换机名
    $routing_key // 路由key
);

/**
 * 6,消费消息
 * 该代码表示使用basic.qos方法，并设置prefetch_count=1。
 * 这样是告诉RabbitMQ，再同一时刻，不要发送超过1条消息给一个工作者（worker），直到它已经处理了上一条消息并且作出了响应。
 * 这样，RabbitMQ就会把消息分发给下一个空闲的工作者（worker），轮询、负载均衡配置
 */
$channel->basic_qos(null, 1, null);
// 当no_ack=false时， 需要写下行代码，否则可能出现内存不足情况
// $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
$channel->basic_consume(
    $queue_name,
    '',
    false,
    false, // 是否不需要消费者发送 ack,表示进行ack应答，确保消息已经处理
    false,
    false,
    'processMessage' //表示回调函数，传入消息参数
);

echo '[start] [', date('Y-m-d H:i:s'), '] queue:', $queue_name, PHP_EOL;

//监听消息，一有消息，立马就处理
while (count($channel->callbacks)) {
    $channel->wait();
}

$channel->close();
$connection->close();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81