摘要:需要特别明确的概念交换机的持久化,并不等于消息的持久化。消息的处理,是有两种方式,一次性。在上述示例中,使用的,意味着接收全部的消息。注意与是两个不同的队列。后端处理,可以针对每一个启动一个或多个,以提高消息处理的实时性。
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! "); } $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()." "; for($i=0; $i<5; ++$i){ echo "Send Message:".$ex->publish($message . date("H:i:s"), $k_route[i%2])." "; }
由以上代码可以看到,发送消息时,只要有“交换机”就够了。至于交换机后面有没有对应的处理队列,发送方是不用管的。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! "); } $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()." "; //创建队列 $q = new AMQPQueue($channel); $q->setName($q_name); $q->setFlags(AMQP_DURABLE); //持久化 //绑定交换机与队列,并指定路由键 echo "Queue Bind: ".$q->bind($e_name, $k_route)." "; //阻塞模式接收消息 echo "Message: "; $q->consume("processMessage", AMQP_AUTOACK); //自动ACK应答 $conn->disconnect(); /** * 消费回调函数 * 处理消息 */ function processMessage($envelope, $queue) { var_dump($envelope->getRoutingKey); $msg = $envelope->getBody(); echo $msg." "; //处理消息 }
从上述示例中可以看到,交换机既可以由消息发送端创建,也可以由消息消费者创建。
创建一个队列(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
文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。
转载请注明本文地址:https://www.ucloud.cn/yun/21006.html
摘要:在客户端中,当我们将队列名称作为空字符串提供时,我们创建一个带有生成名称的非持久队列方法返回时,变量包含一个随机生成的队列名称。交换和队列之间的关系称为绑定。 使用 php-amqplib 介绍 在前面的教程中,我们创建了一个工作队列。工作队列背后的假设是每个任务都交付给一个工作人员处理。在这一部分中,我们将做一些完全不同的事情——我们将向多个消费者发送消息。此模式称为发布/订阅。 ...
摘要:在中间的框是一个队列的消息缓冲区,保持代表的消费。本教程介绍,这是一个开放的通用的协议消息。我们将在本教程中使用,解决依赖管理。发送者将连接到,发送一条消息,然后退出。注意,这与发送发布的队列匹配。 介绍 RabbitMQ是一个消息代理器:它接受和转发消息。你可以把它当作一个邮局:当你把邮件放在信箱里时,你可以肯定邮差先生最终会把邮件送到你的收件人那里。在这个比喻中,RabbitMQ就...
摘要:前提必读本教程假设是安装在标准端口上运行。这些词可以是任何东西,但通常它们指定连接到消息的某些特性。如果我们违背合同,用一个或四个词,如或那么,这些消息将不匹配任何绑定并将丢失。代码与前面的教程几乎相同。 (using php-amqplib) 前提必读 本教程假设RabbitMQ是安装在标准端口上运行(5672)。如果您使用不同的主机、端口或凭据,则连接设置需要调整。 在哪里得到帮助...
摘要:概述是一款消息队列中间件。他提供了几乎覆盖所有语言的与文档,简直强大的不的了。要详细的去了解学习,我建议还是看官方文档吧。对文章有什么问题或疑问,欢迎在评论区留言。 showImg(https://segmentfault.com/img/bVbc7Yy?w=1066&h=608); 概述 RabbitMQ是一款消息队列中间件。他提供了几乎覆盖所有语言的SDK与文档,简直强大的不的了。...
摘要:我们以前的教程中的日志系统将所有消息广播给所有消费者。我们希望扩展这一点,允许基于其一定严重性程度来过滤消息。在这种情况下,交换机将表现为交换机,并将消息发送到所有匹配队列。 using php-amqplib 前提必读 本教程假设 RabbitMQ 是运行在标准端口上运行(5672). 如果您使用不同的主机、端口或凭据,则连接设置需要调整。 如果您在本教程中遇到困难,可以通过邮件列...
阅读 2222·2021-11-23 09:51
阅读 5581·2021-09-22 15:39
阅读 3307·2021-09-02 15:15
阅读 3460·2019-08-30 15:54
阅读 2313·2019-08-30 15:53
阅读 1364·2019-08-30 14:04
阅读 2402·2019-08-29 18:33
阅读 2316·2019-08-29 13:08