一点点的给你分析RabbiMQ的几种消息模型

<–more–>

1.RabbitMQ的交换机

Direct Exchange（直连交换机）， Fanout Exchange（扇型交换机）， Topic Exchange（主题交换机）与 Headers Exchange（头交换机），RabbitMQ的交换机是这四种，而他们也是各有特点，我们这个不能看其他的文章，首先就得从RabbitMQ的官网上去看这块的内容，毕竟，官网才是正道。

为什么会有交换机，不是直接有队列呢？官网给了我们一个最经典的解释：

The core idea in the messaging model in RabbitMQ is that the producer never sends any messages directly to a queue. Actually, quite often the producer doesn’t even know if a message will be delivered to any queue at all.

意思就是：RabbitMQ中消息传递模型的核心思想是生产者从不将任何消息直接发送到队列。实际上，生产者通常甚至不知道消息是否会被传递到任何队列。

也就是说，RabbitMQ在传递消息的时候，是先从生产者发送给交换机，然后这个时候，由交换机发送给队列，我们在生产消息的时候，实际上如果不关注交换机和队列的绑定的话，我们都不知道消息是发送到哪一个队列。

1.Direct Exchange（直连交换机）

直连型交换机其实是属于最简单的一种交换模式，在这里队列使用路由密钥K绑定到交换机，当具有路由键R的新消息到达直接交换时，如果K=R，则交换会将其路由到队列。

直连型交换机通常用于以轮循方式在多个工作程序（同一应用程序的实例）之间分配任务。

以上都是官网上给我们的解释，大家可以翻译一下他的英文注释即可。

2.Fanout Exchange（扇型交换机）

扇形交换机将消息路由到与其绑定的所有队列，并且路由键将被忽略。如果将N个队列绑定到扇出交换机，则将新消息发布到该交换机时，该消息的副本将传递到所有N个队列。

大家划重点，文档中是不是说明了路由键将被忽略，也就是说路由键在扇形交换机里没有作用，故消息队列绑定扇形交换机时，路由键可为空。而他实际上就相当于一个广播的性质。

3.Topic Exchange（主题交换机）

主题交换机实际上就是一个发布/订阅，根据消息路由键和用于将队列绑定到交换机的模式之间的匹配将消息路由到一个或多个队列。主题交换类型通常用于实现各种发布/订阅模式变体。主题交换通常用于消息的多播路由。

主体交换机官方案例没有图，但是我们可以画一个图来理解一下

图中，如果你要绑定的时候使用了通配符了，那么你再给交换机传递消息的时候，注意，RoutingKey开头是你绑定的时候设置的通配符才可以，符号“#”匹配一个或多个词。

4.Headers Exchange（头交换机）

头交换机不处理路由键。而是根据发送的消息内容中的headers属性进行匹配。在绑定Queue与Exchange时指定一组键值对；当消息发送到RabbitMQ时会取到该消息的headers与Exchange绑定时指定的键值对进行匹配；如果完全匹配则消息会路由到该队列，否则不会路由到该队列。

既然我们已经说明了这么多的交换机，那么就应该给大家写一个配置，然后让大家更快的能够实际运用到项目中。

2.DemoConfig

Direct类型绑定

    //这里先定义Queue和交换机以及RoutingKey

    /**
     * 消息的routing key与队列的binding key相同的队列
     */
    public static final String DIRECT_QUEUE_NAME = "Direct_Queue";
    /**
     * direct 交换机
     */
    public static final String DIRECT_EXCHANGE_NAME = "Direct_Exchange";
    /**
     * routing key
     */
    public static final String DIRECT_ROUTING_KEY_NAME = "Direct_Routing_Key";
    
    //创建队列+队列交换机绑定
    /**
     * Direct交换机
     * @return
     */
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange(DIRECT_EXCHANGE_NAME);
    }
    /**
     * 创建一条持久化的、非排他的、非自动删除的队列
     * @return
     */
    @Bean
    public Queue directQueue(){
        return new Queue(DIRECT_QUEUE_NAME);
    }

    /**
     * Binding,将该routing key的消息通过交换机转发到该队列 匹配键：DIRECT_ROUTING_KEY_NAME
     * @return
     */
    @Bean
    public Binding directBinding(){
        return BindingBuilder.bind(directQueue()).to(directExchange()).with(DIRECT_ROUTING_KEY_NAME);
    }

