RabbitMQ 使用之四种交换机模式

柔情只为你懂 2022-05-21 09:54 171阅读 0赞

#### 一、集成RabbitMQ ####

*  . 添加依赖

<dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>

*  . 添加配置

#RabbitMQ
    spring.rabbitmq.host=39.106.128.50
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    spring.rabbitmq.virtual-host=/
    #消费者数量
    spring.rabbitmq.listener.simple.concurrency=10
    spring.rabbitmq.listener.simple.max-concurrency=10
    #消费者每次从队列中获取的消息数量
    spring.rabbitmq.listener.simple.prefetch=1
    #消费者自动启动
    spring.rabbitmq.listener.simple.auto-startup=true
    #消费失败，自动重新入队
    spring.rabbitmq.listener.simple.default-requeue-rejected=true
    #启用发送重试
    spring.rabbitmq.template.retry.enabled=true
    spring.rabbitmq.template.retry.initial-interval=1000
    spring.rabbitmq.template.retry.max-attempts=3
    spring.rabbitmq.template.retry.max-interval=10000
    spring.rabbitmq.template.retry.multiplier=1.0

> 注意使用guest用户远程访问RabbitMQ，如果访问失败，需要在远程的RabbitMQ中进行如下配置：  
> 在 /usr/local/rabbitmq/rabbitmq\_server-3.7.4/etc/rabbitmq/ 目录下，新建rabbitmq.config文件，里面写入：“\[\{rabbit, \[\{loopback\_users, \[\]\}\]\}\]. ”即可，不用加双引号。

#### 二、RabbitMQ的使用 ####

