kafka消费消息-java版-demo

@SpringBootApplication
public class CcApplication {

    public static void main(String[] args) {
        SpringApplication.run(CcApplication.class, args);

        /**
         * 主要参数：
         * 1.bootstrap.servers,group.id,key.deserializer,value.deserializer
         * 2.session.timeout.ms coordinator检测失败的时间，设置为比较小的值
         * 3. max.poll.interval.ms consumer处理逻辑最大时间
         * 4. auto.offset.reset  [earliest,lastest,none]
         * 5. enable.auto.commit 是否自动提交位移，设置为false，由用户自行提交位移
         * 6. fetch.max.bytes 指定consumer单次获取数据的最大字节数
         * 7. max.poll.records 单次调用poll的最大返回消息数，默认500
         * 8. heartbeat.interval.ms  越小越好
         * 9. connections.max.idle.ms Kafka定期关闭空闲Socket的时间
         */
        String topic = "ty_analysis";
        String groupId = "analysis";

        Properties properties = new Properties();
        properties.put("bootstrap.servers","172.16.9.10:9092");
        //必须指定有业务意义的名字
        properties.put("group.id",groupId);
        properties.put("enable.auto.commit","true");
        properties.put("auto.commit.interval.ms","1000");
        //从最早的消息开始读取
        properties.put("auto.offset.reset","earliest");
        properties.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(properties);
        //订阅主题，可以订阅多个主题，还可以使用正则表达式订阅主题
        //注意：多次订阅，会覆盖前面的
        consumer.subscribe(Arrays.asList(topic));

        try{
            while(true){
                //1000是超时设定，如果有定时要求，可设置，否则建议设置个比较大的值
                //通常consumer拿到足够多的数据，会立即返回，否则会阻塞
                //poll返回则认为是成功消费了消息,如果发现消费慢需要分析是poll慢还是本身业务逻辑处理慢
                ConsumerRecords<String,String> records = consumer.poll(1000);
                for(ConsumerRecord<String,String> record : records){
                    System.out.printf("offset=%d, key=%s,value= %s%n",record.offset(),record.key(),record.value());
                }
            }
        }finally {
            consumer.close();
        }
    }

}

# 主要的pom依赖

<dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.6.0</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.9.5</version>
        </dependency>