Fanout类型绑定

    /**
     * 与fanout绑定的第一个队列
     */
    public static final String FIRST_FANOUT_QUEUE_NAME = "First_Fanout_Queue";
    
    /**
     * 与fanout交换机绑定的第二个队列
     */
    public static final String SECOND_FANOUT_QUEUE_NAME = "Send_Fanout_Queue";
    
    /**
     * fanout 交换机
     */
    public static final String FANOUT_EXCHANGE_NAME = "Fanout_Exchange";


     @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(FANOUT_EXCHANGE_NAME);
    }

    @Bean
    public Queue firstFanoutQueue() {
        return new Queue(FIRST_FANOUT_QUEUE_NAME);
    }

    @Bean
    public Queue secondFanoutQueue() {
        return new Queue(SECOND_FANOUT_QUEUE_NAME);
    }

    @Bean
    public Binding firstFanoutBinding() {
        return BindingBuilder.bind(firstFanoutQueue()).to(fanoutExchange());
    }

    @Bean
    public Binding secondFanoutBinding() {
        return BindingBuilder.bind(secondFanoutQueue()).to(fanoutExchange());
    }

Header类型绑定

    /**
     * 处理对象的MQ队列
     */
    public static final String HANDLER_OBJECT_QUEUE_NAME = "Handler_Object_Queue";

    /**
     * 处理对象的MQ队列
     */
    public static final String HANDLER_OBJECT_EXCHANGE = "Handler_Object_Exchange";
    
    @Bean
    public Queue handleObjectQueue() {
        return new Queue(HANDLER_OBJECT_QUEUE_NAME);
    }

    @Bean
    public HeadersExchange headersExchange(){
        return new HeadersExchange(HANDLER_OBJECT_EXCHANGE);
    }
    @Bean

    public Binding headerBinding(){
        Map<String, Object> map = new HashMap<>();
        map.put("header1", "value1");
        map.put("header2", "value2");
        return BindingBuilder.bind(handleObjectQueue()).to(headersExchange()).whereAll(map).match();
    }

Topic类型绑定

 /**
     * 与topic绑定的第一个队列
     *
     */
    public final static String FIRST_TOPIC_QUEUE = "First_Topic_Queue";
    /**
     * 与topic绑定的第二个队列
     *
     */
    public final static String SECOND_TOPIC_QUEUE = "Second_Topic_Queue";
    /**
     * 与topic绑定的第二个队列
     *
     */
    public final static String TOPIC_EXANGE_NAME = "Topic_Exchange";
    
    /**
     *  Topic型交换机模型:将routingKey与binding key做通配符匹配，转发消息到匹配的队列
     *  “#”匹配一个词或多个词，“*”只匹配一个词。
     */
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXANGE_NAME);
    }
    @Bean
    public Queue firstTopicQueue() { return new Queue(FIRST_TOPIC_QUEUE);}

    @Bean
    public Queue secondTopicQueue() {
        return new Queue(SECOND_TOPIC_QUEUE);
    }

    /**
     * 将FirstQueue和TopicExchange绑定，绑定的键为FIRST_TOPIC_QUEUE
     */
    @Bean
    public Binding firstTopicBinding() {
        return BindingBuilder.bind(firstTopicQueue()).to(topicExchange()).with(FIRST_TOPIC_QUEUE);
    }
    /**
     * 将SecondQueue和TopicExchange绑定，绑定的键为通配符使用的 Topic_#
     * 消息携带的路由键是以Topic.开头,都会分发到该队列
     */
    @Bean
    public Binding secondTopicBinding() {
        return BindingBuilder.bind(secondTopicQueue()).to(topicExchange()).with("Topic.#");
    }

至于发送消息，相信大家肯定也都会，

 rabbitTemplate.convertAndSend(RabbitMQConfig.TOPIC_EXANGE_NAME, routingKey, message,new CorrelationData(UUID.randomUUID().toString()));

第一个参数：交换机配置

第二个参数： RoutingKey

第三个参数：消息

第四个参数：对于使用消息确认的时候，需要的一个唯一ID。

发送成功之后，我们同样需要看是否发送成功，这时候就有了confirm，而这个confirm，就是用来确认消息是否发送成功，而这块也是从你在配置RabbitMQ的时候配置决定的，是否手动确认消息发送成功。

    #是否支持发布确认
    publisher-confirms: true
    #是否支持发布返回
    publisher-returns: true

相对应的，你就应该在消息发出的时候加入一些代码，比如：

        //ACK确认机制
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.convertAndSend(RabbitMQConfig.SIMPLE_QUEUE_NAME, message);

而ConfirmCallback里面的参数呢，是从下面来的。

confirm确认 此处的confirm确认是发送的ACK确认机制
    RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            if(ack){
                System.out.println(correlationData);
                //如果confirm返回成功 则进行更新
                System.out.println("confirm确认消息已经发出");
            } else {
                //失败则进行具体的后续操作:重试 或者补偿等手段
                System.err.println("异常处理...");
            }
        }
    };

以上的代码可以直接和上一篇阿粉讲的RabbitMQ整合Springboot的案例直接放入就可以使用。

《RabbitMQ官方文档》《RabbitMQ实战》