这个帖子我们就可以使用springboot经典的配置文件的方式来实现消息的确认和失败回调

image.png

消息确认和失败回调 代码示例

配置文件

spring.application.name=rabbit-sample
server.port=63001


#rabbitMQ连接字符串
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host = /
spring.rabbitmq.connection-timeout=15000

spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true

消息发送类，我们在消息发送类中定义了消息的确认方法，消息的失败回调方法：

@Component
public class RabbitSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.out.println("correlationData:"+correlationData);
            System.out.println("ack:"+ack);
            if(!ack){
                System.out.println("异常处理机制：-----");
            }
        }
    };

    final RabbitTemplate.ReturnsCallback returnsCallback = new RabbitTemplate.ReturnsCallback() {
        @Override
        public void returnedMessage(ReturnedMessage returned) {
            System.out.println("return:"+returned);
        }
    };

    public void send(Object message, Map<String,Object> properties) throws Exception{
        MessageHeaders messageHeaders = new MessageHeaders(properties);
        Message<Object> msg = MessageBuilder.createMessage(message,messageHeaders);
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnsCallback(returnsCallback);

        // CorrelationData cd = new CorrelationData();
        // cd.setId("444555666");//全局唯一id，可以加时间戳
        rabbitTemplate.convertAndSend("exchange-boot","rabbit.a",msg);
    }

}

测试类：

@Test
    public void send1() throws Exception{
        Map<String,Object> messageProperties = new HashMap<>();
        messageProperties.put("number",1234);
        messageProperties.put("sendTime",new Date());
        rabbitSender.send("this is a msg!",messageProperties);
    }

消费端配置

image.png

消费端@RabbitListener注解使用

在前面的基础上加上下面配置：

spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.concurrency = 5
spring.rabbitmq.listener.simple.max-concurrency = 10

创建一个消息接收类：

@Component
public class RabbitReceiver {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RabbitListener(
             bindings = @QueueBinding(
                     value = @Queue(value = "queue-boot",durable = "true"),
                     exchange = @Exchange(value="exchange-boot",durable = "true",
                             type="topic",ignoreDeclarationExceptions ="true"),
                     key = "spring.*"
             )
    )
    @RabbitHandler
    public void onMessage(Message message, Channel channel) throws Exception{
        System.out.println("-------");
        System.out.println("消费端payload："+message.getPayload());
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        //手动ACK
        channel.basicAck(deliveryTag,false);
    }
}

可以看到我们使用注解的方式创建了一个exchange，一个队列，然后还建立了绑定关系，然后注解的方法就监听这个队列，接收消息并手动ack。

使用配置文件配置exchange以及队列等信息

直接在代码中写死是不灵活的方式，所以我们在配置文件写好，然后以变量的方式注入到rabbit的注解中以达到创建队列的方式。

在接收端我们也可以直接接收java对象，这种方式更加便捷：