Rabbit @RabbitListener消息队列使用

水深无声 2023-01-12 15:52 202阅读 0赞

@RabbitListener(bindings = @QueueBinding(
                value = @Queue(value = queueName, durable = "true", ignoreDeclarationExceptions = "true"),
                exchange = @Exchange(value = EventConfig.exchangeGR, type = ExchangeTypes.FANOUT, durable = "true"),
                key = "orderRoutingKey", ignoreDeclarationExceptions = "true"))
        public void save(@Payload GoodsReciptEvent event, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag){ 
            log.info("SaveEvent: " + event.toString());
            try { 
                channel.basicAck(deliveryTag, false);
            }catch (Exception e){ 
            
                log.info("event json -->"+JSONObject.toJSONString(event));
                try { 
                    channel.basicAck(deliveryTag, false);
                }catch (Exception ee){ 
                    log.error("save failed ！",e);
                }
    
            }

*上面是项目java代码，只要rabbit环境已经安装好，依赖的jar包spring-rabbit-1.7.3.RELEASE.jar也引入进来，就可以正常发送接收消息。 有几个注意点，说明一下。*

**1、exchange = @Exchange(value = EventConfig.exchangeGR,  
表示接收的交换器名称是EventConfig.exchangeGR，自己根据项目情况灵活配置上。  
2、value = @Queue(value = queueName  
表示接收的消息队列名称是queueName，自己根据项目情况灵活配置上。  
3、type = ExchangeTypes.FANOUT  
表示接收消息的类型，分为常用的三种FANOUT，广播模式，只要连接上，消息就全部通过接收到；DIRECT，直连模式，表示消息单独发；TOPIC，订阅模式，根据key匹配规则，匹配符合规则的消息数据  
4、key=xxx，表示接收消息时候的key，TOPIC模式必须配置上  
5、消息接收消费后，一定要channel.basicAck(deliveryTag, false)删除消息。**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODk4MTkx_size_16_color_FFFFFF_t_70]

*上面截图是项目中rabbit消息队列的网页登录信息，可以看见当前消息是否阻塞，正常状态（Ready）的消息数量，阻塞的状态（Ucacked）数量。  
注意：一般接收消息端，出现消息阻塞，服务重启后，阻塞消息就释放到正常状态，也可以根据情况，手动在界面上删除QUE。*

实战模拟消息分组

**1. 发送方：**

1
        /**
         * 模拟消息分组 发送方
         */
        public void sendOrder(){ 
            String msg = "mqsender send ..." + new Date();
            // 参数：交换机，路由key, 消息
            amqpTemplate.convertAndSend("myOrder","computer", msg);
        }

**2. 接收方：**

/**----------- 模拟消息分组 --------------------*/
        /**
         * 数码供应商服务  接收消息
         * 消息发到交换机，交换机根据不同的key 发送到不同的队列
         */
        @RabbitListener(bindings = @QueueBinding(
                exchange = @Exchange("myOrder"),
                key = "computer",
                value = @Queue("computerOrder")
        ))
        public void receiveComputer(String msg){ 
            log.info(" receiveComputer service = {}" , msg );
        }
        /**
         * 水果供应商服务  接收消息
         */
        @RabbitListener(bindings = @QueueBinding(
                value = @Queue("fruitOrder"),
                key = "fruit",
                exchange = @Exchange("myOrder")
        ))
        public void receiveFruit(String msg){ 
            log.info(" receiveFruit service = {}" , msg );
        }

**3. 测试用例**

@Autowired
    private MQSender sender;
    
    @Test
    public void sendOrderTest() { 
        sender.sendOrder();
    }

**4. 结果：**  
*消息发送到交换机，交换机通过路由key 发送到对应的队列。  
因此computerOrder队列得到了消息，进而receiveComputer()接收到了消息。*

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODk4MTkx_size_16_color_FFFFFF_t_70]: /images/20221022/8b9bd97ceb114f7f83674bed4282c5cb.png