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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
| <?php
namespace app\common\library;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use think\Config;
use think\Log;
class Rabbitmq
{
/**
* 配置
*
* @var array
*/
public $config;
/**
* @var AMQPChannel
*/
public $channel;
/**
* initialized
*
* @var AMQPStreamConnection
*/
public $connect;
public function __construct(array $config = [])
{
$this->config = array_merge(Config::get('rabbitmq'), $config);
}
/**
* 处理消息格式 A Message for use with the Channnel.basic_* methods.
*
* @param $msg
* @param $message_durable
* @return AMQPMessage
*/
public function message($msg, $message_durable)
{
if (is_array($msg)) {
$msg = json_encode($msg);
}
if (!is_object($msg) && $message_durable === true) {
return new AMQPMessage($msg, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);
}
return new AMQPMessage($msg);
}
/**
* acknowledge one or more messages
*
* @param string $delivery_tag
*/
public function basicAck($delivery_tag)
{
$this->channel->basic_ack($delivery_tag);
}
/**
* 生产者 生产实时队列 只生产一个队列
*
* @param $data
* @param $queue
* @param $exchange_name
* @param string $routing_key
* @param string $type
* @param bool $queue_durable
* @param bool $message_durable
*/
public function send(
$data, $queue, $exchange_name, $routing_key = '', $type = 'direct', $queue_durable = false, $message_durable = false
) {
$this->connect = new AMQPStreamConnection(
$this->config['host'],
$this->config['port'],
$this->config['login'],
$this->config['password'],
$this->config['vhost']
);
$this->channel = $this->connect->channel();
//声明交换机
$this->channel->exchange_declare($exchange_name, $type, false, $message_durable, false, false, false, [], null);
//声明队列
$this->channel->queue_declare($queue, false, $queue_durable, false, false, false, [], null);
//绑定交换机与队列
$this->channel->queue_bind($queue, $exchange_name, $routing_key, false, [], null);
//处理压入消息格式
$msg = $this->message($data, $message_durable);
//压入消息到交换机
$this->channel->basic_publish($msg, $exchange_name, $routing_key, false, false, null);
//关闭
$this->channel->close(); // 关闭信道
$this->connect->close(); // 关闭链接
}
/**
* 延时生产者 生产延时队列 生产两个队列（1个延时死信队列，1个实时消费队列）
*
* @param array|string $data 消息内容
* @param string $delay_queue 延时队列
* @param string $delay_exchange_name 延时交换机
* @param string $delay_routing_key 延时队列routing_key
* @param string $delay_type 延时交换机类型
* @param int $delay_expire 延时队列声明周期（过期会触发死信规则）
* @param bool $delay_queue_durable 延时队列持久化
* @param bool $delay_message_durable 延时消息持久化
* @param string $receive_queue 接收队列（接收死信规则）
* @param string $receive_exchange_name 接收交换机
* @param string $receive_routing_key 接收队列routing_key
* @param string $receive_type 接收交换机类型
* @param bool $receive_queue_durable 接收队列持久化
* @param bool $receive_message_durable 接收消息持久化
*/
public function sendDelay(
$data,
$delay_queue,
$delay_exchange_name,
$delay_routing_key = '',
$delay_type = 'direct',
$delay_expire = 0,
$delay_queue_durable = false,
$delay_message_durable = false,
$receive_queue,
$receive_exchange_name,
$receive_routing_key = '',
$receive_type = 'direct',
$receive_queue_durable = false,
$receive_message_durable = false
) {
$this->connect = new AMQPStreamConnection(
$this->config['host'],
$this->config['port'],
$this->config['login'],
$this->config['password'],
$this->config['vhost']
);
$this->channel = $this->connect->channel();
// 声明主队列 <-- 关联主消费交换机(接收) <-- 数据压入
// | 300 /
// | /
// 关联延时交换机 -> 关联消费队列
//
// 延时交换机（弹出数据） -> 消费
//声明死信规则
$tale = new AMQPTable();
$tale->set('x-dead-letter-exchange', $receive_exchange_name);
$tale->set('x-dead-letter-routing-key', $receive_routing_key);
$tale->set('x-message-ttl', $delay_expire);
//声明延时交换机 与 接收交换机
$this->channel->exchange_declare($delay_exchange_name, $delay_type, false, $delay_message_durable, false, false, false, [], null);
$this->channel->exchange_declare($receive_exchange_name, $receive_type, false, $receive_message_durable, false, false, false, [], null);
//声明延时队列 与 接收队列
$this->channel->queue_declare($delay_queue, false, $delay_queue_durable, false, false, false, $tale, null);
$this->channel->queue_declare($receive_queue, false, $receive_queue_durable, false, false, false, [], null);
//绑定延时队列到延时交换机 与 绑定接收队列到接收交换机
$this->channel->queue_bind($delay_queue, $delay_exchange_name, $delay_routing_key, false, [], null);
$this->channel->queue_bind($receive_queue, $receive_exchange_name, $receive_routing_key, false, [], null);
//处理压入消息格式
$msg = $this->message($data, $delay_message_durable);
//压入消息到交换机
$this->channel->basic_publish($msg, $delay_exchange_name, $delay_routing_key, false, false, null);
//关闭
$this->channel->close(); // 关闭信道
$this->connect->close(); // 关闭链接
}
/**
* 消费者 只消费一个队列
*
* @param string $queue
* @param string $consumer_tag
* @param bool $no_local
* @param bool $no_ack
* @param bool $exclusive
* @param bool $nowait
* @param null $ticket
* @param array $arguments
*/
function receive(
$queue = '', // 队列名
$callback = null, // 回调函数
$queue_durable = false, //持久化
$consumer_tag = '',
$no_local = false,
$no_ack = false,
$exclusive = false, //队列是否可以被其他队列访问
$nowait = false,
$ticket = null,
$arguments = array()
) {
$this->connect = new AMQPStreamConnection(
$this->config['host'],
$this->config['port'],
$this->config['login'],
$this->config['password'],
$this->config['vhost']
);
$this->channel = $this->connect->channel();
//一次只消费一个
$this->channel->basic_qos(0,1,false);
//声明队列
$this->channel->queue_declare($queue, false, $queue_durable, false, false, false, [], null);
//订阅消费 callback仅绑定并不立即执行
$this->channel->basic_consume($queue, $consumer_tag, $no_local, $no_ack, $exclusive, $nowait, $callback,
$ticket, $arguments);
//轮训等待并触发basic_consume绑定的callback
while (count($this->channel->callbacks)) {
$this->channel->wait();
}
//关闭
$this->channel->close(); // 关闭信道
$this->connect->close(); // 关闭链接
}
// $callback = function ($msg) {
// $rabbit = new \app\admin\command\Rabbitmq();
// $rabbit->sendMessage($msg->body);
//// $recharge = new Recharge();
//// $recharge->sendMessage($msg->body);
// $this->basicAck($msg->delivery_info['delivery_tag']);
// };
}
|
