关于MQ消费者的幂等性问题，在于MQ的重试机制，因为网络原因或客户端延迟消费导致重复消费。使用MQ重试机制需要注意的事项以及如何解决消费者幂等性问题以下将逐一讲解。
消费者在消费消息的时候，如果消费者业务逻辑出现程序异常，这个时候我们如何处理？
使用重试机制，RabbitMQ默认开启重试机制。
实现原理：
注意：
spring:
rabbitmq:
# 连接地址
host: 127.0.0.1
# 端口号
port: 5672
# 账号
username: guest
# 密码
password: guest
# 地址(类似于数据库的概念)
virtual-host: /admin_vhost
# 消费者监听相关配置
listener:
simple:
retry:
# 开启消费者(程序出现异常)重试机制，默认开启并一直重试
enabled: true
# 最大重试次数
max-attempts: 5
# 重试间隔时间(毫秒)
initial-interval: 3000
情况1: 消费者获取到消息后，调用第三方接口，但接口暂时无法访问，是否需要重试? 需要重试，可能是因为网络原因短暂不能访问
情况2: 消费者获取到消息后，抛出数据转换异常，是否需要重试? 不需要重试,因为属于程序bug需要重新发布版本
总结：对于情况2，如果消费者代码抛出异常是需要发布新版本才能解决的问题，那么不需要重试，重试也无济于事。应该采用日志记录+定时任务job进行健康检查+人工进行补偿
我们知道了，RabbitMQ在消费者消费发生异常时，会自动进行补偿机制，所以我们（消费者）在调用第三方接口时，可以根据返回结果判断是否成功：
MQ重试机制需要注意的问题)产生原因:网络延迟传输中，消费者出现异常或者消费者延迟消费，会造成进行MQ重试补偿，在重试过程中，可能会造成重复消费。
伪代码：
生产者核心代码:
请求头设置消息id（messageId）
@Component
public class FanoutProducer {
@Autowired
private AmqpTemplate amqpTemplate;
public void send(String queueName) {
String msg = "my_fanout_msg:" + System.currentTimeMillis();
//请求头设置消息id（messageId）
Message message = MessageBuilder.withBody(msg.getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON)
.setContentEncoding("utf-8").setMessageId(UUID.randomUUID() + "").build();
System.out.println(msg + ":" + msg);
amqpTemplate.convertAndSend(queueName, message);
}
}
消费者核心代码：
@RabbitListener(queues = "fanout_email_queue")
public void process(Message message) throws Exception {
// 获取消息Id
String messageId = message.getMessageProperties().getMessageId();
String msg = new String(message.getBody(), "UTF-8");
//② 判断唯一Id是否被消费，消息消费成功后将id和状态保存在日志表中，我们从（①步骤）表中获取并判断messageId的状态即可
//从redis中获取messageId的value
String value = redisUtils.get(messageId)+"";
if(value.equals("1") ){ //表示已经消费
return; //结束
}
System.out.println("邮件消费者获取生产者消息" + "messageId:" + messageId + ",消息内容:" + msg);
JSONObject jsonObject = JSONObject.parseObject(msg);
// 获取email参数
String email = jsonObject.getString("email");
// 请求地址
String emailUrl = "http://127.0.0.1:8083/sendEmail?email=" + email;
JSONObject result = HttpClientUtils.httpGet(emailUrl);
if (result == null) {
// 因为网络原因,造成无法访问,继续重试
throw new Exception("调用接口失败!");
}
System.out.println("执行结束....");
//① 执行到这里已经消费成功，我们可以修改messageId的状态，并存入日志表(可以存到redis中，key为消息Id、value为状态)
}
ACK)1.修改配置simple下添加 acknowledge-mode: manual：
spring:
rabbitmq:
# 连接地址
host: 127.0.0.1
# 端口号
port: 5672
# 账号
username: guest
# 密码
password: guest
# 地址(类似于数据库的概念)
virtual-host: /admin_vhost
# 消费者监听相关配置
listener:
simple:
retry:
# 开启消费者(程序出现异常)重试机制，默认开启并一直重试
enabled: true
# 最大重试次数
max-attempts: 5
# 重试间隔时间(毫秒)
initial-interval: 3000
# 开启手动ack
acknowledge-mode: manual
2.消费者增加代码:
Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG); 手动ack
channel.basicAck(deliveryTag, false);手动签收
//邮件队列
@Component
public class FanoutEamilConsumer {
@RabbitListener(queues = "fanout_email_queue")
public void process(Message message, @Headers Map<String, Object> headers, Channel channel) throws Exception {
System.out
.println(Thread.currentThread().getName() + ",邮件消费者获取生产者消息msg:" + new String(message.getBody(), "UTF-8")
+ ",messageId:" + message.getMessageProperties().getMessageId());
// 手动ack
Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
// 手动签收
channel.basicAck(deliveryTag, false);
}
}