41 lines
1.5 KiB
Java
41 lines
1.5 KiB
Java
|
package com.baiyue.audit.mq;
|
||
|
|
|
||
|
|
import com.baiyue.audit.repository.AuditDeadLetterRepository;
|
||
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||
|
|
import jakarta.annotation.Resource;
|
||
|
|
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
|
||
|
|
import org.apache.rocketmq.spring.core.RocketMQListener;
|
||
|
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||
|
|
import org.springframework.stereotype.Component;
|
||
|
|
|
||
|
|
import java.nio.charset.StandardCharsets;
|
||
|
|
import java.util.UUID;
|
||
|
|
|
||
|
|
/** 将 RocketMQ 最终死信转存数据库，供补偿任务恢复。 */
|
||
|
|
@Component
|
||
|
|
@ConditionalOnProperty(prefix = "baiyue.audit.rocketmq", name = "dead-letter-consumer-enabled", havingValue = "true")
|
||
|
|
@RocketMQMessageListener(
|
||
|
|
topic = "${baiyue.audit.rocketmq.dead-letter-topic}",
|
||
|
|
consumerGroup = "${baiyue.audit.rocketmq.dead-letter-group}")
|
||
|
|
public class AuditDeadLetterConsumer implements RocketMQListener<String> {
|
||
|
|
@Resource
|
||
|
|
private ObjectMapper json;
|
||
|
|
@Resource
|
||
|
|
private AuditDeadLetterRepository repository;
|
||
|
|
|
||
|
|
@Override
|
||
|
|
public void onMessage(String payload) {
|
||
|
|
repository.save(eventId(payload), payload);
|
||
|
|
}
|
||
|
|
|
||
|
|
private String eventId(String payload) {
|
||
|
|
try {
|
||
|
|
String value = json.readTree(payload).path("eventId").asText();
|
||
|
|
if (!value.isBlank()) return value;
|
||
|
|
} catch (Exception ignored) {
|
||
|
|
// 无法解析的毒消息也必须使用确定性 ID 收入死信表。
|
||
|
|
}
|
||
|
|
return "invalid-" + UUID.nameUUIDFromBytes(payload.getBytes(StandardCharsets.UTF_8));
|
||
|
|
}
|
||
|
|
}
|