*  RabbitMQ 有四种交换机模式：
    
     *  Direct Pattern （此模式不需要配置交换机）
     *  Fanout Pattern ( 类似于广播一样，将消息发送给和他绑定的队列 )
     *  Topic Pattern ( 绑定交换机时可以做匹配。 \#：表示零个或多个单词。\*：表示一个单词 )
     *  Header Pattern ( 带有参数的匹配规则 )
 *  详细举例如下：

/** * 发送方 */
    @Service
    public class MQSender { 
    
        private static final Logger log = LoggerFactory.getLogger(MQSender.class);
    
        @Autowired
        AmqpTemplate amqpTemplate;
    
        /** * use Direct Pattern. RabbitMQ default, no need exchange. * @param message */
        public void sendDirect(Object message){
            String msg = RedisService.BeanToString(message);
            log.info("send direct message:" + msg );
            amqpTemplate.convertAndSend(MQConfig.DIRECT_QUEUE,msg);
        }
    
        /** * use Direct Pattern * @param message */
        public void sendFanout(Object message){
            String msg = RedisService.BeanToString(message);
            log.info("send fanout message:" + msg );
            amqpTemplate.convertAndSend(MQConfig.FANOUT_EXCHANGE,"",msg);
        }
    
        /** * use Topic Pattern * @param message */
        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");  // 可以匹配到 topic.# and topic.key1
            amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key2",msg+"2");  // 可以匹配到 topic.#
        }
    
        /** * use Header Pattern * @param message */
        public void sendHeader(Object message){
            String msg = RedisService.BeanToString(message);
            log.info("send Header message:" + msg );
    
            MessageProperties mp = new MessageProperties();
            mp.setHeader("header1","value1");
            mp.setHeader("header2","value2");   // 与 配置中的map完全匹配，因为那边是 whereAll()方法
            Message message1 = new Message(msg.getBytes(),mp);
            amqpTemplate.convertAndSend(MQConfig.HEADER_EXCHANGE,"",message1);  // 可以匹配到 topic.# and topic.key1
        }

/** * 配置中心 */ 
     @Configuration
    public class MQConfig { 
    
        public static final String DIRECT_QUEUE = "direct.queue";
    
        public static final String FANOUT_QUEUE1 = "fanout.queue1";
        public static final String FANOUT_QUEUE2 = "fanout.queue2";
        public static final String FANOUT_EXCHANGE = "fanout.exchange";
    
        public static final String TOPIC_QUEUE1 = "topic.queue1";
        public static final String TOPIC_QUEUE2 = "topic.queue2";
        public static final String TOPIC_EXCHANGE = "topic.exchange";
        public static final String TOPIC_KEY1 = "topic.key1";
        public static final String TOPIC_KEY2 = "topic.#";
    
        public static final String HEADER_QUEUE = "header.queue";
        public static final String HEADER_EXCHANGE = "header.exchange";
    
        /** * Direct 模式 交换机Exchange */
        @Bean
        public Queue directQueue(){
            // 一个参数是名称，，另一个表示是否持久化
            return new Queue(DIRECT_QUEUE,true);
        }
    
        /**---------------------------------------------*/
    
        /** * Fanout Pattern. 类似于广播一样，将消息发送给和他绑定的队列 **/
        @Bean
        public Queue fanoutQueue1(){
            return new Queue(FANOUT_QUEUE1,true);
        }
        @Bean
        public Queue fanoutQueue2(){
            return new Queue(FANOUT_QUEUE2,true);
        }
        @Bean
        public FanoutExchange fanoutExchange(){
            return new FanoutExchange(FANOUT_EXCHANGE);
        }
        /** * 绑定 exchange and queue */
        @Bean
        public Binding fanoutBinding1(){
            return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
        }
        @Bean
        public Binding fanoutBinding2(){
            return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
        }
    
        /**---------------------------------------------*/
    
        /** * Topic Pattern. 绑定交换机时可以做匹配。 #：表示零个或多个单词。*：表示一个单词 **/
        @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);
        }
        /** * 绑定 exchange and queue */
        @Bean
        public Binding topicBinding1(){
            return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with(TOPIC_KEY1);  // 精确匹配, 匹配成功则发送到 TOPIC_QUEUE1队列
        }
        @Bean
        public Binding topicBinding2(){
            return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with(TOPIC_KEY2);  // 模糊匹配，匹配成功则发送到 TOPIC_QUEUE2队列
        }
    
        /**---------------------------------------------*/
    
        /** * Header Pattern. 交换机 Exchange **/
        @Bean
        public Queue headerQueue(){
            return new Queue(HEADER_QUEUE,true);
        }
        @Bean
        public HeadersExchange headersExchange(){
            return new HeadersExchange(HEADER_EXCHANGE);
        }
        @Bean
        public Binding headerBinding(){
            Map<String,Object> map = new HashMap<>();
            map.put("header1","value1");
            map.put("header2","value2");
            return BindingBuilder.bind(headerQueue()).to(headersExchange()).whereAll(map).match();   // whereXxx() 方法代表了匹配规则
    
        }
    }

/** * 消费者 */
    @Service
    public class MQReceiver { 
    
        private static final Logger log = LoggerFactory.getLogger(MQReceiver.class);
    
        /** * Direct Pattern. * @param message */
        @RabbitListener(queues = MQConfig.DIRECT_QUEUE)
        public void directReceive(String message){
            log.info("receive direct message:" + message);
        }
    
        /** ----------------------------- */
    
        /** * fanout Pattern. * @param message */
        @RabbitListener(queues = MQConfig.FANOUT_QUEUE1)
        public void fanoutReceive1(String message){
            log.info("receive fanout1 message:" + message);
        }
        /** * fanout Pattern. * @param message */
        @RabbitListener(queues = MQConfig.FANOUT_QUEUE2)
        public void fanoutReceive2(String message){
            log.info("receive fanout2 message:" + message);
        }
    
        /** ----------------------------- */
    
        /** * topic Pattern. * @param message */
        @RabbitListener(queues = MQConfig.TOPIC_QUEUE1)
        public void topicReceive1(String message){
            log.info("receive topic1 message:" + message);
        }
        /** * topic Pattern. * @param message */
        @RabbitListener(queues = MQConfig.TOPIC_QUEUE2)
        public void topicReceive2(String message){
            log.info("receive topic2 message:" + message);
        }
    
        /** ----------------------------- */
    
        /** * Header Pattern. * @param message */
        @RabbitListener(queues = MQConfig.HEADER_QUEUE)
        public void topicHeader(byte[] message){
            log.info("receive topic1 message:" + new String(message));
        }
    
    }