## 工作过程 1. 创建一个vhost. 2. 在该vhost上创建一个connection. 3. 在该connection上创建一个channel. 4. 在该channel上创建一个exchange(如果有就跳过,前提是属性一致).需要设置的属性有交换机名称,类型,是否持久化,是否rabbitmq内部使用,默认false,扩展参数. 5. 在该channel上创建一个queue.需要设置的属性有队列名称,,是否持久化,是否自动删除, 6. 将exchange和queue通过路由键绑定(binding key). 7. 发布消息. ## vhost 一个RabbitMQ的实体上可以有多个vhosts,用户与权限的设置就是依赖于vhosts.对于一般PHP应用,不需要用户权限设定,直接使用默认就存在的的 "/" 就可以了,用户可以使用默认就存在的"guest"账户 .一个简单的配置示例 ``` $config = [ 'host' => '127.0.0.1', 'port' => '5672', 'login' => 'guest', 'password' => 'guest', 'vhost' => '/' ]; ``` ## conenction 与 channel conenction是值物理的连接,一个client与一个server之间有一个连接,一个连接上可以建立多个channel,可以理解为逻辑上的连接.一般应用的情况下,有一个 channel就够用了,不需要创建更多的channel. ``` //创建连接和channel $conn = new AMQPConnection($config); if (!$conn->connect()) { die("cannot connect to the broker\n"); } $channel = new AMQPChannel($conn); ``` ## exchaneg和routingKey 为了将不同类型的消息进行区分,设置了 交换机与路由两个概念,如果,将A类型的消息发送到名为"C1"的交换机,将类型为B的消息发送到"C2"的交换机.当客户端连接C1处理队列消息时,渠道的就只是A类型消息.进一步的,如果A类型消息也非常多,需要进一步细化分区,比如某个客户端只处理A类型消息中针对K用于的消息,routingKey就是用来做这个用途的. ~~~ $e_name = 'elinvo'; //交换机名 $k_route = ['key_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->declareExchange()} \n"; for ($i = 0; $i < 5; $i++) { echo 'Send Message : ' . $ex->publish($message) . date('H:i:s'), $k_route[i % 2] . "\n"; } ~~~ 从上面代码可以看到,发送消息时,只要有"交换机"就够了,至于交换机后面有没有对应的处理队列,发送发是不用管的.routkingKey可以是空的字符串.在示例中,我们使用了两个可以交替发送消息,是为了下面更便于理解routingKey的作用. 对于交换机,有两个重要的概念: 1. 类型,有三种类型: * Fanout类型最简单,这种模型忽略routingKey. * Direct类型是使用最大的,使用确定的routingKey.这种模式下,接收消息时绑定"key_1"则只接收"key_1"的消息. * Topic类型与Direct类似,但是支持通配符进行匹配,比如:"key_*"就会接收"key_1"和"key_2".Topic貌似美好,但是有可能导致不严谨,所以还是推荐使用Direct. 2. 持久化,指定了持久化的交换机,在重新启动时才能重建,否则需要客户端重新声明生成才行. * 需要特别明确的概念:交换机的持久化,并不等于消息的持久化.只有在持久化队列中的消息,才能持久化.如果没有队列,消息是没有地方存储的.消息本身在投递时也有一个持久化标志的 ,PHP中默认投递到持久化交换机就是持久的消息,不用特别指定. ## QUEUE 事实上,队列仅是针对接收方(customer)的,由接收方根据需求创建的.只有队列创建了,交换机才会将新接收到的消息送到队列中,交换机是不会将在队列创建之前的消息放进来的.换句话说,在建立队列之前,发出的所有消息都被丢弃了.下面这个图比官方的图更清楚--QUEUE是属于ReceiveMessage的一部分.  下面看一下创建队列及接收消息的示例: ~~~ $e_name = 'e_linvo'; //交换机名 $q_name = 'q_linvo'; //队列名 $k_route = ''; //路由key $conn = new AMQPConnection($config); 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->declareExchange() . "\n"; //创建队列 $q = new AMQPQueue($channel); $q->setName($q_name); $q->setFlags(AMQP_DURABLE); //持久化 //绑定交换机与队列,并制定路由key 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"; } ~~~ 从上述示例可以看到,交换机即可以由消息发送端创建,也可以由消息消费者创建. 创建一个队列后,需要将队列绑定到交换机上,队列才能工作,routingKey也是在这里指定的.有的资料写成bindingKey,其实是一回事. 消息的处理,是有两种方式: 1. 一次性: 用$q->get([...]),不管渠道娶不到消息都会立即返回,一般情况下使用轮询处理消息队列就要用这种方式. 2. 阻塞:用$q->consum(callback,[...]) ,程序会进入持续侦听状态,没收到一个消息就会调用callback指定的函数一次,直到某个callback函数返回false才结束. 关于callback,PHP的是指出数组的,可以使用['对象名','对象方法']的方式传参. 在上述示例中,使用的routingKey="",意味着接收全部的消息.我们可以将其改为routingKey = "key_1",可以看到结果中仅有设置routingKey为"key_1"的 内容了. 注意:routingKey="key_1"与routingKey = "key_2"是两个不同的队列. 假设:celint1与celint2都连接到key_1的队列上,一个消息被client1处理之后,就不会被client2处理。而 routingkey = " "是另类，client_all绑定到 " " 上，将消息全都处理后，client1和client2上也就没有消息了. 在程序设计上，需要规划好exchange的名称，以及如何使用key区分开不同类型的标记，在消息产生的地方插入发送消息代码。后端处理，可以针对每一个key启动一个或多个client，以提高消息处理的实时性。如何使用PHP进行多线程的消息处理，将在下一节中讲述.