Java Kafka实现延迟队列的示例代码
作者:整点bug 时间:2022-01-11 16:32:11
kafka作为一个使用广泛的消息队列,很多人都不会陌生,但当你在网上搜索“kafka 延迟队列”,出现的都是一些讲解时间轮或者只是提供了一些思路,并没有一份真实可用的代码实现,今天我们就来打破这个现象,提供一份可运行的代码,抛砖引玉,吸引更多的大神来分享。
基于kafka如何实现延迟队列
想要解决一个问题,我们需要先分解问题。kafka作为一个高性能的消息队列,只要消费能力足够,发出的消息都是会立刻收到的,因此我们需要想一个办法,让消息延迟发送出去。
网上已经有大神给出了如下方案:
在发送延迟消息时不直接发送到目标topic,而是发送到一个用于处理延迟消息的topic,例如
delay-minutes-1
写一段代码拉取
delay-minutes-1
中的消息,将满足条件的消息发送到真正的目标主题里。
就像画一匹马一样简单。
方案是好的,但是我们还需要更多细节。
完善细节
问题出在哪里?
问题出在延迟消息发出去之后,代码程序就会立刻收到延迟消息,要如何处理才能让延迟消息等待一段时间才发送到真正的topic里面。
可能有同学会觉得很简单嘛,在代码程序收到消息之后判断条件不满足,就调用sleep
方法,过了一段时间我再进行下一个循环拉取消息。
真的可行吗?
一切好像都很美好,但这是不可行的。
这是因为在轮询kafka拉取消息的时候,它会返回由max.poll.records
配置指定的一批消息,但是当程序代码不能在max.poll.interval.ms
配置的期望时间内处理这些消息的话,kafka就会认为这个消费者已经挂了,会进行rebalance
,同时你这个消费者就无法再拉取到任何消息了。
举个例子:当你需要一个24小时的延迟消息队列,在代码里面写下了Thread.sleep(1000*60*60*24);
,为了不发生rebalance
,你把max.poll.interval.ms
也改成了1000*60*60*24
,这个时候你或许会感觉到一丝丝的怪异,我是谁?我在哪?我为什么要写出来这样的代码?
其实我们可以更优雅的处理这个问题。
KafkaConsumer 提供了暂停和恢复的API函数,调用消费者的暂停方法后就无法再拉取到新的消息,同时长时间不消费kafka也不会认为这个消费者已经挂掉了。另外为了能够更加优雅,我们会启动一个定时器来替换sleep
。,完整流程如下图,当消费者发现消息不满足条件时,我们就暂停消费者,并把偏移量seek到上一次消费的位置以便等待下一个周期再次消费这条消息。
Java代码实现
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ExecutionException;
@SpringBootTest
public class DelayQueueTest {
private KafkaConsumer<String, String> consumer;
private KafkaProducer<String, String> producer;
private volatile Boolean exit = false;
private final Object lock = new Object();
private final String servers = "";
@BeforeEach
void initConsumer() {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "d");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "5000");
consumer = new KafkaConsumer<>(props, new StringDeserializer(), new StringDeserializer());
}
@BeforeEach
void initProducer() {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producer = new KafkaProducer<>(props);
}
@Test
void testDelayQueue() throws JsonProcessingException, InterruptedException {
String topic = "delay-minutes-1";
List<String> topics = Collections.singletonList(topic);
consumer.subscribe(topics);
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
synchronized (lock) {
consumer.resume(consumer.paused());
lock.notify();
}
}
}, 0, 1000);
do {
synchronized (lock) {
ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(200));
if (consumerRecords.isEmpty()) {
lock.wait();
continue;
}
boolean timed = false;
for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
long timestamp = consumerRecord.timestamp();
TopicPartition topicPartition = new TopicPartition(consumerRecord.topic(), consumerRecord.partition());
if (timestamp + 60 * 1000 < System.currentTimeMillis()) {
String value = consumerRecord.value();
ObjectMapper objectMapper = new ObjectMapper();
JsonNode jsonNode = objectMapper.readTree(value);
JsonNode jsonNodeTopic = jsonNode.get("topic");
String appTopic = null, appKey = null, appValue = null;
if (jsonNodeTopic != null) {
appTopic = jsonNodeTopic.asText();
}
if (appTopic == null) {
continue;
}
JsonNode jsonNodeKey = jsonNode.get("key");
if (jsonNodeKey != null) {
appKey = jsonNode.asText();
}
JsonNode jsonNodeValue = jsonNode.get("value");
if (jsonNodeValue != null) {
appValue = jsonNodeValue.asText();
}
// send to application topic
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(appTopic, appKey, appValue);
try {
producer.send(producerRecord).get();
// success. commit message
OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(consumerRecord.offset() + 1);
HashMap<TopicPartition, OffsetAndMetadata> metadataHashMap = new HashMap<>();
metadataHashMap.put(topicPartition, offsetAndMetadata);
consumer.commitSync(metadataHashMap);
} catch (ExecutionException e) {
consumer.pause(Collections.singletonList(topicPartition));
consumer.seek(topicPartition, consumerRecord.offset());
timed = true;
break;
}
} else {
consumer.pause(Collections.singletonList(topicPartition));
consumer.seek(topicPartition, consumerRecord.offset());
timed = true;
break;
}
}
if (timed) {
lock.wait();
}
}
} while (!exit);
}
}
这段程序是基于SpringBoot 2.4.4
版本和 kafka-client 2.7.0
版本编写的一个单元测试,需要修改私有变量servers
为kafka broker的地址。
在启动程序后,向Topic delay-minutes-
1 发送如以下格式的json字符串数据
{
"topic": "target",
"key": "key1",
"value": "value1"
}
同时启动一个消费者监听topic target
,在一分钟后,将会收到一条 key="key1", value="value1"的数据。
源代码地址
还需要做什么
创建多个topic用于处理不同时间的延迟消息,例如delay-minutes-1
delay-minutes-5
delay-minutes-10
delay-minutes-15
以提供指数级别的延迟时间,这样比一个topic要好很多,毕竟在顺序拉取消息的时候,有一条消息不满足条件,后面的将全部进行排队。
来源:https://zhuanlan.zhihu.com/p/365802989
![](/images/zang.png)
![](/images/jiucuo.png)
猜你喜欢
C#中FileStream的对比及使用方法
![](https://img.aspxhome.com/file/2023/3/91233_0s.jpg)
Java日常练习题,每天进步一点点(41)
C#将隐私信息(银行账户,身份证号码)中间部分特殊字符替换成*
Windows下Java环境变量配置详解
MyBatis-Plus自动填充功能失效导致的原因及解决
![](https://img.aspxhome.com/file/2023/3/68513_0s.jpg)
Android触摸及手势操作GestureDetector
![](https://img.aspxhome.com/file/2023/9/88709_0s.png)
Jenkins安装以及邮件配置详解
![](https://img.aspxhome.com/file/2023/1/63701_0s.png)
android里TextView加下划线的几种方法总结
Android 如何实现动态申请权限
Android 完整购物商城界面的实现案例
![](https://img.aspxhome.com/file/2023/1/84721_0s.png)
应用启动数据初始化接口CommandLineRunner和Application详解
![](https://img.aspxhome.com/file/2023/8/65608_0s.jpg)
JAVA各种OOM代码示例与解决方法
![](https://img.aspxhome.com/file/2023/5/72555_0s.jpg)
Java如何利用状态模式(state pattern)替代if else
![](https://img.aspxhome.com/file/2023/7/87207_0s.jpg)
maven继承父工程统一版本号的实现
![](https://img.aspxhome.com/file/2023/2/102132_0s.png)
使用Postman传递arraylist数据给springboot方式
![](https://img.aspxhome.com/file/2023/9/69359_0s.jpg)
Java try()语句实现try-with-resources异常管理机制操作
![](https://img.aspxhome.com/file/2023/1/78961_0s.png)
SpringBoot整合Mybatis实现多数据源配置与跨数据源事务实例
Java 格式化输出JSON字符串的2种实现操作
Java IO流和文件操作实现过程解析
SpringBoot实现动态多线程并发定时任务
![](https://img.aspxhome.com/file/2023/9/71689_0s.jpg)