Spring Boot整合RabbitMQ之注解实现路由模式（Direct）

上一篇博客Spring Boot整合RabbitMQ之路由模式（Direct）中介绍了通过配置类实现队列和交换机的绑定关系。但是springboot整合rabbitmq的绑定关系也可以使用注解的方式来实现，需要知道的是，如果使用注解实现绑定关系，那么注解绑定的逻辑需要写在消费者上。我们通过代码来实现一下。

配置类中的绑定关系删除，配置类详情如下

package com.study.rabbitmq.config;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Slf4j
@Configuration
public class RabbitMQConfig {

    

    
    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //设置开启消息推送结果回调
        rabbitTemplate.setMandatory(true);
        //设置ConfirmCallback回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("==============ConfirmCallback start ===============");
                log.info("回调数据：{}", correlationData);
                log.info("确认结果：{}", ack);
                log.info("返回原因：{}", cause);
                log.info("==============ConfirmCallback end =================");
            }
        });
        //设置ReturnCallback回调
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.info("==============ReturnCallback start ===============");
                log.info("发送消息：{}", JSONUtil.toJsonStr(message));
                log.info("结果状态码：{}", replyCode);
                log.info("结果状态信息：{}", replyText);
                log.info("交换机：{}", exchange);
                log.info("路由key：{}", routingKey);
                log.info("==============ReturnCallback end =================");
            }
        });
        return rabbitTemplate;
    }
}

在消费者的消息监听类中的@RabbitListener中进行队列与交换机的绑定，代码详情如下

package com.study.rabbitmq.service.direct;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Service;

import java.io.IOException;


@Slf4j
@Service
//监听队列 queues = {"email-direct-queue"}
@RabbitListener(bindings = @QueueBinding(
        exchange = @Exchange(name = "direct-order-exchange", durable = "true", type = "direct"),
        value = @Queue(value = "email-direct-queue", durable = "true"),
        key = "email"
))
public class DirectEmailConsumer {

    @RabbitHandler
    public void emailMessage(String msg, Channel channel, Message message) throws IOException {

        try {
            log.info("Email direct --接收到消息：{}", msg);

            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            if (message.getMessageProperties().getRedelivered()) {
                log.error("消息已重复处理失败,拒绝再次接收...");
                //basicReject: 拒绝消息，与basicNack区别在于不能进行批量操作，其他用法很相似 false表示消息不再重新进入队列
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息
            } else {
                log.error("消息即将再次返回队列处理...");
                // basicNack:表示失败确认，一般在消费消息业务异常时用到此方法，可以将消息重新投递入队列
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }
        }
    }
}

测试效果如下

注解实现direct模式的整合过程就是这些。