RabbitMQ与PHP(一)—— RabbitMQ的原理与操作示例
RabbitMQ是流行的开源消息队列系统,用erlang语言开发,完整的实现了AMQP(高级消息队列协议)。网站在:http://www.rabbitmq.com/上面有教程和实例代码(Python和Java的)。
7af40ad162d9f2d36b6bf89fa8ec8a136327cc4c
AMPQ协议为了能够满足各种消息队列需求,在概念上比较复杂。首先,rabbitMQ启动默认是没有任何配置的,需要客户端连接上去,设置交换机等才能工作。不把这些基础概念弄清楚,后面程序设计就容易产生问题。
1.vhosts:虚拟主机。
一个RabbitMQ的实体上可以有多个vhosts,用户与权限设置就是依附于vhosts。对一般PHP应用,不需要用户权限设定,直接使用默认就存在的"/"就可以了,用户可以使用默认就存在的"guest"。一个简单的配置示例:
$conn_args=array(
'host'=>'127.0.0.1',
'port'=>'5672',
'login'=>'guest',
'password'=>'guest',
'vhost'=>'/'
);
2.connection与channel:连接与信道
connection是指物理的连接,一个client与一个server之间有一个连接;一个连接上可以建立多个channel,可以理解为逻辑上的连接。一般应用的情况下,有一个channel就够用了,不需要创建更多的channel。示例代码:
//创建连接和channel
$conn=newAMQPConnection($conn_args);
if(!$conn->connect()){
die("Cannotconnecttothebroker!\n");
}
$channel=newAMQPChannel($conn);
3.exchange与routingkey:交换机与路由键
为了将不同类型的消息进行区分,设置了交换机与路由两个概念。比如,将A类型的消息发送到名为‘C1’的交换机,将类型为B的发送到'C2'的交换机。当客户端连接C1处理队列消息时,取到的就只是A类型消息。进一步的,如果A类型消息也非常多,需要进一步细化区分,比如某个客户端只处理A类型消息中针对K用户的消息,routingkey就是来做这个用途的。
$e_name='e_linvo';//交换机名
$k_route=array(0=>'key_1',1=>'key_2');//路由key
//创建交换机
$ex=newAMQPExchange($channel);
$ex->setName($e_name);
$ex->setType(AMQP_EX_TYPE_DIRECT);//direct类型
$ex->setFlags(AMQP_DURABLE);//持久化
echo"ExchangeStatus:".$ex->declare()."\n";
for($i=0;$i<5;++$i){
echo"SendMessage:".$ex->publish($message.date('H:i:s'),$k_route[i%2])."\n";
}
由以上代码可以看到,发送消息时,只要有“交换机”就够了。至于交换机后面有没有对应的处理队列,发送方是不用管的。routingkey可以是空的字符串。在示例中,我使用了两个key交替发送消息,是为了下面更便于理解routingkey的作用。
对于交换机,有两个重要的概念:
A,类型。有三种类型:Fanout类型最简单,这种模型忽略routingkey;Direct类型是使用最多的,使用确定的routingkey。这种模型下,接收消息时绑定'key_1'则只接收key_1的消息;最后一种是Topic,这种模式与Direct类似,但是支持通配符进行匹配,比如:'key_*',就会接受key_1和key_2。Topic貌似美好,但是有可能导致不严谨,所以还是推荐使用Direct。
B,持久化。指定了持久化的交换机,在重新启动时才能重建,否则需要客户端重新声明生成才行。
需要特别明确的概念:交换机的持久化,并不等于消息的持久化。只有在持久化队列中的消息,才能持久化;如果没有队列,消息是没有地方存储的;消息本身在投递时也有一个持久化标志的,PHP中默认投递到持久化交换机就是持久的消息,不用特别指定。
4.queue:队列
讲了这么多,才讲到队列呀。事实上,队列仅是针对接收方(consumer)的,由接收方根据需求创建的。只有队列创建了,交换机才会将新接受到的消息送到队列中,交换机是不会在队列创建之前的消息放进来的。换句话说,在建立队列之前,发出的所有消息都被丢弃了。下面这个图比RabbitMQ官方的图更清楚——Queue是属于ReceiveMessage的一部分。
024f78f0f736afc37053e415b219ebc4b7451266
接下来看一下创建队列及接收消息的示例:
$e_name='e_linvo';//交换机名
$q_name='q_linvo';//队列名
$k_route='';//路由key
//创建连接和channel
$conn=newAMQPConnection($conn_args);
if(!$conn->connect()){
die("Cannotconnecttothebroker!\n");
}
$channel=newAMQPChannel($conn);
//创建交换机
$ex=newAMQPExchange($channel);
$ex->setName($e_name);
$ex->setType(AMQP_EX_TYPE_DIRECT);//direct类型
$ex->setFlags(AMQP_DURABLE);//持久化
echo"ExchangeStatus:".$ex->declare()."\n";
//创建队列
$q=newAMQPQueue($channel);
$q->setName($q_name);
$q->setFlags(AMQP_DURABLE);//持久化
//绑定交换机与队列,并指定路由键
echo'QueueBind:'.$q->bind($e_name,$k_route)."\n";
//阻塞模式接收消息
echo"Message:\n";
$q->consume('processMessage',AMQP_AUTOACK);//自动ACK应答
$conn->disconnect();
/**
*消费回调函数
*处理消息
*/
functionprocessMessage($envelope,$queue){
var_dump($envelope->getRoutingKey);
$msg=$envelope->getBody();
echo$msg."\n";//处理消息
}
从上述示例中可以看到,交换机既可以由消息发送端创建,也可以由消息消费者创建。
创建一个队列(line:20)后,需要将队列绑定到交换机上(line:25)队列才能工作,routingkey也是在这里指定的。有的资料上写成bindingkey,其实一回事儿,弄两个名词反倒容易混淆。
消息的处理,是有两种方式:
A,一次性。用$q->get([...]),不管取到取不到消息都会立即返回,一般情况下使用轮询处理消息队列就要用这种方式;
B,阻塞。用$q->consum(callback,[...])程序会进入持续侦听状态,每收到一个消息就会调用callback指定的函数一次,直到某个callback函数返回FALSE才结束。
关于callback,这里多说几句:PHP的call_back是支持使用数组的,比如:$c=newMyClass();$c->counter=100;$q->consume(array($c,'myfunc'))这样就可以调用自己写的处理类。MyClass中myfunc的参数定义,与上例中processMessage一样就行。
在上述示例中,使用的$routingkey='',意味着接收全部的消息。我们可以将其改为$routingkey='key_1',可以看到结果中仅有设置routingkey为key_1的内容了。
注意:routingkey='key_1'与routingkey='key_2'是两个不同的队列。假设:client1与client2都连接到key_1的队列上,一个消息被client1处理之后,就不会被client2处理。而routingkey=''是另类,client_all绑定到''上,将消息全都处理后,client1和client2上也就没有消息了。
在程序设计上,需要规划好exchange的名称,以及如何使用key区分开不同类型的标记,在消息产生的地方插入发送消息代码。后端处理,可以针对每一个key启动一个或多个client,以提高消息处理的实时性。如何使用PHP进行多线程的消息处理,将在下一节中讲述。
更多消息模型,可以参考:http://www.rabbitmq.com/tutorials/tutorial-two-python.html
b03533fa828ba61e15fc0e5f4034970a304e59b4
http://nonfu.me/p/8833.html