最新版本的RabbitMQ有四种交换机类型，分别是Direct exchange、Fanout exchange、Topic exchange、Headers exchange。本文用SpringBoot工程具体介绍四种交换机的使用。

一、Direct Exchange

需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。

    @Bean
    DirectExchange directExchange(){
        return new DirectExchange("direct_change");
    }
    @Bean
    Queue queue1(){
        return new Queue(Direct_QUEUE_NAME,true,false,false,null);
    }

    @Bean
    Queue queue2(){
        return new Queue(Direct_QUEUE_NAME2,true,false,false,null);
    }

    @Bean
    Binding binding(){
        return BindingBuilder.bind(queue1()).to(directExchange()).with("a");
    }

    @Bean
    Binding binding2(){
        return BindingBuilder.bind(queue2()).to(directExchange()).with("b");
    }

二、Fanout Exchange

发送到交换机的消息会被转发到与该交换机绑定的所有队列上。消息广播。Fanout交换机转发消息是最快的。

    @Bean
    FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanout_change");
    }
    @Bean
    Queue queue3(){
        return new Queue(Direct_QUEUE_NAME3,true,false,false,null);
    }

    @Bean
    Queue queue4(){
        return new Queue(Direct_QUEUE_NAME4,true,false,false,null);
    }

    @Bean
    Binding binding3(){
        return BindingBuilder.bind(queue3()).to(fanoutExchange());
    }

    @Bean
    Binding binding4(){
        return BindingBuilder.bind(queue4()).to(fanoutExchange());
    }

三、Topic Exchange

在路由模式的基础上，支持了对key的通配符匹配（星号以及井号），以满足更加复杂的消息分发场景。。“#” : 匹配一个或者多个
“*”：匹配一个。abc.#会匹配所有abc开头的key

    @Bean
    TopicExchange topicExchange(){
        return new TopicExchange("topic_change");
    }

    @Bean
    Queue queue7(){
        return new Queue(Direct_QUEUE_NAME7,true,false,false,null);
    }

    @Bean
    Queue queue8(){
        return new Queue(Direct_QUEUE_NAME8,true,false,false,null);
    }


    @Bean
    Binding binding7(){
        return BindingBuilder.bind(queue7()).to(topicExchange()).with("abc.#");
    }

    @Bean
    Binding binding8(){
        return BindingBuilder.bind(queue8()).to(topicExchange()).with("#.d");
    }

四、Headers Exchanges

不处理路由键。而是根据发送的消息内容中的headers属性进行匹配。在绑定Queue与Exchange时指定一组键值对；当消息发送到RabbitMQ时会取到该消息的headers与Exchange绑定时指定的键值对进行匹配；如果完全匹配则消息会路由到该队列，否则不会路由到该队列。headers属性是一个键值对，可以是Hashtable，键值对的值可以是任何类型。而fanout，direct，topic 的路由键都需要要字符串形式的。

匹配规则x-match有下列两种类型：

x-match = all ：表示所有的键值对都匹配才能接受到消息

