java - 重新发布 kafka DLQ 主题
问题描述
重试 2 次后,我的话题转到了 DLQ。
有什么方法可以按需重新发布 dlq 主题(手动使用 api)?
在这里,我添加了三个 api 的代码
1) 将发送消息并成功消费。
2)将发送主题并抛出错误,最后将移动到DLQ。
3)将尝试从 DLQ(我卡住的地方)阅读和发布主题。
public class KafkaProducerConfig {
private String bootstrapAddress="localhost:9092";
private String groupId = "dlq_topics_group";
@Bean
public ProducerFactory<String, Object> producerFactory() {
Map<String, Object> configProps = new HashMap<String, Object>();
configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapAddress);
configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
configProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
return new DefaultKafkaProducerFactory<String, Object>(configProps);
}
@Bean
public KafkaTemplate<String, Object> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
@EnableKafka
@Configuration
public class KafkaConsumerConfig {
private String bootstrapAddress="localhost:9092";
private String groupId = "dlq_topics_group";
@Autowired
KafkaTemplate<Object, Object> template;
private Integer maxRetry = 3;
private Integer concurrency=1;
private Integer pollTimeout=1000;
@Bean
public ConsumerFactory<Object, Object> consumerFactory() {
Map<String, Object> props = new HashMap<String, Object>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
return new DefaultKafkaConsumerFactory<>(props);
}
@Bean
public ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setConcurrency(concurrency);
factory.getContainerProperties().setPollTimeout(pollTimeout);
factory.getContainerProperties().setSyncCommits(true);
factory.getContainerProperties().setAckOnError(false);
factory.setErrorHandler(new SeekToCurrentErrorHandler(new DeadLetterPublishingRecovererHandler(template), maxRetry));
configurer.configure(factory, consumerFactory());
return factory;
}
@Bean
public KafkaConsumer<Object, Object> kafkaConsumer() {
return new KafkaConsumer<(consumerFactory().getConfigurationProperties());
}
}
@Controller
@RequestMapping(value="/kafka/controller")
public class KafkaController {
@Autowired
MessageService messageService;
@PostMapping(value="/send/{message}")
public void sendMessage1(@PathVariable String message) {
messageService.sendMessage(message);
}
@PostMapping(value="/send-kafka-msg/{message}")
public void sendMessage(@PathVariable String message) {
messageService.sendMessage(message);
}
@PostMapping(value="/resend-dlq-topic")
public void reprocessDLQTopics() {
**//LOGIC TO READ DLQ AND REPROCESS THE TOPICS**
messageService.reprocessDLQTopics();
}
}
@Service
public class MessageService {
private static final Logger logger = LoggerFactory.getLogger(MessageService.class);
private static final String TOPIC_EX = "kafka-msg-topic-ex";
private static final String TOPIC = "kafka-msg-topic";
private static final String DLQ_TOPICS = "dql_topics";
private static final String DLQ_NOTIFICATION_TOPIC = "dlq_notification_topic";
@Autowired
KafkaTemplate<String, Object> kafkaTemplate;
@Autowired
KafkaAdmin kafkaAdmin;
@Autowired
KafkaConsumer<Object, Object> kafkaConsumer;
public void sendMessage(String message) {
logger.info("sending message ", message);
kafkaTemplate.send(TOPIC, message);
}
public void sendMessageEx(String message) {
logger.info("sending message ", message);
kafkaTemplate.send(TOPIC, message);
}
@KafkaListener(topics = { TOPIC, TOPIC_EX }, groupId = "group-id")
public void messageReciver(String message) {
logger.info("message receiver" +kafkaAdmin.getConfig().entrySet().parallelStream().toString());
throw new RuntimeException();
}
@KafkaListener(id = "DLQ_TOPICS_GROUP", topics = DLQ_TOPICS)
public void dltListen(String dlqMsg) {
logger.info("Received from DLT: " + dlqMsg.toString());
}
public void reprocessDLQTopics() {
**//LOGIC TO RE-PUBLISH DLQ TOPICS**
}
}
解决方案
目前尚不清楚“重新发布 dlq 主题”是什么意思,但您可以使用KafkaTemplate
.
如果您的意思是从 DLQ 主题消费并重新发布到原始主题,那么是的,您也可以这样做。
编辑
这是一种解决方案:
@Component
class Config {
@Autowired
private KafkaTemplate<String, String> template;
@Autowired
private KafkaListenerEndpointRegistry registry;
@KafkaListener(id = "so57157134", topics = "so57157134")
public void listen(String in) {
System.out.println(in);
}
@KafkaListener(id = "so57157134_DLT", topics = "so57157134_DLT", autoStartup = "false")
public void listenDLT(String in) {
System.out.println("Re-publish DLT");
this.template.send("so57157134", in);
}
@EventListener
public void listenForIdleContainer(ListenerContainerIdleEvent event) {
System.out.println(event);
this.registry.getListenerContainer("so57157134_DLT").stop();
}
public void reprocessDLQTopics() {
this.registry.getListenerContainer("so57157134_DLT").start();
}
@Bean
public NewTopic topic() {
return new NewTopic("so57157134", 10, (short) 1);
}
@Bean
public NewTopic topicDLT() {
return new NewTopic("so57157134_DLT", 10, (short) 1);
}
}
唯一的问题是,如果重放记录仍然失败,会发生什么情况;容器永远不会闲置,因此您将永远循环。一种解决方案是以某种方式检测循环并在该条件下停止容器。
推荐阅读
- javascript - JSXGraph 在 down 事件后停止板平移
- c# - 如何在没有使用 F#/C# 的 WinForms 的情况下将像素绘制到屏幕上?
- java - 导入的模块 OpenCV 的导入不再可用,导致使用的每个关键字都成为 Android Studio 中的错误
- javascript - 我怎么知道一个字符串中有多少匹配被替换?
- r - 从数据框创建文本字符串
- javascript - 将一个对象的值匹配到另一个对象
- css - 覆盖 Materialise 1.0.0 单选风格
- php - 卷曲水滴参数
- java - Travis 运行检查权限被拒绝错误解决办法?
- css - 在 CSS 中为边框底部添加偏移量