java——spring boot集成kafka——消费端代码实现

刺骨的言语ヽ痛彻心扉 2023-09-30 21:08 21阅读 0赞

![01fc1e5c69320de7eec52119c29db388.png][]

![1bbb882150d32607153a5de2544f94b9.png][]

public class MyConsumer {
    private final static String TOPIC_NAME = "my-replicated-topic";
    private final static String CONSUMER_GROUP_NAME = "testGroup";
    
    public static void main(String[] args) {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.31.167.10:9092,10.31.167.10:9093,10.31.167.10:9094");
    // 消费分组名
    props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
    //创建一个消费者的客户端
    KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);
    // 消费者订阅主题列表
    consumer.subscribe(Arrays.asList(TOPIC_NAME));
    
    while (true) {
    /*
    * poll() API 是拉取消息的⻓轮询
    */
    ConsumerRecords<String, String> records =consumer.poll(Duration.ofMillis( 1000 ));
    for (ConsumerRecord<String, String> record : records) {
    System.out.printf("收到消息：partition = %d,offset = %d, key =%s, value = %s%n", record.partition(),record.offset(), record.key(), record.value());
                }
            }
        }
    }

![4f39bfce4b0b2209a1e67346831b5afc.png][]

![d7000f34a9a53da19dd6af8f96b1dca5.png][]

[01fc1e5c69320de7eec52119c29db388.png]: https://img-blog.csdnimg.cn/img_convert/01fc1e5c69320de7eec52119c29db388.png
[1bbb882150d32607153a5de2544f94b9.png]: https://img-blog.csdnimg.cn/img_convert/1bbb882150d32607153a5de2544f94b9.png
[4f39bfce4b0b2209a1e67346831b5afc.png]: https://img-blog.csdnimg.cn/img_convert/4f39bfce4b0b2209a1e67346831b5afc.png
[d7000f34a9a53da19dd6af8f96b1dca5.png]: https://img-blog.csdnimg.cn/img_convert/d7000f34a9a53da19dd6af8f96b1dca5.png