竹磬网-邵珠庆の日记 生命只有一次，你可以用它来做些更多伟大的事情–Make the world a little better and easier

PHP高级编程之消息队列

消息队列（英语：Message queue）是一种进程间通信或同一进程的不同线程间的通信方式

消息队列技术是分布式应用间交换信息的一种技术。消息队列可驻留在内存或磁盘上,队列存储消息直到它们被应用程序读出。通过消息队列，应用程序可独立地执行，它们不需要知道彼此的位置、或在继续执行前不需要等待接收程序接收此消息。

你首先需要弄清楚，消息队列与远程过程调用的区别，在很多读者咨询我的时候，我发现他们需要的是RPC(远程过程调用)，而不是消息队列。

消息队列有同步或异步实现方式，通常我们采用异步方式使用消息队列，远程过程调用多采用同步方式。

MQ与RPC有什么不同？ MQ通常传递无规则协议，这个协议由用户定义并且实现存储转发；而RPC通常是专用协议，调用过程返回结果。

同步需求，远程过程调用(PRC)更适合你。

异步需求，消息队列更适合你。

目前很多消息队列软件同时支持RPC功能，很多RPC系统也能异步调用。

通常的做法，如果小的项目团队可以有一个人实现，包括消息的推送，接收处理。如果大型团队，通常是定义好消息协议，然后各自开发各自的部分，例如一个团队负责写推送协议部分，另一个团队负责写接收与处理部分。

那么为什么我们不讲消息队列框架化呢？

下面是作者开发的一个SOA框架，该框架提供了三种接口，分别是SOAP,RESTful,AMQP(RabbitMQ)，理解了该框架思想，你很容易进一步扩展，例如增加XML-RPC, ZeroMQ等等支持。

https://github.com/netkiller/SOA

本文只讲消息队列框架部分。

消息队列框架是本地应用程序（命令行程序），我们为了让他在后台运行，需要实现守护进程。

每个实例处理一组队列，实例化需要提供三个参数，$queueName = '队列名', $exchangeName = '交换名', $routeKey = '路由'

$daemon = new \framework\RabbitDaemon($queueName = 'email', $exchangeName = 'email', $routeKey = 'email');
			

守护进程需要使用root用户运行，运行后会切换到普通用户，同时创建进程ID文件，以便进程停止的时候使用。

守护进程核心代码https://github.com/netkiller/SOA/blob/master/system/rabbitdaemon.class.php

消息协议是一个数组，将数组序列化或者转为JSON推送到消息队列服务器，这里使用json格式的协议。

$msg = array(
	'Namespace'=>'namespace',
	"Class"=>"Email",
	"Method"=>"smtp",
	"Param" => array(
		$mail, $subject, $message, null
	)
);			
			

序列化后的协议

{"Namespace":"single","Class":"Email","Method":"smtp","Param":["netkiller@msn.com","Hello"," TestHelloWorld",null]}			
			

使用json格式是考虑到通用性，这样推送端可以使用任何语言。如果不考虑兼容，建议使用二进制序列化，例如msgpack效率更好。

消息队列处理核心代码

https://github.com/netkiller/SOA/blob/master/system/rabbitmq.class.php

所以消息的处理在下面一段代码中进行

$this->queue->consume(function($envelope, $queue) {

	$speed = microtime(true);
	
	$msg = $envelope->getBody();
	$result = $this->loader($msg);
	$queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答

	//$this->logging->info(''.$msg.' '.$result)
	$this->logging->debug('Protocol: '.$msg.' ');
	$this->logging->debug('Result: '. $result.' ');
	$this->logging->debug('Time: '. (microtime(true) - $speed) .'');
});
			

public function loader($msg = null) 负责拆解协议，然后载入对应的类文件，传递参数，运行方法，反馈结果。

Time 可以输出程序运行所花费的时间，对于后期优化十分有用。

测试代码 https://github.com/netkiller/SOA/blob/master/test/queue/email.php