應用場景

• 用戶下單5分鐘後,給他發簡訊
• 用戶下單30分鐘後，如果用戶不付款就自動取消訂單

kafka無死信隊列

kafka本身沒有這種延時隊列的機制，像rabbitmq有自己的死信隊列，當一些消息在一定時間不消費時會發到死信隊列，由死信隊列來處理它們，上面的兩個需求如果是rabbitmq可以通過死信隊列實現的。

kafka有生產者攔截器

通過對生產者攔截器實現一個TTL的檢查，然後再通過類似netty里的延時隊列組件來實現消息的延時發送，發到咱們的死信隊列里

• ProducerInterceptorTTL源碼

public class ProducerInterceptorTTL implements ProducerInterceptor<Integer, String>, ApplicationContextAware {

	// 消息延時,單位秒
	public static String TTL = "ttl";

	// 死信隊列,延時後發送到的隊列，我們稱為死信隊列
	public static String DEAD_TOPIC = "dead_topic";

	// 靜態化的上下文，用於獲取bean，因為ConsumerInterceptor是通過反射創建的，所以無法通過註入的方式獲取bean
	private static ApplicationContext applicationContext;

	// 時間輪，用於延時發送消息
	private static LindTimeWheel timeWheel = new LindTimeWheel(1000, 8);

	@Override
	public ProducerRecord onSend(ProducerRecord<Integer, String> record) {
		final String topic = record.topic();
		final Integer partition = record.partition();
		final Integer key = record.key();
		final String value = record.value();
		final Long timestamp = record.timestamp();
		final Headers headers = record.headers();
		long ttl = -1;
		String deadTopic = null;
		for (Header header : headers) {
			if (header.key().equals(TTL)) {
				ttl = toLong(header.value());
			}
			if (header.key().equals(DEAD_TOPIC)) {
				deadTopic = new String(header.value());
			}
		}
		// 消息超時判定
		if (deadTopic != null && ttl > 0) {
			// 可以放在死信隊列中
			String finalDeadTopic = deadTopic;
			long finalTtl = ttl * 1000;
			timeWheel.addTask(() -> {
				System.out.println("消息超時了," + finalTtl + "需要發到topic:" + record.key());
				KafkaTemplate kafkaTemplate = applicationContext.getBean(KafkaTemplate.class);
				kafkaTemplate.send(finalDeadTopic, record.value());

			}, finalTtl);
		}
		// 攔截器攔下來之後改變原來的消息內容
		ProducerRecord<Integer, String> newRecord = new ProducerRecord<Integer, String>(topic, partition, timestamp,
				key, value, headers);
		// 傳遞新的消息
		return newRecord;

	}

	@Override
	public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

	}

	@Override
	public void close() {

	}

	@Override
	public void configure(Map<String, ?> map) {

	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		this.applicationContext = applicationContext;
	}

}

• 註冊攔截器

spring:
  kafka:
    producer:
      properties:
        interceptor.classes: com.ruoyi.lawyer.delay.ProducerInterceptorTTL

• 延時消息在某個時間段之後會送出
  

作者：倉儲大叔，張占嶺，
榮譽：微軟MVP
QQ：853066980

支付寶掃一掃，為大叔打賞!