apache-kafka - 优先考虑 Kafka 主题
问题描述
我需要完全从 topic1 读取消息,然后从 topic2 读取消息。我每天都会收到一次关于这些主题的消息。在读取 topic1 中的所有消息之前,我设法停止读取来自 topic2 的消息,但这仅在服务器启动时发生一次。有人可以帮我解决这个问题。
监听器配置代码
@EnableKafka
@Configuration
public class ListenerConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "batch");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");
return props;
}
@Bean
public ConsumerFactory<String, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setBatchListener(true);
return factory;
}
@Bean("kafkaListenerContainerTopic1Factory")
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerTopic1Factory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setIdleEventInterval(60000L);
factory.setBatchListener(true);
return factory;
}
@Bean("kafkaListenerContainerTopic2Factory")
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerTopic2Factory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setBatchListener(true);
return factory;
}
}
侦听器代码
@Service
public class Listener {
private static final Logger LOG = LoggerFactory.getLogger(Listener.class);
@Autowired
private KafkaListenerEndpointRegistry registry;
@KafkaListener(id = "first-listener", topics = "topic1", containerFactory = "kafkaListenerContainerTopic1Factory")
public void receive(@Payload List<String> messages,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
@Header(KafkaHeaders.OFFSET) List<Long> offsets) {
for (int i = 0; i < messages.size(); i++) {
LOG.info("received first='{}' with partition-offset='{}'",
messages.get(i), partitions.get(i) + "-" + offsets.get(i));
}
}
@KafkaListener(id = "second-listener", topics = "topic2", containerFactory = "kafkaListenerContaierTopic2Factory" , autoStartup="false" )
public void receiveRel(@Payload List<String> messages,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
@Header(KafkaHeaders.OFFSET) List<Long> offsets) {
for (int i = 0; i < messages.size(); i++) {
LOG.info("received second='{}' with partition-offset='{}'",
messages.get(i), partitions.get(i) + "-" + offsets.get(i));
}
}
@EventListener()
public void eventHandler(ListenerContainerIdleEvent event) {
LOG.info("Inside event");
this.registry.getListenerContainer("second-listener").start();
}
请帮助我解决,因为这个循环应该每天都发生。完全读取 topic1 消息,然后从 topic2 读取消息。
解决方案
您已经在使用空闲事件侦听器来启动第二个侦听器 - 它还应该停止第一个侦听器。
当第二个监听器空闲时;停下来。
您应该检查事件是针对哪个容器来决定停止和/或启动哪个容器。
然后,使用 a ,在下次您希望它启动时TaskScheduler
安排第一个侦听器的 a 。start()
推荐阅读
- javascript - 使用 CloudFront 与 S3TA 的 PUT/POST 请求
- python - PySpark - 将单元素数组/列表转换为字符串
- r - 如何平均 7 年的数据
- python - 使用相同的索引形成子列表
- python - 正则表达式满足两个字符串条件
- java - “CharSequence 接口”和“CharSequence 键”之间的区别
- json - 将 Json 对象集合转换为数组
- jquery - 限制 jQuery 弹出窗口?
- angular - 如何使用 colspan 在动态表中添加列?
- r - 设置数据帧时的 R 错误:“== 中的错误:未实现这些类型的比较”