- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
重试 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)。
如果您的意思是从 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);
}
}
唯一的问题是如果重播记录仍然失败会发生什么情况;容器永远不会闲置,所以你将永远循环。一种解决方案是以某种方式检测循环并在这种情况下停止容器。
关于java - 重新发布kafka DLQ主题,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/57157134/
我试图修改Logstash 7.6,以便将失败的消息发送到SQS DLQ [1]。具体来说,当Logstash使用SQS输入插件和elasticsearch输出插件时,我需要进行此修改。我在过滤器部分
有谁知道如何从 jBoss 中清除 DeadLetterQueue? 当我启动 jBoss 时,它在以下位置等待 4 分钟: 12:09:06,281 INFO [ConnectionFactory
我正在使用以下服务器端重新传递配置
private BrokerService createBroker() throws IOException, Exception { BrokerService broker =
作为操作人员(不是开发人员)并基于 http://activemq.apache.org/message-redelivery-and-dlq-handling.html我试图在集成到 JBoss F
关于允许应用程序拒绝什么样的消息,是否有一个好的最佳实践? 我的理解是所有无法处理的消息都应该被拒绝到死信队列 - 无论问题是消息中的语法错误还是语义错误,或者应用程序是否暂时无法处理消息(例如,因为
我们有一个外部 SB 主题订阅,我们是其消费者。我们当前正在读取最大并发线程数为 20 的 SB。 我想知道偶尔耗尽 DLQ 的最佳方法是什么?它应该包含在同一个应用程序中还是我应该创建另一个应用程序
我最近开始研究目前在我们的项目中使用的 Azure 服务总线模块。当前的模型是,我们向一个主题发送消息,并且该主题有多个订阅。订阅没有过滤器,订阅者使用它(当前不打算添加过滤器)。 问题:假设有 1
我正在使用 AMQ 5.5。我想禁用将死信发送到 ActiveMQ.DLQ 目的地的选项,并完全丢弃(自动)否则将发送到那里的消息。为此,我按如下方式配置了代理:
我们将 spring jms 与 activeMQ 一起使用,我们想重命名 DLQ。这背后的原因是我们有几个项目,我们希望每个项目都有一个单独的 DLQ。 到目前为止,我只看到了如何设置单独的 DLQ
我有一个触发 Lambda 的 DynamoDB 流,我想将任何失败的事件推送到 DLQ。 如果 DLQ 的源是 SQS 队列,看起来您可以执行称为重新驱动回源队列的操作,DLQ 中的消息将被移回源队
当本地队列管理器在其 AMQ 错误日志中收到以下消息时: 09/13/12 08:00:19 - Process(3017.20) User(mqm) Program(amqrmppa_nd) AMQ
Azure 服务总线实体(队列/主题)支持生存时间 (TTL)。当 TTL 过去时,消息就会过期。到期时,系统会删除该消息或将其移至死信队列 (DLQ)。 服务总线是否有其他设置可以在指定时间段后从
当消息传递到 Azure 服务总线死信队列时,我尝试发送邮件警报。我没有找到任何合适的解决方案来发送邮件警报。我们可以在 azure 还是 .net core 的哪一侧处理这个问题?如果是这样,任何人
我已经设置了 dlq 和 dlx,但是失败的消息没有重定向到 dlq。我正在尝试从 java 应用程序以及从 rabbitmq 服务器向 MESSAGES.EXCHANGE 发送消息,在这两种情况下我
Azure 队列存储和 Azure 服务在死信队列和有害消息方面有何区别? 如何从这些队列中读取消息? 最佳答案 有害消息是发送到队列或主题但消费应用程序无法正确处理的消息。 一旦DeliveryCo
当消息传递到 Azure 服务总线死信队列时,我尝试发送邮件警报。我没有找到任何合适的解决方案来发送邮件警报。我们可以在 azure 还是 .net core 的哪一侧处理这个问题?如果是这样,任何人
我已经设置了 dlq 和 dlx,但是失败的消息没有重定向到 dlq。我正在尝试从 java 应用程序以及从 rabbitmq 服务器向 MESSAGES.EXCHANGE 发送消息,在这两种情况下我
Azure 队列存储和 Azure 服务在死信队列和有害消息方面有何区别? 如何从这些队列中读取消息? 最佳答案 有害消息是发送到队列或主题但消费应用程序无法正确处理的消息。 一旦DeliveryCo
我必须使用 JMS API 处理死信队列 (DLQ) 中的消息。目标是读取原始消息的正文及其用户属性。我意识到这种 DLQ 处理方法可能被认为是糟糕的设计,但无论如何我都必须处理它。 使用 JMS 读
我是一名优秀的程序员,十分优秀!