一. 重復消息

為什么會出現消息重復？消息重復的原因有兩個：1.生產時消息重復，2.消費時消息重復。

1.1 生產時消息重復

由于生產者發送消息給MQ，在MQ確認的時候出現了網絡波動，生產者沒有收到確認，實際上MQ已經接收到了消息。這時候生產者就會重新發送一遍這條消息。

生產者中如果消息未被確認，或確認失敗，我們可以使用定時任務+(redis/db)來進行消息重試。

@Component@Slf4Jpublic class SendMessage { @Autowired private MessageService messageService; @Autowired private RabbitTemplate rabbitTemplate; // 最大投遞次數 private static final int MAX_TRY_COUNT = 3; /** * 每30s拉取投遞失敗的消息, 重新投遞 */ @Scheduled(cron = "0/30 * * * * ?") public void resend() { log.info("開始執行定時任務(重新投遞消息)"); List msgLogs = messageService.selectTimeoutMsg(); msgLogs.forEach(msgLog -> { String msgId = msgLog.getMsgId(); if (msgLog.getTryCount() >= MAX_TRY_COUNT) { messageService.updateStatus(msgId, Constant.MsgLogStatus.DELIVER_FAIL); log.info("超過最大重試次數, 消息投遞失敗, msgId: {}", msgId); } else { messageService.updateTryCount(msgId, msgLog.getNextTryTime());// 投遞次數+1 CorrelationData correlationData = new CorrelationData(msgId); rabbitTemplate.convertAndSend(msgLog.getExchange(), msgLog.getRoutingKey(), MessageHelper.objToMsg(msgLog.getMsg()), correlationData);// 重新投遞 log.info("第 " + (msgLog.getTryCount() + 1) + " 次重新投遞消息"); } }); log.info("定時任務執行結束(重新投遞消息)"); }}

1.2 消費時消息重復

消費者消費成功后，再給MQ確認的時候出現了網絡波動，MQ沒有接收到確認，為了保證消息被消費，MQ就會繼續給消費者投遞之前的消息。這時候消費者就接收到了兩條一樣的消息。

修改消費者，模擬異常

@RabbitListener(queuesToDeclare = @Queue(value = "javatrip", durable = "true"))public void receive(String message, @Headers Map headers, Channel channel) throws Exception{ System.out.println("重試"+System.currentTimeMillis()); System.out.println(message); int i = 1 / 0;}

配置yml重試策略

spring: rabbitmq: listener: simple: retry: enabled: true # 開啟消費者進行重試 max-attempts: 5 # 最大重試次數 initial-interval: 3000 # 重試時間間隔

由于重復消息是由于網絡原因造成的，因此不可避免重復消息。但是我們需要保證消息的冪等性。

二. 如何保證消息冪等性

讓每個消息攜帶一個全局的唯一ID，即可保證消息的冪等性，具體消費過程為：

消費者獲取到消息后先根據id去查詢redis/db是否存在該消息

如果不存在，則正常消費，消費完畢后寫入redis/db

如果存在，則證明消息被消費過，直接丟棄。

生產者

@PostMapping("/send")public void sendMessage(){ JSONObject jsonObject = new JSONObject(); jsonObject.put("message","Java旅途"); String json = jsonObject.toJSONString(); Message message = MessageBuilder.withBody(json.getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("UTF-8").setMessageId(UUID.randomUUID()+"").build(); amqpTemplate.convertAndSend("javatrip",message);}

消費者

@Component@RabbitListener(queuesToDeclare = @Queue(value = "javatrip", durable = "true"))public class Consumer { @RabbitHandler public void receiveMessage(Message message) throws Exception { Jedis jedis = new Jedis("localhost", 6379); String messageId = message.getMessageProperties().getMessageId(); String msg = new String(message.getBody(),"UTF-8"); System.out.println("接收到的消息為："+msg+"==消息id為："+messageId); String messageIdRedis = jedis.get("messageId"); if(messageId == messageIdRedis){ return; } JSONObject jsonObject = JSONObject.parseObject(msg); String email = jsonObject.getString("message"); jedis.set("messageId",messageId); }}

如果需要存入db的話，可以直接將這個ID設為消息的主鍵，下次如果獲取到重復消息進行消費時，由于數據庫主鍵的唯一性，則會直接拋出異常。

總結

以上是生活随笔為你收集整理的rabbitmq如何保证消息不被重复消费_如何保证消息不被重复消费的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。