x-match = any ：表示只要有键值对匹配就能接受到消息

    @Bean
    HeadersExchange headersExchange(){
        return new HeadersExchange("head_change");
    }


   @Bean
    Queue queue5(){
        return new Queue(Direct_QUEUE_NAME5,true,false,false,null);
    }

    @Bean
    Queue queue6(){
        return new Queue(Direct_QUEUE_NAME6,true,false,false,null);
    }


  @Bean
    Binding binding5(){
        return BindingBuilder.bind(queue5()).to(headersExchange()).where("age").matches("12");
    }

    @Bean
    Binding binding6(){
        return BindingBuilder.bind(queue6()).to(headersExchange()).where("name").matches("luomo");

单元测试

@SpringBootTest
class DemoApplicationTests {


    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() {
        rabbitTemplate.convertSendAndReceive("direct_change","a","hello queue1");
        rabbitTemplate.convertSendAndReceive("direct_change","b","hello queue2");
    }

    @Test
    void fanoutExchange() {
        rabbitTemplate.convertSendAndReceive("fanout_change","","hello queue");
    }


    @Test
    void headerExchange() {
        Message build = MessageBuilder.withBody("hello luomo".getBytes()).setHeader("name", "luomo").build();
        rabbitTemplate.convertSendAndReceive("head_change","",build);

        Message message = MessageBuilder.withBody("hello ddd".getBytes()).setHeader("age", "12").build();
        rabbitTemplate.convertSendAndReceive("head_change","",message);
    }


  @Test
    void topicExchange() {
        rabbitTemplate.convertSendAndReceive("topic_change","abc.d","test");
    }
}

一、RabbitMQ的简单介绍

RabbitMQ作为一个消息队列，它负责提供一个通用的消息发送和接收平台,并且保证消息在传输过程中的安全可靠。
消息(Message)由Client(客户端)发送，RabbitMQ接收到消息之后通过交换机转发到对应的队列上面。Worker会从队列中获取未被读取的数据处理。

二、RabbitMQ的四种交换机

• 直连交换机：Direct exchange
• 扇形交换机：Fanout exchange
• 主体交换机：Topic exchange
• 首部交换机：Headers exchange

1、直连交换机

直连交换机是一种带路由功能的交换机，一个队列会和一个交换机绑定，除此之外再绑定一个routing_key，当消息被发送的时候，需要指定一个binding_key，这个消息被送达交换机的时候，就会被这个交换机送到指定的队列里面去。同样的一个binding_key也是支持应用到多个队列中的。
这样当一个交换机绑定多个队列时，就会被送到对应的队列去处理。

如图所示：当指定routing_key为key1时，消息队列1和2都会收到消息，下面用伪代码演示一下（只做演示，相关依赖省略）：

MQ配置类

@Configuration
public class MQConfig {
	//创建三个队列1,2,3
	//Queue的第一个参数为队列名称，第二个参数为是否持久存在
	 @Bean
    public Queue directQueue1() {
        return new Queue("queue1", true);
    }

    @Bean
    public Queue directQueue2() {
        return new Queue("queue2", true);
    }

    @Bean
    public Queue directQueue3() {
        return new Queue("queue3", true);
    }
	//创建直连交换机，参数为交换机的名称
	@Bean
    public DirectExchange directExchange() {
        return new DirectExchange("DIRECT_EXCHANGE");
    }
	//将三个队列都与该直连交换机绑定起来，并赋予上面说的binding_key(也可以说是routing_key)
	@Bean
    public Binding bindingDirectExchange1() {
        return BindingBuilder.bind(directQueue1()).to(directExchange()).with("key.1");
    }

    @Bean
    public Binding bindingDirectExchange2() {
        return BindingBuilder.bind(directQueue2()).to(directExchange()).with("key.1");
    }

    @Bean
    public Binding bindingDirectExchange3() {
        return BindingBuilder.bind(directQueue3()).to(directExchange()).with("key.2");
    }
}

MQSnder发送者类

@Service
public class MQSender {
	//注入AmqpTemplate接口，该接口定义了发送和接收消息的基本操作
	@Autowired
    AmqpTemplate amqpTemplate;
    
    public void send(String message) {
    	//第一个参数指将消息发送到该名称的交换机，第二个参数为对应的routing_key,第三个参数为发送的具体消息   
    	amqpTemplate.convertAndSend("DIRECT_EXCHANGE", "key.1", message);   				 
    }
}

MQReceive消费者类

@Service
public class MQReceiver {
	private static final Logger logger = LoggerFactory.getLogger(MQReceiver.class);
	//此注解表示监听某个队列，参数为队列名
    @RabbitListener(queues = "queue1")
    public void receive1(String message) {
        logger.info("receive : dir1 message {}", message);
    }

    @RabbitListener(queues = "queue2")
    public void receive2(String message) {
        logger.info("receive : dir2 message {}", message);
    }

    @RabbitListener(queues = "queue3")
    public void receive3(String message) {
        logger.info("receive : dir3 message {}", message);
    }
}

调用发送者的send方法后，发现结果是队列1和2收到了消息，符合预期
适用场景：有优先级的任务，根据任务的优先级把消息发送到对应的队列，这样可以派更多的资源去处理高优先级的队列。

2、扇形交换机

扇形交换机是最基本的交换机类型，它能做的事非常简单——广播消息，扇形交换机会把能接收到的消息全部发送给绑定在自己身上的队列。因为广播不需要"思考"，所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的。

下面用伪代码演示一下。
MQ配置类

@Configuration
public class MQConfig {
	//创建三个队列1,2,3
	//Queue的第一个参数为队列名称，第二个参数为是否持久存在
	 @Bean
    public Queue fanoutQueue1() {
        return new Queue("queue1", true);
    }

    @Bean
    public Queue fanoutQueue2() {
        return new Queue("queue2", true);
    }

    @Bean
    public Queue fanoutQueue3() {
        return new Queue("queue3", true);
    }
	//创建扇形交换机，参数为交换机的名称
	@Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange ("FANOUT_EXCHANGE");
    }
	//将三个队列都与该交换机绑定起来，无需binding_key
	@Bean
    public Binding bindingFanoutExchange1() {
        return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
    }

    @Bean
    public Binding bindingFanoutExchange2() {
        return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
    }

    @Bean
    public Binding bindingFanoutExchange3() {
        return BindingBuilder.bind(fanoutQueue3()).to(fanoutExchange());
    }
}

