背景

当前业务存在以下场景：在一个事务内的最后一步是发送kafka消息，消费端收到通知后读取数据并做处理。但是由于kafka几乎是即时收到消息，导致偶尔出现“在发完kafka和提交事务的间隙，消费端收到了消息并读取到了事务提交前的数据”。

这个问题可以通过延迟消息来解决。

发送端 vs 消费端

要做延迟，那么首先要考虑的是：延迟放在发送端，还是放在消费端？最终选择放在消费端：让数据先被kafka存储起来，数据更安全。

想把延迟消息做成一个服务，不只是支持某一个场景/业务，在这种设计前提下，让延迟逻辑放在消费端，可以统一调整逻辑，也方便排查问题。

思路

是在整体外面包一层代理：另外创建一个延迟Topic，延迟消息都发到延迟Topic里。

有专门的服务来消费延迟Topic的消息，取到消息之后存储起来，定期检查消息是否已经延迟时间。

已到延迟时间的消息，重新发送到原先Topic。

这样做的好处是，不需要对kafka做任何改造。

存储

延迟队列消费者拉取到消息之后，要怎么存储？第三方存储，其需要满足以下几个条件：高性能：写入延迟要低，MQ的一个重要作用是削峰填谷，在选择临时存储时，写入性能必须要高，关系型数据库(如Mysql)通常不满足需求。

高可靠：延迟消息写入后，不能丢失，需要进行持久化，并进行备份

存储成本低：可以支持大量消息存储，(Redis存储成本太高)。

支持排序： 支持按照某个字段对消息进行排序，对于延迟消息需要按照时间进行排序。普通消息通常先发送的会被先消费，延迟消息与普通消息不同，需要进行排序。例如先发一条延迟10s的消息，再发一条延迟5s的消息，那么后发送的消息需要被先消费。

支持长时间保存：一些业务的延迟消息，需要延迟几个月，甚至更长，所以延迟消息必须能长时间保留。不过通常不建议延迟太长时间，存储成本比较大，且业务逻辑可能已经发生变化，已经不需要消费这些消息。

基于以上条件，选择了RocksDB来存储数据：高性能嵌入式KV存储引擎。

数据持久化到磁盘。

基于LMS存储，key自然排序，迭代器(Iterator)根据key顺序遍历。

代码

发送端

消息基类public class DelayMessage {

/**

* 事件唯一ID,用于去重检查

*/

private String eventId = UUIDGenerator.generateString();

/**

* 事件时间

*/

@JSONField(format = KafkaConstants.DATETIME_FORMAT)

private Date eventTime = new Date();

/**

* 真实事件时间

*/

@JSONField(format = KafkaConstants.DATETIME_FORMAT)

private Date actualTime;

/**

* 真实Topic

*/

private String actualTopic;

public Date getActualTime() {

return actualTime;

}

public T setActualTime(Date actualTime) {

this.actualTime = actualTime;

return (T) this;

}

public String getActualTopic() {

return actualTopic;

}

public T setActualTopic(String actualTopic) {

this.actualTopic = actualTopic;

return (T) this;

}

public Date getEventTime() {

return eventTime;

}

public T setEventTime(Date eventTime) {

this.eventTime = eventTime;

return (T) this;

}

}

消息对象继承DelayMessage，将消息发送到延迟Topic。

延迟服务消费端

接收延迟消息@KafkaListener(topics = {KafkaConstants.KAFKA_TOPIC_MESSAGE_DELAY}, containerFactory = "kafkaContainerFactory")

public boolean onMessage(String json) throws Throwable {

try {

DelayMessage delayMessage = deserialize(json, DelayMessage.class);

if (!isDelay(delayMessage)) {

// 如果接收到消息时，消息已经可以发送了，直接发送到实际的队列

sendActualTopic(delayMessage, json);

} else {

// 存储

localStorage(delayMessage, json);

}

} catch (Throwable e) {

log.error("consumer kafka delay message[{}] error!", json, e);

throw e;

}

return true;

}

private void sendActualTopic(DelayMessage delayMessage, String message) {

kafkaSender.send(message, delayMessage.getActualTopic());

}

@SneakyThrows

private void localStorage(DelayMessage delayMessage, String message) {

String key = generateRdbKey(delayMessage);

if (rocksDb.keyMayExist(RocksDbUtils.toByte(key), null)) {

return;

}

rocksDb.put(RocksDbUtils.toByte(key), RocksDbUtils.toByte(message));

}

private String generateRdbKey(DelayMessage delayMessage) {

return delayMessage.getActualTime().getTime() + RDB_KEY_SPLITTER + delayMessage.getEventId();

}

这里要注意生成key的方法：RocksDB是按key自然排序，迭代器遍历时是按key顺序遍历。

按时间来生成key，遍历时遇到第一个不符合的key，即可结束遍历。

key里加上消息ID，用以去重。

处理存储的延迟消息

启动定时任务(ScheduledExecutorService)定时检查消息。private void handleRdbMessage() {

try {

try (RocksIterator rocksIterator = rocksDb.newIterator()) {

for (rocksIterator.seekToFirst(); rocksIterator.isValid(); rocksIterator.next()) {

String key = "";

String value = "";

try {

byte[] keyByte = rocksIterator.key();

key = RocksDbUtils.toString(keyByte);

if (!isMessageExpired(key)) {

break;

}

value = RocksDbUtils.toString(rocksIterator.value());

DelayMessage delayMessage = JSON.parseObject(value, DelayMessage.class);

sendActualTopic(delayMessage, value);

rocksDb.delete(keyByte);

} catch (NumberFormatException e) {

// 异常key

log.error("handler kafka rocksdb delay message[{}:{}] NumberFormatException error!", key, value, e);

if (StringUtils.isNotBlank(key)) {

rocksDb.delete(RocksDbUtils.toByte(key));

}

} catch (Exception e) {

log.error("handler kafka rocksdb delay message[{}:{}] error!", key, value, e);

}

}

}

} catch (Exception e) {

// 捕获异常，否则ScheduledExecutorService会停止定时任务

log.error("handler kafka rocksdb delay message error!", e);

}

}

private boolean isMessageExpired(String rdbKey) {

long actualTime = Long.valueOf(rdbKey.split(RDB_KEY_SPLITTER)[0]);

return actualTime <= System.currentTimeMillis();

}

这里sendActualTopic和rocksDb.delete两个操作并不是原子性，但一般kafka消费端都会做防重复，所以也不会有问题。

其他

当前仅仅简易实现了延迟队列，还有很多需要完成完善的地方，比如：当前数据分散到不同的消费节点上，如果某一个节点服务器异常导致数据丢失，就只能人工介入，从kafka文件里获取数据；可通过部署不同的kafka group来达到数据备份，通过选主方式来决定哪一个group执行业务。

一条消息被存储三份：实际队列，延迟队列，RocksDB，可以通过操作kafka CommitLog的方式，让RocksDB里仅存储CommitLog offset 相关信息，减小RocksDB占用空间。

参考：