文章目录

- direct模式：直来直往，指哪到哪儿
- - 配置：
  - 生产者：
  - 消费者：
  - 特点：
交换机模式：
- 交换机模式1：topic模式，
- - 配置：
  - 生产者：
  - 消费者：
  - 过程：
- 交换机模式2:Fanout模式：
- - 配置：
  - 生产者：
  - 消费者：
  - 特点：
- 交换机模式3：Headers模式，
- - 配置：
  - 生产者：
  - 消费者：
  - 特点：
- 总结：

生产者：制造消息
消费者：获取消息，处理消息
路由机制：中转消息
四种模式：

后三种是交换机模式

Direct：不需要交换机，直接发送到指定队列，一对一
Topic：  需要key
Fanout： 广播模式，不需要key
Headers：注意模式比较奇特，需要一个map，里面包含需要的header，只有匹配(或者不匹配）其中的header，才能放到对应的队列中

direct模式：直来直往，指哪到哪儿

代码demo：
direct模式：

配置：

@Configuration
public class MQConfig { 
    public static final String QUEUE="queue";
  /** * 1, diect 模式 * @return */
    @Bean
    public Queue queue(){ 
        return new Queue("queue",true);
    }
}

生产者：

@Service
@Slf4j
public class MQSender { 
    @Autowired
    AmqpTemplate amqpTemplate;
    public void send(Object message){ 
        String msg=RedisService.beanToString(message);
        log.info("send at:{},msg:{} ", MQConfig.QUEUE,msg);
        amqpTemplate.convertAndSend( MQConfig.QUEUE,msg);
    }
}

消费者：

@Service
@Slf4j
public class MQReceiver { 
    @Autowired
    AmqpTemplate amqpTemplate;
    @RabbitListener(queues=MQConfig.QUEUE)
    public void receive(String message){ 
      log.warn("receive message from {},data:{}",MQConfig.QUEUE,message);
    }
}

特点：

很直接，除了生产者和消费者，就没有更多的部分

交换机模式：

比direct多了个中间商（媒婆）,交换机

交换机模式1：topic模式，

生产者发送消息时需要指定key和交换机，交换机知道key对应的queue，那么交换机就能将消息放到适合的队列中，交换机任务也就完成了。

配置：

@Configuration
public class MQConfig { 
    public static final String TOPIC_QUEUE1="topic.queue1";
    public static final String TOPIC_QUEUE2="topic.queue2";
  /** * 2 ,topic 模式 交换机 * @return */
    @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);
    }
    /** *给队列绑定上key和交换机 */
    @Bean
    public Binding topicBinding1(){ 
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("topic.key1");
    }
    @Bean
    public Binding topicBinding2(){ 
    /** * # 代表通配符 */
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("topic.#");
    }
}

生产者：

发送消息时需要指定交换机和key

@Service
@Slf4j
public class MQSender { 
    @Autowired
    AmqpTemplate amqpTemplate;
 public void sendTopic(Object message){ 
        String msg=RedisService.beanToString(message);
        log.info("send topic message: {}",msg);
        amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key1",msg+" 1");
        amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key2",msg+" 2");
    }
}

消费者：

只需要监听需要的队列即可

@Service
@Slf4j
public class MQReceiver { 
    @Autowired
    AmqpTemplate amqpTemplate;
 /** * topic 交换机模式下的receiver */
    @RabbitListener( queues = MQConfig.TOPIC_QUEUE1)
    public void  receiveTopic1(String message){ 
        log.info("receiveTopic queue1 :{}",message);
    }
    @RabbitListener( queues = MQConfig.TOPIC_QUEUE2)
    public void  receiveTopic2(String message){ 
        log.info("receiveTopic queue2 :{}",message);
    }
}

过程：

生产者->message+key1->Exchange->queue(key1)->消费者

交换机模式2:Fanout模式：

广播模式，不指定key，通过交换机将消息发给与该交换机绑定的队列

配置：

@Configuration
public class MQConfig { 
    public static final String QUEUE="queue";
    public static final String TOPIC_QUEUE1="topic.queue1";
    public static final String TOPIC_QUEUE2="topic.queue2";
    public static final String FANOUT_EXCHANGE="fanoutExchange";
/** * 3, Fanout 模式 （广播模式） */
    @Bean
    public FanoutExchange fanoutExchange(){ 
        return new FanoutExchange(FANOUT_EXCHANGE);
    }
    /** * 广播模式绑定 */
    @Bean
    public Binding fanoutBinding1(){ 
        return BindingBuilder.bind(topicQueue1()).to(fanoutExchange());
    }
    @Bean
    public Binding fanoutBinding2(){ 
        return BindingBuilder.bind(topicQueue2()).to(fanoutExchange());
    }
}

生产者：

@Service
@Slf4j
public class MQSender { 
    @Autowired
    AmqpTemplate amqpTemplate;
  public void sendFanout(Object message){ 
        String msg=RedisService.beanToString(message);
        log.info("send FANOUT message: {}",msg);
        amqpTemplate.convertAndSend(MQConfig.FANOUT_EXCHANGE,"",msg+" FANOUT 1");
    }
}

消费者：

同topic模式下的消费者

特点：

无需key，发送的是一条消息，但是接受这条小东西的队列一般是多个（广播）

交换机模式3：Headers模式，

匹配要求很多（通过一个map的key和value逐个与消息发送时设置的properties相比对，有wehereAll，whereAny,where 等等）

配置：

@Configuration
public class MQConfig { 
    public  static final String HEADER_QUEUE="header.queue";
    public static final String  HEADER_EXCHANGE="headerExchange";
/** * 4 header 模式 */
    @Bean
    public HeadersExchange headersExchange(){ 
        return new HeadersExchange(HEADER_EXCHANGE);
    }
    @Bean
    public Queue headerQueue(){ 
        return new Queue(HEADER_QUEUE,true);
    }
    @Bean
    public Binding headerBinding(){ 
        Map<String,Object> map=new HashMap<>();
        map.put("header1","value1");
        map.put("header2","value2");
        //满足符合这个map里面的数据才能放入队列
        return BindingBuilder.bind(headerQueue()).to(headersExchange()).whereAll(map).match();
    }
}

生产者：

@Service
@Slf4j
public class MQSender { 
    @Autowired
    AmqpTemplate amqpTemplate;
public void sendHeader(Object message){ 
        String msg=RedisService.beanToString(message);
        log.info("send HEADERS message: {}",msg);
        MessageProperties properties=new MessageProperties();
        properties.setHeader("header1","value1");
        properties.setHeader("header2","value2");
        Message obj=new Message(msg.getBytes(),properties);
        amqpTemplate.convertAndSend(MQConfig.HEADER_EXCHANGE,"",obj);
    }
}

消费者：

比较特别，发送和接收都是byte[]

@Service
    @Slf4j
    public class MQReceiver { 
        @Autowired
        AmqpTemplate amqpTemplate;
    /** * headers 模式下的receiver */
        @RabbitListener(queues = MQConfig.HEADER_QUEUE)
        public void receiveHeader(byte [] message){ 
            String msg=new String(message);
            log.info("header receiver :{}",msg);
        }
    }