MQSnder发送者类

@Service
public class MQSender {
	//注入AmqpTemplate接口，该接口定义了发送和接收消息的基本操作
	@Autowired
    AmqpTemplate amqpTemplate;
    
    public void send(String message) {
    	//第一个参数指将消息发送到该名称的交换机，第二个参数为对应的routing_key(此时设置为空字符串即可),第三个参数为发送的具体消息   
    	amqpTemplate.convertAndSend("FANOUT_EXCHANGE", "", message);   				 
    }
}

MQReceive消费者类

@Service
public class MQReceiver {
	private static final Logger logger = LoggerFactory.getLogger(MQReceiver.class);
	//此注解表示监听某个队列，参数为队列名
    @RabbitListener(queues = "queue1")
    public void receive1(String message) {
        logger.info("receive : fanout message {}", message);
    }

    @RabbitListener(queues = "queue2")
    public void receive2(String message) {
        logger.info("receive : fanout message {}", message);
    }

    @RabbitListener(queues = "queue3")
    public void receive3(String message) {
        logger.info("receive : fanout message {}", message);
    }
}

调用MQSend的send()方法，三个队列都能接收到消息。
适用场景：需要给所有绑定该交换机的队列直接发送消息时使用。

3、主题交换机

直连交换机的routing_key方法非常简单，如果希望将一条消息发送给多个队列，那么这个交换机需要绑定非常多的routing_key，这样的话消息的管理就会非常的困难。
所以RabbitMQ提供了一种主题交换机，发送到主题交换机上的消息需要携带制定规则的routing_key，主题交换机会根据这个规则将数据发送到对应的队列上。
主题交换机的routing_key需要有一定的规则，交换机和队列绑定时候设置的binding_key需要采用*.#.*…的格式，每个部分用.分开，其中：

• *表示一个单词
• #表示任意数量(零个或多个)单词。
  假设有一条消息的routing_key为com.lrving.www，那么带有这样binding_key的几个队列都有收到消息:

1. com…
2. …www
3. com.#
4. …

下面是网上的一张图，清楚描述了主题交换机的消息发送规则

当一个队列的绑定键为#的时候，这个队列将会无视消息的路由键，接收所有的消息。下面用伪代码演示一下
MQ配置类

@Configuration
public class MQConfig {
	@Bean
    public Queue topicQueue1() {
        return new Queue("TOPIC_QUEUE1", true);
    }
    @Bean
    public Queue topicQueue2() {
        return new Queue("TOPIC_QUEUE2", true);
    }
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("TOPIC_EXCHANGE");
    }
  	//将topicQueue1与topicExchange交换机绑定
    @Bean
    public Binding bindQueue1() {
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("topic.key1");
    }
    //将topicQueue2与topicExchange交换机绑定
    @Bean
    public Binding bindQueue2() {
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("topic.#");
    }
}

MQSnder发送者类

@Service
public class MQSender {
	//注入AmqpTemplate接口，该接口定义了发送和接收消息的基本操作
	@Autowired
    AmqpTemplate amqpTemplate;
   public void sendTopic(String message) {
        amqpTemplate.convertAndSend("TOPIC_EXCHANGE", "topic.key1", message);
        amqpTemplate.convertAndSend("TOPIC_EXCHANGE", "topic.key2", message);
    }
}

MQReceive消费者类

@Service
public class MQReceiver {
	private static final Logger logger = LoggerFactory.getLogger(MQReceiver.class);
	@RabbitListener(queues = "TOPIC_QUEUE1")
    public void receiveQueue1(String message) {
        logger.info("receive : queue1 {}", message);
    }

    @RabbitListener(queues ="TOPIC_QUEUE2")
    public void receiveQueue2(String message) {
        logger.info("receive : queue2 {}", message);
    }
}

调用MQSender的send()方法，发现队列1收到了一条消息，而队列2收到了两条消息,符合参数#的描述。

