RabbitMQ消息队列-交换机Direct模式

2018-12-11 21:00:00
赵勤松
原创
1853
摘要:对于复杂的消息发送机制,我们一般使用RabbitMQ内转的交换机模式进行消息的分发,相对于一对一,一对多模式下对单个消息队列的处理模式,通常交换机同时管理着多个消息队列,下面我们将详细剖析交换机方式中的Direct类型

在Direct类型的路由控制下,交换机将根据消息中含有的路由键信息,发送给指定监听该路由键的消息队列,下面我们以ThinkPHP的代码来详细说明这一过程。


创建一个控制器类MQExchangeDirect,其对应的类文件MQExchangeDirect.php中的代码如下


<?php
/**
 * Created by PhpStorm.
 * User: zhaoqinsong
 * Date: 2018/12/11
 * Time: 9:44 AM
 */
namespace app\msq\controller;
use think\App;
use think\Controller;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
class MQExchangeDirect extends Controller
{
    var $msq_connect;
    var $msq_channel;
    public function __construct(App $app)
    {
        parent::__construct($app);
        $this->msq_connect = new AMQPStreamConnection("localhost", 5672, "phpmsq", "123456");
        $this->msq_channel = $this->msq_connect->channel();
        $this->msq_channel->exchange_declare("exchange1", "direct");
    }
    public function index()
    {
        echo "rabbitmq of test";
    }
    public function mq_send()
    {
        $msg = new AMQPMessage("Hello World to test1");
        $this->msq_channel->basic_publish($msg, "exchange1", "test1");
        $msg = new AMQPMessage("Hello World to test2");
        $this->msq_channel->basic_publish($msg, "exchange1", "test2");
    }
    public function mq_recv1()
    {
        $this->msq_channel->queue_declare("simple1", false, true, false, false);    //  生成一个持久化消息队列
        $this->msq_channel->queue_bind("simple1", "exchange1", "test1");
        $this->msq_channel->basic_qos(null,1, null);
        //  处理消息队列simple中的消息
        $callback = function($msg) {
            $body = $msg->body;
            echo "recv:$body\n";
        };
        $this->msq_channel->basic_consume("simple1", "", false, true, false, false, $callback);
        while ($this->msq_channel->callbacks) {
            $this->msq_channel->wait();
        }
    }
    public function mq_recv2()
    {
        $this->msq_channel->queue_declare("simple2", false, true, false, false);    //  生成一个持久化消息队列
        $this->msq_channel->queue_bind("simple2", "exchange1", "test2");
        $this->msq_channel->basic_qos(null,1, null);
        //  处理消息队列simple中的消息
        $callback = function($msg) {
            $body = $msg->body;
            echo "recv:$body\n";
        };
        $this->msq_channel->basic_consume("simple2", "", false, true, false, false, $callback);
        while ($this->msq_channel->callbacks) {
            $this->msq_channel->wait();
        }
    }
    public function __destruct()
    {
        $this->msq_channel->close();
        $this->msq_connect->close();
    }
}


指定路由,设定/direct/send指向此类的mq_send方法,/direct/recv1指向mq_recv1方法,/direct/recv2指向mq_recv2方法。

在命令行下,先创建两个服务器连接,用于执行接收消息的程序


# php public/index.php /direct/recv1(接收路由键test1)
# php public/index.php /direct/recv2(接收路由键test2)
然后再创建一个服务器连接,用于执行发送消息的程序


# php public/index.php /direct/send
这时,可以在两个消息接收端,看到生产者发送的3条消息,分别被两个消息接送端各自接收了1条,有1条被抛弃了


# php public/index.php /direct/recv1
recv:Hello World to test1
# php public/index.php /direct/recv2
recv:Hello World to test2
可以看到,消息队列1接收了路由键为test1的消息,消息队列2接收了路由键为test2的消息,而路由键指定为test3的消息,因为没有监听该路由键的消息队列,被交换机抛弃了。


文章分类
联系我们
联系人: powereye
Email: zqs@someapp.cn
QQ: 1134846
微信: powereye