《RabbitMQ》如何保證消息不被重復消費

一 重復消息

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

1.1 生產時消息重復

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

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

@Component
@Slf4J
public?class?SendMessage?{
????@Autowired
????private?MessageService?messageService;

????@Autowired
????private?RabbitTemplate?rabbitTemplate;

????//?最大投遞次數(shù)
????private?static?final?int?MAX_TRY_COUNT?=?3;

????/**
?????*?每30s拉取投遞失敗的消息,?重新投遞
?????*/
????@Scheduled(cron?=?"0/30?*?*?*?*??")
????public?void?resend()?{
????????log.info("開始執(zhí)行定時任務(重新投遞消息)");

????????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("超過最大重試次數(shù),?消息投遞失敗,?msgId:?{}",?msgId);
????????????}?else?{
????????????????messageService.updateTryCount(msgId,?msgLog.getNextTryTime());//?投遞次數(shù)+1

????????????????CorrelationData?correlationData?=?new?CorrelationData(msgId);
????????????????rabbitTemplate.convertAndSend(msgLog.getExchange(),?msgLog.getRoutingKey(),?MessageHelper.objToMsg(msgLog.getMsg()),?correlationData);//?重新投遞

????????????????log.info("第?"?+?(msgLog.getTryCount()?+?1)?+?"?次重新投遞消息");
????????????}
????????});

????????log.info("定時任務執(zhí)行結束(重新投遞消息)");
????}
}

1.2消費時消息重復

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

修改消費者，模擬異常

@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?#?最大重試次數(shù)
??????????initial-interval:?3000?#?重試時間間隔

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

二 如何保證消息冪等性

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

1. 消費者獲取到消息后先根據(jù)id去查詢redis/db是否存在該消息。

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

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

生產者

@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設為消息的主鍵，下次如果獲取到重復消息進行消費時，由于數(shù)據(jù)庫主鍵的唯一性，則會直接拋出異常。

< END >

往期精選?《RabbitMQ》如何保證消息的可靠性??一文搞懂TCP和UDP的區(qū)別??程序員接私活的19個平臺??Spring AOP實現(xiàn)原理??Spring IOC實現(xiàn)原理??Nginx超簡單教程