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

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

1. 修改配置类

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

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;/*** @Author alen* @DATE 2022/6/7 23:50*/
@Slf4j
@Configuration
public class RabbitMQConfig {/*** 将自定义的RabbitTemplate对象注入bean容器** @param connectionFactory* @return*/@Beanpublic RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate();rabbitTemplate.setConnectionFactory(connectionFactory);//设置开启消息推送结果回调rabbitTemplate.setMandatory(true);//设置ConfirmCallback回调rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic 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() {@Overridepublic 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;}
}

2. 消费者实现绑定关系

在消费者的消息监听类中的@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;/*** @Author alen* @DATE 2022/6/10 22:54*/
@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 {@RabbitHandlerpublic 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模式的整合过程就是这些。 

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！