rabbitmq注意事项

发送各种类型的消息

如果是以字节数组的形式包装成Message,那么接收的时候是用byte[]的形式接收，之后再转换成对应的类型
以byte[]形式发送：

    // String类型Message msg=MessageBuilder.withBody(objectMapper.writeValueAsBytes(message)).build();rabbitTemplate.send(msg);// pojoMessage msg=MessageBuilder.withBody(objectMapper.writeValueAsBytes(user)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();rabbitTemplate.convertAndSend(msg);// Map类型Message message=MessageBuilder.withBody(objectMapper.writeValueAsBytes(dataMap)).build();rabbitTemplate.convertAndSend(message);

以byte[]形式接收

    @RabbitListener(queues = "${basic.info.mq.queue.name}",containerFactory = "singleListenerContainer")public void consumeMessage(@Payload byte[] message){try {//TODO：接收StringString result=new String(message,"UTF-8");log.info("接收到消息： {} ",result);//TODO：接收对象User user=objectMapper.readValue(message, User.class);log.info("接收到消息： {} ",user);//TODO：接收多类型字段信息Map<String,Object> resMap=objectMapper.readValue(message,Map.class);log.info("接收到消息： {} ",resMap);}catch (Exception e){log.error("监听消费消息 发生异常： ",e.fillInStackTrace());}}

如果发送的时候不是发送的message,而是发送的对象，那么接收的时候直接用相应的对象接收就可以。

以对象的形式发送

 // 直接发送字符串格式rabbitTemplate.convertAndSend(str);// 直接发送pojorabbitTemplate.convertAndSend(user);// 直接发送maprabbitTemplate.convertAndSend(message);

以对象的形式接收

    @RabbitListener(queues = "${basic.info.mq.queue.name}",containerFactory = "singleListenerContainer")public void consumeMessage(Map<String,Object> resMap){try {//Map<String,Object> resMap=objectMapper.readValue(message,Map.class);log.info("接收到消息： {} ",resMap);}catch (Exception e){log.error("监听消费消息 发生异常： ",e.fillInStackTrace());}}

    @RabbitListener(queues = "${basic.info.mq.queue.name}",containerFactory = "singleListenerContainer")public void consumeMessage(@Payload User user){try {log.info("接收到实体消息： {} ",user);}catch (Exception e){log.error("监听消费消息 发生异常： ",e.fillInStackTrace());}}

注意：此时接收的都是消息体的数据，如果想要接收消息头中的信息

可以直接接收Message对象，从中取body和Properties.

    //支持自动声明绑定，声明之后自动监听队列的队列，此时@RabbitListener注解的queue和bindings不能同时指定，否则报错@RabbitListener(bindings ={@QueueBinding(value = @Queue(value = "q5",durable = "true"),exchange =@Exchange(value = "zhihao.miao.exchange",durable = "true"),key = "welcome")})public void handleMessage(Message message){System.out.println("====消费消息"+message.getMessageProperties().getConsumerQueue()+"===handleMessage");System.out.println(message.getMessageProperties());System.out.println(new String(message.getBody()));}

使用注解@Payload取的是消息体的信息， @Header取的是消息头的信息

    //获取消息的头属性和body属性@RabbitListener(queues = "zhihao.miao.order")public void handleMessage(@Payload String body, @Headers Map<String,Object> headers){System.out.println("====消费消息===handleMessage");System.out.println(headers);System.out.println(body);}

    //获取特定的消息@RabbitListener(queues = "zhihao.miao.order")public void handleMessage(@Payload String body,@Header String token){System.out.println("====消费消息===handleMessage");System.out.println(token);System.out.println(body);}

关于消息转换的问题可以参考RabbitMQ笔记十三：使用@RabbitListener注解消费消息

发送消息的回馈

发送消息的消息反馈可以参考RabbitMQ：消息发送确认与消息接收确认（ACK）

    @Beanpublic RabbitTemplate rabbitTemplate(){connectionFactory.setPublisherConfirms(true);connectionFactory.setPublisherReturns(true);RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);rabbitTemplate.setMandatory(true);rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {log.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);}});rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);}});return rabbitTemplate;}

如果Mandatory为false，找不到路由的队列，会直接丢弃消息，除了设置为true,如果路由失败，会将消息返回的方式外，还有一个方式是给目的队列设置为一个备份交换机，没有路由的消息直接发送给这个备份的就可以，详细参考：springboot整合rabbirmq(3.7.9)中使用mandatory参数获取匹配失败的消息以及存入rabbitmq备份交换器中!

多线程处理消息

默认情况下，消费者一次只能消费一个消息，为了提高性能，一个消费者可以同时消费多条消息，setConcurrentConsumers，setMaxConcurrentConsumers，setPrefetchCount，设置三个属性，

    @Bean(name = "singleListenerContainer")public SimpleRabbitListenerContainerFactory listenerContainer(){SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setMessageConverter(new Jackson2JsonMessageConverter());factory.setConcurrentConsumers(1);factory.setMaxConcurrentConsumers(1);factory.setPrefetchCount(1);factory.setTxSize(1);factory.setAcknowledgeMode(AcknowledgeMode.AUTO);return factory;}

    @Bean(name = "multiListenerContainer")public SimpleRabbitListenerContainerFactory multiListenerContainer(){SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factoryConfigurer.configure(factory,connectionFactory);factory.setMessageConverter(new Jackson2JsonMessageConverter());factory.setAcknowledgeMode(AcknowledgeMode.NONE);factory.setConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.concurrency",int.class));factory.setMaxConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.max-concurrency",int.class));factory.setPrefetchCount(env.getProperty("spring.rabbitmq.listener.prefetch",int.class));return factory;}

接收消息的时候，设置使用的factory，设置多线程的containerFactory。

    @RabbitListener(queues = "${mail.queue.name}",containerFactory = "multiListenerContainer")public void consumeMailQueue(@Payload byte[] message){try {log.info("监听消费邮件发送 监听到消息： {} ",new String(message,"UTF-8"));mailService.sendEmail();}catch (Exception e){e.printStackTrace();}}

消息的确认机制

可以参考文章RabbitMQ：消息发送确认与消息接收确认（ACK）

项目实战

本文参考SpringBoot整合RabbitMQ之典型应用场景实战一