转: RabbitMQ与PHP（一）

RabbitMQ与PHP（一）

项目中使用RabbitMQ做为队列处理用户消息通知，消息由前端PHP代码产生，处理消息使用Python，这就致使代码一致性问题，调整消息定义时须要PHP和Python都进行修改。这两天抽时间研究了下，如何将消息的产生与处理（消费）所有用PHP来作。查资料时发现，关于PHP处理消息队列的资料不多，有必要把一些初学者容易混淆的地方再讲一下。

拟分红两部分： 一，RabbitMQ的原理与操做示例；二，具体服务安装及如何用PHP做为守护模式处理消息。

RabbitMQ是流行的开源消息队列系统，用erlang语言开发，完整的实现了AMPQ（高级消息队列协议）。网站在： http://www.rabbitmq.com/ 上面有教程和实例代码（Python和Java的）。

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 = new AMQPConnection($conn_args);   
if (!$conn->connect()) {   
    die("Cannot connect to the broker!\n");   
}   
$channel = new AMQPChannel($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 = new AMQPExchange($channel);   
$ex->setName($e_name); 
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型  
$ex->setFlags(AMQP_DURABLE); //持久化 
echo "Exchange Status:".$ex->declare()."\n";   
for($i=0; $i<5; ++$i){ 
    echo "Send Message:".$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的一部分。

接下来看一下建立队列及接收消息的示例：

$e_name = 'e_linvo'; //交换机名 
$q_name = 'q_linvo'; //队列名 
$k_route = ''; //路由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, $k_route)."\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"; //处理消息 
}

从上述示例中能够看到，交换机既能够由消息发送端建立，也能够由消息消费者建立。

建立一个队列(line:20)后，须要将队列绑定到交换机上(line:25)队列才能工做，routingkey也是在这里指定的。有的资料上写成bindingkey，其实一回事儿，弄两个名词反倒容易混淆。

消息的处理，是有两种方式：

A，一次性。用 $q->get([...])，无论取到取不到消息都会当即返回，通常状况下使用轮询处理消息队列就要用这种方式；

B，阻塞。用 $q->consum( callback, [...] ) 程序会进入持续侦听状态，每收到一个消息就会调用callback指定的函数一次，直到某个callback函数返回FALSE才结束。

关于callback，这里多说几句： PHP的call_back是支持使用数组的，好比： $c = new MyClass(); $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