4、首部交换机

首部交换机是忽略routing_key的一种路由方式。路由器和交换机路由的规则是通过Headers信息来交换的，这个有点像HTTP请求中的请求头。将一个交换机声明成首部交换机，绑定一个队列的时候，定义一个Hash的数据结构，消息发送的时候，会携带一组hash数据结构的信息，当Hash内容匹配上的时候，消息就会被写入队列。
绑定交换机和队列的时候，Hash结构中要求携带一个键"x-match"，这个键的Value可以是any或者all，这代表消息携带的Hash是需要全部匹配(all)，还是仅仅匹配一个键(any)就可以了。相比较直连交换机，首部交换机的优势是匹配的规则不被限定为字符串(string)。
下面用伪代码演示一下：
MQ配置类

@Configuration
public class MQConfig {
	 @Bean
    public Queue headersQueue() {
        return new Queue("HEADERS_QUEUE");
    }
    @Bean
    public HeadersExchange headersExchange() {
        return new HeadersExchange("HEADERS_EXCHANGE");
    }
    //将headersQueue与HeadersExchange交换机绑定
    @Bean
    public Binding bingHeadersQueue() {
    	//map为绑定的规则
        Map<String, Object> map = new HashMap<>();
        map.put("headers1", "value1");
        map.put("headers2", "value2");
        //whereAll表示需要满足所有条件
        return BindingBuilder.bind(headersQueue()).to(headersExchange()).whereAll(map).match();
    }
}

MQSnder发送者类

@Service
public class MQSender {
	//注入AmqpTemplate接口，该接口定义了发送和接收消息的基本操作
	@Autowired
    AmqpTemplate amqpTemplate;
   public void sendHeaders(String message) {
   		//配置消息规则
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setHeader("headers1", "value1");
        messageProperties.setHeader("headers2", "value2");
        //要发送的消息，第一个参数为具体的消息字节数组，第二个参数为消息规则
        Message msg = new Message(result.getBytes(), messageProperties);
        amqpTemplate.convertAndSend("HEADERS_EXCHANGE", "", msg);
    }
}

MQReceive消费者类

@Service
public class MQReceiver {
	private static final Logger logger = LoggerFactory.getLogger(MQReceiver.class);
   @RabbitListener(queues = "HEADERS_QUEUE")
    public void receiveHeadersQueue(byte[] message) {
        logger.info("receive : HeadersQueue {}", new String(message));
    }
}

调用MQSender的send()方法，发现队列收到了消息，符合预期。

参考资料

参考资料

交换机概念

交换机可以理解成具有路由表的路由程序，仅此而已。每个消息都有一个称为路由键（routing key）的属性，就是一个简单的字符串。

最新版本的RabbitMQ有四种交换机类型，分别是Direct exchange、Fanout exchange、Topic exchange、Headers exchange。

交换机的作用： 生产者向broker（rabbitmq服务器）发送消息，交换机通过生产者绑定的路由键，将消息推送到不同的消息队列中。而消费者，只绑定队列，从队列中获取消息。

direct 直连交换机

工作模式图解

生产者发送的消息；交换机会根据不同的路由键，发送到对应的队列；

springboot代码

MQ配置类，声明交换机、队列，路由键绑定

/**
 * 声明交换机、队列、路由键绑定
 * /
@Configuration
puvlic class RabbitConfig {
	/**
	 * 创建直连交换机
	 */
	@Bean
	public DirectExchange createExchange() {
	    // 交换机名字；是否持久化；是否自动删除
		return new DirectExchange("testE", true, false);
	}
	/**
	 * 创建队列
	 */
	@Bean
	public Queue createQueue() {
	    // 交换机名字；是否持久化；是否自动删除
		return new Queue ("testQ", true, false, false);
	}
	/**
	 * 通过路由键绑定交换机和队列
	 */
	@Bean
	public Binding createBinding() {
	    // 交换机名字；是否持久化；是否自动删除
		return BindingBuilder
		.bind(this.createQueue())
		.to(this.createExchange())
		.with("testR");
	}
}

生产者

/**
 * 消息生产者
 */
@Service
public class ProduceMsg {
	@Autowire
	private RabbitTemplate rabbitTemplate;
	public void sendMsg(Object msg){
		// 消息唯一标识
	 	CorrelationData correlationData= new CorrelationData();
	 	correlationData.setId(msg.getId());
		rabbitTemplate.converAndSend("testE", "testR", msg, correlationData);
	}
}

消费者

