前一篇文章,寫了訊息傳送確認的一些內容.

就是訊息傳送成功或失敗的時候,都會呼叫confirmListener 或者returnListener.

如果訊息傳送成功,就不考慮了.當訊息傳送失敗時,怎麼處理這個訊息呢.

1.自動重發

2.系統預警人工處理等

以上操作,都需要知道是哪條訊息,具體什麼內容傳送失敗了,才能進行後續處理.

在returnListener中,引數是有訊息內容,exchange,routingKey 這些內容的.

但是在confirmListener中,卻是什麼都沒有,只有個correlationData,而我們打印出的correlationData都是null

看來問題的關鍵就在這個correlationData上了, CorrelationData類只有一個屬性ID, 很明顯,我們在傳送訊息時,將訊息和correlationData的ID做一個繫結,就可以根據id拿到訊息. 然後進行重發,報警等操作了.

到傳送訊息的類裡面, 發現AmqpTemplate 裡面竟然沒有和CorrelationData相關的方法,沒辦法把CorrelationData.id和訊息進行繫結..

原來需要使用RabbitTemplate 而不是 AmqpTemplate 類

找到方法後,下面就非常簡單了.

@Service("rabbitTemplatePublishService")
public class RabbitTemplatePublishService {
	@Autowired
	private RabbitTemplate rmqpTemplate;

	public void send(String exchange, String routingKey, Object obj) {
		String msgId = UUID.randomUUID().toString();
		Message message = MessageBuilder.withBody(obj.toString().getBytes())
				.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
				.setCorrelationId(msgId.getBytes()).build();
		CorrelationData date = new CorrelationData(msgId);
		// TODO 將 msgId 與 message 的關係儲存起來,例如放到快取中
		rmqpTemplate.send(exchange, routingKey, message, date);
	}
}
 

上面的程式碼,message中,也可以setCorrelationId為自己指定的id,這樣方便confirm和return一起處理.

上面將msgId和message關係儲存後, 在confirmListener或者returnListener中,就可以根據msgId獲取message進行對於操作.

上面的思路看起來解決了失敗後訊息處理的問題.但是實際上卻是不可靠的

3.如果進行confirm的時候,或者return的時候, 失敗了怎麼辦?  

4.msgId和message的關係要儲存多久?

但是上面的思路已經抓到了問題的重點,即msgId和message做繫結.

針對3,4兩個問題,只需要變通下即可解決

方案如下:

傳送訊息前,繫結並儲存msgId和message的關係

當confirm或return回撥時,根據ack類別等,分別處理. 例如return或者ack=false則說明有問題,報警, 如果ack=true則刪除關係

(因為return在confirm前,所以一條訊息在return後又ack=true的情況也是按return處理)

定時檢查這個繫結關係列表,如果發現一些已經超時(自己設定的超時時間)未被處理(即未return和confirm),則手動處理這些訊息.

至此,傳送訊息基本已經沒問題了.

需要注意如果是自動重發的話,消費端需要做冪等或去重處理.