一. 重復(fù)消息

為什么會(huì)出現(xiàn)消息重復(fù)？消息重復(fù)的原因有兩個(gè)：1.生產(chǎn)時(shí)消息重復(fù)，2.消費(fèi)時(shí)消息重復(fù)。

1.1 生產(chǎn)時(shí)消息重復(fù)

由于生產(chǎn)者發(fā)送消息給MQ，在MQ確認(rèn)的時(shí)候出現(xiàn)了網(wǎng)絡(luò)波動(dòng)，生產(chǎn)者沒有收到確認(rèn)，實(shí)際上MQ已經(jīng)接收到了消息。這時(shí)候生產(chǎn)者就會(huì)重新發(fā)送一遍這條消息。

生產(chǎn)者中如果消息未被確認(rèn)，或確認(rèn)失敗，我們可以使用定時(shí)任務(wù)+(redis/db)來進(jìn)行消息重試。

@Component@Slf4Jpublic 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í)行定時(shí)任務(wù)(重新投遞消息)"); 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("定時(shí)任務(wù)執(zhí)行結(jié)束(重新投遞消息)"); }}

1.2 消費(fèi)時(shí)消息重復(fù)

消費(fèi)者消費(fèi)成功后，再給MQ確認(rèn)的時(shí)候出現(xiàn)了網(wǎng)絡(luò)波動(dòng)，MQ沒有接收到確認(rèn)，為了保證消息被消費(fèi)，MQ就會(huì)繼續(xù)給消費(fèi)者投遞之前的消息。這時(shí)候消費(fèi)者就接收到了兩條一樣的消息。

修改消費(fèi)者，模擬異常

@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 # 開啟消費(fèi)者進(jìn)行重試 max-attempts: 5 # 最大重試次數(shù) initial-interval: 3000 # 重試時(shí)間間隔

由于重復(fù)消息是由于網(wǎng)絡(luò)原因造成的，因此不可避免重復(fù)消息。但是我們需要保證消息的冪等性。

二. 如何保證消息冪等性

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

消費(fèi)者獲取到消息后先根據(jù)id去查詢r(jià)edis/db是否存在該消息

如果不存在，則正常消費(fèi)，消費(fèi)完畢后寫入redis/db

如果存在，則證明消息被消費(fèi)過，直接丟棄。

生產(chǎn)者

@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);}

消費(fèi)者

@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的話，可以直接將這個(gè)ID設(shè)為消息的主鍵，下次如果獲取到重復(fù)消息進(jìn)行消費(fèi)時(shí)，由于數(shù)據(jù)庫(kù)主鍵的唯一性，則會(huì)直接拋出異常。

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。