@Conponent
public class ConsumeMsg {
	/**
	 * 消费者监听队列
	 */
	@RabbitListener(queues = "testQ")
	public void sendMsg(String msg){
		log.info("接收到消息：{}", msg);
		// ......业务逻辑消费消息；
	}
}

Fanout扇出交换机

工作模式图解

生产者发送到交换机的消息；会发送到绑定到该交换机的所有队列

springboot代码

MQ配置类，声明交换机、队列，绑定

/**
 * RabbitMQ配置类
 */
@Configuration
public class RabbitMqConfig {
    @Bean
    public Queue fanoutQueueA()
    {
        return new Queue("queueA", true, false, false);
    }
 
    @Bean
    public Queue fanoutQueueB()
    {
        return new Queue("queueB", true, false, false);
    }
 
    @Bean
    public Queue fanoutQueueC()
    {
        return new Queue("queueC", true, false, false);
    }
 
    @Bean
    FanoutExchange fanoutExchange()
    {
        return new FanoutExchange("exchangeFanout");
    }
 
    @Bean
    Binding bindingExchangeA()
    {
        return BindingBuilder.bind(fanoutQueueA()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeB()
    {
        return BindingBuilder.bind(fanoutQueueB()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeC()
    {
        return BindingBuilder.bind(fanoutQueueC()).to(fanoutExchange());
    }
}

生产者

/**
 * 消息生产者
 */
@Service
public class ProduceMsg {
	@Autowire
	private RabbitTemplate rabbitTemplate;
	public void sendMsg(Object msg){
		// 消息唯一标识
	 	CorrelationData correlationData= new CorrelationData();
	 	correlationData.setId(msg.getId());
		rabbitTemplate.converAndSend("exchangeFanout", "", msg, correlationData);
	}
}

消费者

@Conponent
public class ConsumeMsg {
	/**
	 * 消费者监听队列
	 */
	@RabbitListener(queues = "testQ")
	public void sendMsg(String msg){
		log.info("接收到消息：{}", msg);
		// ......业务逻辑消费消息；
	}
}

Topic主题交换机

topic模式跟direct的区别是，topic模式可以用通配符的方式，对路由键进行绑定；达到更灵活路由消息的效果。
交换机的routingKey不能随意写；必须是以点号分隔；如aa.bb; cc.dd.ee等形式
*号代表一个单词；#号代表0个或多个单词

工作模式图解

图中队列1绑定的路由键是 *.*.routeA
图中队列2绑定的路由键是 routeA.#
生产者向该交换机的routeA.xxx.routeA路由键发送消息；那么队列1和2都会收到消息

springboot代码

MQ配置类，声明交换机、队列，绑定

@Configuration
public class TopicRabbitMqConfig
{
    /**
     * 队列A
     */
    @Bean
    public Queue topicQueueA() {
        return new Queue("topic_queue_A", true, false, false);
    }
 
    /**
     * 队列B
     */
    @Bean
    public Queue topicQueueB() {
        return new Queue("topic_queue_B", true, false, false);
    }
 
    /**
     * Topic交换器
     */
    @Bean
    TopicExchange exchange() {
        return new TopicExchange("topic_exchange", true, false);
    }
 
    /**
     * 绑定A
     */
    @Bean
    Binding bindingExchangeQueueA() {
        //将队列和交换机绑定, 并设置用于匹配键：routingKey
        return BindingBuilder.bind(topicQueueA()).to(exchange()).with("*.*.routeKey");
    }
 
    /**
     * 绑定B
     */
    @Bean
    Binding bindingExchangeQueueB(Queue topicQueueB, TopicExchange exchange) {
        //将队列和交换机绑定, 并设置用于匹配键：routingKey
        return BindingBuilder.bind(topicQueueB()).to(exchange()).with("routeKey.#");
    }
}

生产者

/**
 * 消息生产者
 */
@Service
public class ProduceMsg {
	@Autowire
	private RabbitTemplate rabbitTemplate;
	public void sendMsg(Object msg){
		// 消息唯一标识
	 	CorrelationData correlationData= new CorrelationData();
	 	correlationData.setId(msg.getId());
		rabbitTemplate.converAndSend("topic_exchange", "routeKey.test.routeKey", 
	
	msg, correlationData);
	}
}

这个生产者发送的消息；队列topic_queue_A和topic_queue_B都会接收到该生产者发送的消息

header交换机

该交换机不同于其他机制，且实际开发不常用，此处不作讲解