RabbitMQ（1）三种exchange总结

た入场券 2022-07-12 13:00 124阅读 0赞

……List……

1.AMQP

2.几个基本概念

3.三种 exchange

4.思考总结

# ……1.AMQP…… #

在AMQP（高级消息队列协议）协议中，queue、exchange、binding构成了协议的核心。exchange是一个重要的组件，那么如何理解exchange呢？刚刚开始学习消息队列的时候，想起作品展时候做的一个基于SMTP协议的邮箱。不同的是MQ中没有发送者和接受者的概念，也不是客户端服务器的概念。MQ专注于应用程序之间的消息通信，是一种生产者和消费者之间的关系。exchange是producer和consumer之间消息交换机。

![Center][]

# ……2.几个基本概念…… #

在介绍exchange类型之前，需要了解几个基本的概念。

*  Broker：消息队列服务器实体
 *  exchange：消息交换机，指定消息规则，处理消息和队列之间的关系
 *  queue：队列载体，消息投入队列中
 *  binding：绑定，把exchange和queue按照路由规则绑定起来
 *  Routing Key：路由关键字。exchange根据这个进行消息投递
 *  vhost：虚拟消息服务器，每个RabbitMQ服务器都能够创建虚拟消息服务器。
 *  producer：生产者，投递消息的程序
 *  consumer：消费者，接受消息的程序
 *  channel：信道（重要概念），打开信道才能进行通信，一个channel代码一个会话任务。

# ……3.三种exchange…… #

vhost通过创建channel连接程序，会根据routingKey将M从exchange到queue，那根据什么规则进行投递消息？如果有好多不同类型的M，那么应该如何选择？AMQP中定义了不同类型的exchange就可以发挥作用了。一种三种类型：fanout,topic,direct。每一种类型实现不同的路由算法。

![Center 1][]

**3.1 Fanout Exchange 不规则路由**

将收到的消息广播到绑定的队列上。当你发送一条消息到fanout exchange上，它会把消息投递给所有附加到此交换机上面的队列。很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。

![Center 2][]

生产者（提供服务的系统）：

package cn.itcast.rabbitmq.ps;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class Send {
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
    
            // 消息内容
            String message = "商品已经新增，id = 1000";
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }

消费者（调用服务的系统）：

package cn.itcast.rabbitmq.ps;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class Recv {
    
        private final static String QUEUE_NAME = "test_queue_fanout_1";
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" 前台系统： '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

**3.2 Direct Exchange 处理路由**

生产者在发送消息的时候都需要指定一个routingkey和exchange，exchange在接到rooutingkey的时候与该exchange关联的所有binding中的bindingkey进行比较，如果相等，则发送到binding对应的queue中。

![Center 3][]

生产者（提供服务的系统）：

package cn.itcast.rabbitmq.routing;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class Send {
    
        private final static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
            // 消息内容
            String message = "删除商品， id = 1001";
            channel.basicPublish(EXCHANGE_NAME, "delete", null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }

消费者（调用服务的系统）：

package cn.itcast.rabbitmq.routing;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class Recv {
    
        private final static String QUEUE_NAME = "test_queue_direct_1";
    
        private final static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" 前台系统： '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

**3.3 Topic Exchange 通配符路由**

将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。这个很像Javascript中的正则表达式。符号“\#”匹配一个或多个词，符号“\*”匹配不多不少一个词。因此“audit.\#”能够匹配到“audit.irs.corporate”，但是“audit.\*” 只会匹配到“audit.irs”。

![Center 4][]

生产者（提供服务的系统）：

package cn.itcast.rabbitmq.topic;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class Send {
    
        private final static String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
    
            // 消息内容
            String message = "删除商品，id = 1001";
            channel.basicPublish(EXCHANGE_NAME, "item.delete", null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }

消费者（调用服务的系统）：

消费者1：

package cn.itcast.rabbitmq.topic;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class Recv {
    
        private final static String QUEUE_NAME = "test_queue_topic_1";
    
        private final static String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" 前台系统： '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

消费者2：

package cn.itcast.rabbitmq.topic;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class Recv2 {
    
        private final static String QUEUE_NAME = "test_queue_topic_2";
    
        private final static String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.#");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" 搜索系统： '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

# ……4.思考总结…… #

总结一下生产者和消费者都需要做什么事情。多学习多思考。

![Center 5][]

[Center]: /images/20220710/b90411328f8b414c92f89a2bdbb30e79.png
[Center 1]: /images/20220710/027b50f7efe14d0193ce0abc1e1e04c2.png
[Center 2]: /images/20220710/e95116abfb934a9b931f958c5f01ef85.png
[Center 3]: /images/20220710/8621b904e6c44c0a913bbdd1ae1ae1ab.png
[Center 4]: /images/20220710/5e71a9af7a484781b8eee61a4f0ab718.png
[Center 5]: /images/20220710/a2b88fd73b26416eb03961c4d5ecbbf2.png