amqp协议与php下的rabbitMQ
2013-09-25 14:36
309 查看
【AMQP协议】转自博客园知识库
链接:http://kb.cnblogs.com/page/73759/
当前各种应用大量使用异步消息模型,并随之产生众多消息中间件产品及协议,标准的不一致使应用与中间件之间的耦合限制产品的选择,并增加维护成本。AMQP是一个提供统一消息服务的应用层标准协议,基于此协议的客户端与消息中间件可传递消息,并不受客户端/中间件不同产品,不同开发语言等条件的限制。
当然这种降低耦合的机制是基于与上层产品,语言无关的协议。AMQP协议是一种二进制协议,提供客户端应用与消息中间件之间异步、安全、高效地交互。从整体来看,AMQP协议可划分为三层:
这种分层架构类似于OSI网络协议,可替换各层实现而不影响与其它层的交互。AMQP定义了合适的服务器端域模型,用于规范服务器的行为(AMQP服务器端可称为broker)。在这里Model层决定这些基本域模型所产生的行为,这种行为在AMQP中用”command”表示,在后文中会着重来分析这些域模型。Session层定义客户端与broker之间的通信(通信双方都是一个peer,可互称做partner),为command的可靠传输提供保障。Transport层专注于数据传送,并与Session保持交互,接受上层的数据,组装成二进制流,传送到receiver后再解析数据,交付给Session层。Session层需要Transport层完成网络异常情况的汇报,顺序传送command等工作。
上面是对AMQP协议的大致说明。下面会以我们对消息服务的需求来理解AMQP所提供的域模型。消息中间件的主要功能是消息的路由(Routing)和缓存(Buffering)。在AMQP中提供类似功能的两种域模型:Exchange 和 Message queue。
Exchange接收消息生产者(Message Producer)发送的消息根据不同的路由算法将消息发送往Message queue。Message queue会在消息不能被正常消费时缓存这些消息,具体的缓存策略由实现者决定,当message queue与消息消费者(Message consumer)之间的连接通畅时,Message queue有将消息转发到consumer的责任。
Message是当前模型中所操纵的基本单位,它由Producer产生,经过Broker被Consumer所消费。它的基本结构有两部分: Header和Body。Header是由Producer添加上的各种属性的集合,这些属性有控制Message是否可被缓存,接收的queue是哪个,优先级是多少等。Body是真正需要传送的数据,它是对Broker不可见的二进制数据流,在传输过程中不应该受到影响。
一个broker中会存在多个Message queue,Exchange怎样知道它要把消息发送到哪个Message queue中去呢? 这就是上图中所展示Binding的作用。Message queue的创建是由client application控制的,在创建Message queue后需要确定它来接收并保存哪个Exchange路由的结果。Binding是用来关联Exchange与Message queue的域模型。Client application控制Exchange与某个特定Message
queue关联,并将这个queue接受哪种消息的条件绑定到Exchange,这个条件也叫Binding key或是 Criteria。
在与多个Message queue关联后,Exchange中就会存在一个路由表,这个表中存储着每个Message queue所需要消息的限制条件。Exchange就会检查它接受到的每个Message的Header及Body信息,来决定将Message路由到哪个queue中去。Message的Header中应该有个属性叫Routing Key,它由Message发送者产生,提供给Exchange路由这条Message的标准。Exchange根据不同路由算法有不同有Exchange Type。比如有Direct类似,需要Binding
key等于Routing key;也有Binding key与Routing key符合一个模式关系;也有根据Message包含的某些属性来判断。一些基础的路由算法由AMQP所提供,client application也可以自定义各种自己的扩展路由算法。那么一个Message的处理流程类似于这样:
在这里有个新名词需要介绍: Virtual Host。一个Virtual Host可持有一些Exchange和Message queue。它是一个虚拟概念,一个Virtual Host可以是一台服务器,也可以是由多台服务器组成的集群。同步扩展下,Exchange与Message queue的部署也可以是一台或是多台服务器上。
Message的产生者和消费者可能是同一个应用。整个AMQP定义的就是Client application与Broker之间的交互。在粗略介绍完AMQP的域模型后,可以关注下Client是怎样与Broker建立起连接的。
在AMQP中,Client application想要与Broker沟通,就需要建立起与Broker的connection,这种connection其实是与Virtual Host相关联的,也就是说,connection是建立在client与Virtual Host之间。可以在一个connection上并发运行多个channel,每个channel执行与Broker的通信,我们前面提供的session就是依附于channel上的。
这里的Session可以有多种定义,既可以表示AMQP内部提供的command分发机制,也可以说是在宏观上区别与域模型的接口。正常理解就是我们平时所说的交互context,主要作用就是在网络上可靠地传递每一个command。在AMQP的设计中,应当是借鉴了TCP的各种设计,用于保证这种可靠性。
在Session层,为上层所需要交互的每个command分配一个惟一标识符(可以是一个UUID),是为了在传输过程中可以对command做校验和重传。Command发送端也需要记录每个发送出去的command到Replay Buffer,以期得到接收方的回馈,保证这个command被接收方明确地接收或是已执行这个command。对于超时没有收到反馈的command,发送方再次重传。如果接收方已明确地回馈信息想要告知command发送方但这条信息在中途丢失或是其它问题发送方没有收到,那么发送方不断重传会对接收方产生影响,为了降低这种影响,command接收方设置一个过滤器Idempotency
Barrier,来拦截那些已接收过的command。 关于这种重传及确认机制,可以参考下TCP的相关设计。
上面大致介绍了AMQP的域模型及连接机制中的确认及重传模型,不涉及AMQP的详细二进制规范。
http://hi.baidu.com/cnjimmydong/item/4955163003b493a3124b14de
项目中使用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"。一个简单的配置示例:
2,connection 与 channel : 连接与信道
connection是指物理的连接,一个client与一个server之间有一个连接;一个连接上可以建立多个channel,可以理解为逻辑上的连接。一般应用的情况下,有一个channel就够用了,不需要创建更多的channel。示例代码:
3,exchange 与 routingkey : 交换机 与 路由键
为了将不同类型的消息进行区分,设置了交换机与路由两个概念。比如,将A类型的消息发送到名为‘C1’的交换机,将类型为B的发送到'C2'的交换机。当客户端连接C1处理队列消息时,取到的就只是A类型消息。进一步的,如果A类型消息也非常多,需要进一步细化区分,比如某个客户端只处理A类型消息中针对K用户的消息,routingkey就是来做这个用途的。
由以上代码可以看到,发送消息时,只要有“交换机”就够了。至于交换机后面有没有对应的处理队列,发送方是不用管的。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的一部分。
接下来看一下创建队列及接收消息的示例:
从上述示例中可以看到,交换机既可以由消息发送端创建,也可以由消息消费者创建。
创建一个队列(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
http://hi.baidu.com/cnjimmydong/item/b9190a841f3406db99255fe6
在上一节中,详细介绍了RabbitMQ的exchange/routingkey/queue等概念,以及示例了如何使用PHP发送和处理消息的代码。这一节,将介绍在项目中如何使用PHP多线程的进行消息实时处理,以及简要介绍一些RabbitMQ的安装相关。熟悉的可以将安装这部分跳过。
一, RabbitMQ的安装:
需要首先安装erlong
注意: 这里将erlong安装到了指定的目录: /usr/local/erlong,而不是使用默认的路径。这是一个好的习惯,对于版本控制等都会有好处。但是这会导致后面 rabbitMQ报错:找不到erl 执行文件,需要多做一些处理才行。
安装RabbitMQ
到这里会出现一个报错信息: ./rabbitmq-server: line 86: erl: command not found 这是因为erlong指定了安装路径,在系统的PATH中找不到。只要export PATH=$PATH:/usr/local/erlong/bin 就可以了。
如果为了rc.local启动方便,可以将 export PATH=$PATH:/usr/local/erlong/bin 这一行写入到 rabbitmq-server 文件中:
执行后,ps -aux 一下,看到进程中有/usr/local/erlong/lib/erlang/erts-5.10.1/bin/epmd -daemon 和 /usr/local/erlong/lib/erlang/erts-5.10.1/bin/beam.smp 就OK了。
sbin目录下还有一个脚本: rabbitmqctl 也很常用,与 rabbitmq-server 一样需要指明erlong的路径才能正确工作。
常用的方法:
rabbitmqctl start-app 启动后,执行一下这个比较保险
rabbitmqctl list_exchanges 显示当前所有的交换机
rabbitmqctl list_queue 查看当前有效队列情况
二, PHP extension的安装:
PHP操作rabbitmq需要AMQP扩展的支持。下载扩展: http://pecl.php.net/package/amqp ,安装过程与一般扩展一样,/usr/local/php/bin/phpize; ./configure --with-php-config=/usr/local/php/bin/php-config; make && make install
然后编辑php.ini 插入:
[amqp]
extension = amqp.so
重启apache或nginx,查看phpinfo其中有关于anqp的段落,就OK了。
三,如何使用PHP进行实时后端消息处理
首先,要保证PHP文件可以正确的以堵塞方式处理消息。代码可参见上一节RabbitMQ与PHP(一)。
然后,我们来借助Python实现一个多线程的守护进程,由这个守护进程来调用PHP,把PHP作为工作线程。
启动脚本:start_jobs.py
线程脚本: MyThread.py
任务脚本: MyJobs.py
在任务调度(start_jobs.py)中,设计了两种工作模式:
一种工作模式是一共启动N个线程去干活,适合于尽快完成一个大任务;
另一种是保持进程数量,当发现某个进程完成后,再重新将进程启动起来。显然,用户守护处理消息适合这种模式。
具体工作在MyJob.py中,提供了系统Shell调用和采用URL调用两种方式。推荐使用shell直接调用php的方式,这样可以灵活控制Php.ini,比如增加auto_prepend_file、增长max_execution_time等。
实际项目中,假定有5种类型的消息,可以启动20个线程,将thread_id当作参数传递给PHP。PHP将thread_id%5当作待处理类型,就可以得到每种类型有4个线程工作的场景了。
考虑到PHP的执行时间限制及内存泄露问题,可以将consume.php脚本进行一下改进,让PHP脚本每次处理指定数量的消息后就退出,由Python多线程框架重新启动线程,以保证运行稳定可靠。另外,将应答改为手工应答,确保消息获得正确有效处理。
用 两个线程,检查间隔2秒,SHELL模式 测试运行结果:
由上图可见,运行开始后,检查2个线程都处于活跃状态,并对消息进行了正确处理,当处理到一定数量后,PHP程序结束,父进程检查到有进程处于完成状态,重新将其启动(第二个绿色框)。完全与预期相符。
链接:http://kb.cnblogs.com/page/73759/
当前各种应用大量使用异步消息模型,并随之产生众多消息中间件产品及协议,标准的不一致使应用与中间件之间的耦合限制产品的选择,并增加维护成本。AMQP是一个提供统一消息服务的应用层标准协议,基于此协议的客户端与消息中间件可传递消息,并不受客户端/中间件不同产品,不同开发语言等条件的限制。
当然这种降低耦合的机制是基于与上层产品,语言无关的协议。AMQP协议是一种二进制协议,提供客户端应用与消息中间件之间异步、安全、高效地交互。从整体来看,AMQP协议可划分为三层:
这种分层架构类似于OSI网络协议,可替换各层实现而不影响与其它层的交互。AMQP定义了合适的服务器端域模型,用于规范服务器的行为(AMQP服务器端可称为broker)。在这里Model层决定这些基本域模型所产生的行为,这种行为在AMQP中用”command”表示,在后文中会着重来分析这些域模型。Session层定义客户端与broker之间的通信(通信双方都是一个peer,可互称做partner),为command的可靠传输提供保障。Transport层专注于数据传送,并与Session保持交互,接受上层的数据,组装成二进制流,传送到receiver后再解析数据,交付给Session层。Session层需要Transport层完成网络异常情况的汇报,顺序传送command等工作。
上面是对AMQP协议的大致说明。下面会以我们对消息服务的需求来理解AMQP所提供的域模型。消息中间件的主要功能是消息的路由(Routing)和缓存(Buffering)。在AMQP中提供类似功能的两种域模型:Exchange 和 Message queue。
Exchange接收消息生产者(Message Producer)发送的消息根据不同的路由算法将消息发送往Message queue。Message queue会在消息不能被正常消费时缓存这些消息,具体的缓存策略由实现者决定,当message queue与消息消费者(Message consumer)之间的连接通畅时,Message queue有将消息转发到consumer的责任。
Message是当前模型中所操纵的基本单位,它由Producer产生,经过Broker被Consumer所消费。它的基本结构有两部分: Header和Body。Header是由Producer添加上的各种属性的集合,这些属性有控制Message是否可被缓存,接收的queue是哪个,优先级是多少等。Body是真正需要传送的数据,它是对Broker不可见的二进制数据流,在传输过程中不应该受到影响。
一个broker中会存在多个Message queue,Exchange怎样知道它要把消息发送到哪个Message queue中去呢? 这就是上图中所展示Binding的作用。Message queue的创建是由client application控制的,在创建Message queue后需要确定它来接收并保存哪个Exchange路由的结果。Binding是用来关联Exchange与Message queue的域模型。Client application控制Exchange与某个特定Message
queue关联,并将这个queue接受哪种消息的条件绑定到Exchange,这个条件也叫Binding key或是 Criteria。
在与多个Message queue关联后,Exchange中就会存在一个路由表,这个表中存储着每个Message queue所需要消息的限制条件。Exchange就会检查它接受到的每个Message的Header及Body信息,来决定将Message路由到哪个queue中去。Message的Header中应该有个属性叫Routing Key,它由Message发送者产生,提供给Exchange路由这条Message的标准。Exchange根据不同路由算法有不同有Exchange Type。比如有Direct类似,需要Binding
key等于Routing key;也有Binding key与Routing key符合一个模式关系;也有根据Message包含的某些属性来判断。一些基础的路由算法由AMQP所提供,client application也可以自定义各种自己的扩展路由算法。那么一个Message的处理流程类似于这样:
在这里有个新名词需要介绍: Virtual Host。一个Virtual Host可持有一些Exchange和Message queue。它是一个虚拟概念,一个Virtual Host可以是一台服务器,也可以是由多台服务器组成的集群。同步扩展下,Exchange与Message queue的部署也可以是一台或是多台服务器上。
Message的产生者和消费者可能是同一个应用。整个AMQP定义的就是Client application与Broker之间的交互。在粗略介绍完AMQP的域模型后,可以关注下Client是怎样与Broker建立起连接的。
在AMQP中,Client application想要与Broker沟通,就需要建立起与Broker的connection,这种connection其实是与Virtual Host相关联的,也就是说,connection是建立在client与Virtual Host之间。可以在一个connection上并发运行多个channel,每个channel执行与Broker的通信,我们前面提供的session就是依附于channel上的。
这里的Session可以有多种定义,既可以表示AMQP内部提供的command分发机制,也可以说是在宏观上区别与域模型的接口。正常理解就是我们平时所说的交互context,主要作用就是在网络上可靠地传递每一个command。在AMQP的设计中,应当是借鉴了TCP的各种设计,用于保证这种可靠性。
在Session层,为上层所需要交互的每个command分配一个惟一标识符(可以是一个UUID),是为了在传输过程中可以对command做校验和重传。Command发送端也需要记录每个发送出去的command到Replay Buffer,以期得到接收方的回馈,保证这个command被接收方明确地接收或是已执行这个command。对于超时没有收到反馈的command,发送方再次重传。如果接收方已明确地回馈信息想要告知command发送方但这条信息在中途丢失或是其它问题发送方没有收到,那么发送方不断重传会对接收方产生影响,为了降低这种影响,command接收方设置一个过滤器Idempotency
Barrier,来拦截那些已接收过的command。 关于这种重传及确认机制,可以参考下TCP的相关设计。
上面大致介绍了AMQP的域模型及连接机制中的确认及重传模型,不涉及AMQP的详细二进制规范。
RabbitMQ与PHP(一)
http://hi.baidu.com/cnjimmydong/item/4955163003b493a3124b14de项目中使用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"。一个简单的配置示例:
connection是指物理的连接,一个client与一个server之间有一个连接;一个连接上可以建立多个channel,可以理解为逻辑上的连接。一般应用的情况下,有一个channel就够用了,不需要创建更多的channel。示例代码:
为了将不同类型的消息进行区分,设置了交换机与路由两个概念。比如,将A类型的消息发送到名为‘C1’的交换机,将类型为B的发送到'C2'的交换机。当客户端连接C1处理队列消息时,取到的就只是A类型消息。进一步的,如果A类型消息也非常多,需要进一步细化区分,比如某个客户端只处理A类型消息中针对K用户的消息,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的一部分。
接下来看一下创建队列及接收消息的示例:
创建一个队列(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
RabbitMQ与PHP(二)
http://hi.baidu.com/cnjimmydong/item/b9190a841f3406db99255fe6在上一节中,详细介绍了RabbitMQ的exchange/routingkey/queue等概念,以及示例了如何使用PHP发送和处理消息的代码。这一节,将介绍在项目中如何使用PHP多线程的进行消息实时处理,以及简要介绍一些RabbitMQ的安装相关。熟悉的可以将安装这部分跳过。
一, RabbitMQ的安装:
需要首先安装erlong
安装RabbitMQ
如果为了rc.local启动方便,可以将 export PATH=$PATH:/usr/local/erlong/bin 这一行写入到 rabbitmq-server 文件中:
执行后,ps -aux 一下,看到进程中有/usr/local/erlong/lib/erlang/erts-5.10.1/bin/epmd -daemon 和 /usr/local/erlong/lib/erlang/erts-5.10.1/bin/beam.smp 就OK了。
sbin目录下还有一个脚本: rabbitmqctl 也很常用,与 rabbitmq-server 一样需要指明erlong的路径才能正确工作。
常用的方法:
rabbitmqctl start-app 启动后,执行一下这个比较保险
rabbitmqctl list_exchanges 显示当前所有的交换机
rabbitmqctl list_queue 查看当前有效队列情况
二, PHP extension的安装:
PHP操作rabbitmq需要AMQP扩展的支持。下载扩展: http://pecl.php.net/package/amqp ,安装过程与一般扩展一样,/usr/local/php/bin/phpize; ./configure --with-php-config=/usr/local/php/bin/php-config; make && make install
然后编辑php.ini 插入:
[amqp]
extension = amqp.so
重启apache或nginx,查看phpinfo其中有关于anqp的段落,就OK了。
三,如何使用PHP进行实时后端消息处理
首先,要保证PHP文件可以正确的以堵塞方式处理消息。代码可参见上一节RabbitMQ与PHP(一)。
然后,我们来借助Python实现一个多线程的守护进程,由这个守护进程来调用PHP,把PHP作为工作线程。
启动脚本:start_jobs.py
一种工作模式是一共启动N个线程去干活,适合于尽快完成一个大任务;
另一种是保持进程数量,当发现某个进程完成后,再重新将进程启动起来。显然,用户守护处理消息适合这种模式。
具体工作在MyJob.py中,提供了系统Shell调用和采用URL调用两种方式。推荐使用shell直接调用php的方式,这样可以灵活控制Php.ini,比如增加auto_prepend_file、增长max_execution_time等。
实际项目中,假定有5种类型的消息,可以启动20个线程,将thread_id当作参数传递给PHP。PHP将thread_id%5当作待处理类型,就可以得到每种类型有4个线程工作的场景了。
考虑到PHP的执行时间限制及内存泄露问题,可以将consume.php脚本进行一下改进,让PHP脚本每次处理指定数量的消息后就退出,由Python多线程框架重新启动线程,以保证运行稳定可靠。另外,将应答改为手工应答,确保消息获得正确有效处理。
由上图可见,运行开始后,检查2个线程都处于活跃状态,并对消息进行了正确处理,当处理到一定数量后,PHP程序结束,父进程检查到有进程处于完成状态,重新将其启动(第二个绿色框)。完全与预期相符。
相关文章推荐
- RabbitMQ与AMQP协议
- centos安装php的rabbitmq扩展amqp
- RabbitMQ(二)AMQP协议mandatory和immediate标志位区别
- RabbitMQ与AMQP协议详解
- rabbitMQ之AMQP协议
- debian(wheezy)安装rabbitMQ,php-amqp来支持消息队列.
- 使用php-amqplib连接rabbitMQ 学习笔记及总结
- AMQP 0.9.1和1.0协议差别以及rabbitmq支持情况
- RabbitMQ源码解析前奏--AMQP协议
- RabbitMQ介绍2 - AMQP协议
- rabbitmq AMQP协议
- php 的rabbitmq 扩展模块amqp安装
- php 安装 rabbitMQ 扩展 amqp 【一键安装脚本】
- rabbitmq php扩展amqp安装
- rabbitmq入门和ubuntu下安装php-amqp
- windows下安装rabbitmq的php扩展amqp
- windows下安装rabbitmq的php扩展amqp(原创)
- 消息队列基础 RabbitMQ与AMQP协议详解——超大规模高可用OpenStack核心技术深入解析系列(二)
- Linux rabbitmq的安装和安装amqp的php插件