KafkaListener annotation

Posted on | Edited on | In

Listener container factory

Define a config loader function fist

1
2
3
4
5
6
7
8
private Map<String, Object> consumerConfigs(String groupId) {
    Map<String, Object> consumerConfig = new HashMap<>();
    initKafkaClientConfiguration(consumerConfig);
    consumerConfig.put("group.id", groupId);
    consumerConfig.put(..., ...);
    ...
    return consumerConfig;
}

Create factory bean

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@Bean
public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactoryOne() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConcurrency(10);
    factory.getContainerProperties().setPollTimeout(3000);
    factory.setConsumerFactory(consumerFactory("group-one"));
    return factory;
}

@Bean
public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactoryTwo() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConcurrency(10);
    factory.getContainerProperties().setPollTimeout(3000);
    factory.setConsumerFactory(consumerFactory("group-two"));
    return factory;
}

Point factory in @KafkaListener

1
2
3
4
5
@KafkaListener(topics = "Topic-1" , containerFactory = "kafkaListenerContainerFactoryOne")
...

@KafkaListener(topics = "Topic-2" , containerFactory = "kafkaListenerContainerFactoryTwo")
...

Get header body by annotation

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@KafkaListener(id = "anno", topics = "topic-3")
public void annoListener(@Payload String data,
                         @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) Integer key,
                         @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                         @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                         @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts) {
    log.info("topic.quick.anno receive : \n"+
        "data : "        + data +      "\n" +
        "key : "         + key +       "\n" +
        "partitionId : " + partition + "\n" +
        "topic : "       + topic +     "\n" +
        "timestamp : "   + ts +        "\n"
    );
}

https://blog.csdn.net/my_momo_csdn/article/details/89366205
https://www.jianshu.com/p/a64defb44a23