Kafka客户端API

创建项目，引入依赖：

        
        
            org.apache.kafka
            kafka-clients
            2.8.1
        

针对生产者和消费者使用进行示例演示。前面了解了Kafa服务安装和基本概念，接下来直接敲代码。

一、快速入门

创建一个 Order实体类，作为消息数据发送。

public class Order {

    private Integer orderId;
    private String orderDesc;
    private Double orderAmount;
    ...
} 

1、生产者

生产者需要用到的几个类：

• KafkaProducer： 需要一个生产者对象，用来发送数据。
• ProducerConfig： 获取所需一系列配置参数。
• ProducerRecord： 每条消息数据都要封装成一个 ProducerRecord对象。

public class Producer {

    private final static String TOPIC_NAME = "my-test-topic";
    private final static String KAFKA_SERVERS_CLUSTER_STR = "192.168.xxx.xxx:9092";//集群用逗号分隔

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 1.设置生产者的配置信息
        Properties props = makeProducerProperties();

        // 2.创建生产者对象。通过配置信息
        KafkaProducer producer = new KafkaProducer(props);

        // 3.发送消息
        for (int i = 1; i  0) {
                // 手动同步提交offset，当前线程会阻塞直到offset提交成功
                // 一般使用同步提交，因为提交之后一般也没有什么逻辑代码了
                //consumer.commitSync();

                // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面的程序逻辑
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map offsets, Exception exception) {
                        if (exception != null) {
                            System.err.println("Commit failed for " + offsets);
                            System.err.println("Commit failed exception: " + exception.getStackTrace());
                        }
                    }
                });
            }

2.1 数据漏消费和重复消费分析：

无论是同步提交还是异步提交 offset，都有可能会造成数据漏消费或重复消费。

• 先提交 offset 后消费，有可能造成数据的漏消费；
• 先消费后提交 offset，有可能造成数据的重复消费。

3、消费者订阅主题

消费者对象订阅主题有多种方式：

• 直接订阅主题
• 指定分区订阅
• 消息回溯消费订阅
• 指定offset消费
• 从指定时间点开始消费等等

        // 3.消费者对象订阅主题
        // 直接订阅主题
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        // 消费指定分区
        //consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));

        //消息回溯消费
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));*/

        //指定offset消费
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);*/

        //从指定时间点开始消费

        /*List topicPartitions = consumer.partitionsFor(TOPIC_NAME);
        //从1小时前开始消费
        long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;
        Map map = new HashMap();
        for (PartitionInfo par : topicPartitions) {
            map.put(new TopicPartition(TOPIC_NAME, par.partition()), fetchDataTime);
        }
        Map parMap = consumer.offsetsForTimes(map);
        for (Map.Entry entry : parMap.entrySet()) {
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if (key == null || value == null) continue;
            Long offset = value.offset();
            System.out.println("partition-" + key.partition() + "|offset-" + offset);
            System.out.println();
            //根据消费里的timestamp确定offset
            if (value != null) {
                consumer.assign(Arrays.asList(key));
                consumer.seek(key, offset);
            }
        }*/

注意：消息回溯消费订阅，每次都会从头消费。理解它与 ConsumerConfig.AUTO_OFFSET_RESET_CONFIG = "earliest" 配置参数的区别。

4、配置参数

部分参数说明：

    private static Properties makeConsumerProperties() {
        Properties props = new Properties();
        //指定连接的 kafka集群
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS_CLUSTER_STR);
        // 消费分组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);

       /* // 是否自动提交offset，默认就是true，推荐使用手动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 自动提交offset的间隔时间
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");*/
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        /*
            当消费主题的是一个新的消费组，或者指定offset的消费方式，offset不存在，那么应该如何消费
            latest(默认) ：只消费自己启动之后发送到主题的消息
            earliest：第一次从头开始消费，以后按照消费offset记录继续消费，这个需要区别于consumer.seekToBeginning(每次都从头开始消费)
        */
        //props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        /*
            consumer给broker发送心跳的间隔时间，broker接收到心跳，如果此时有rebalance发生会通过心跳响应将
            rebalance方案下发给consumer，这个时间可以稍微短一点
		*/
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        /*
            服务端broker多久感知不到一个consumer心跳就认为他故障了，会将其踢出消费组，
            对应的Partition也会被重新分配给其他consumer，默认是10秒
        */
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);

        //一次poll最大拉取消息的条数，如果消费者处理速度很快，可以设置大点，如果处理速度一般，可以设置小点
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        /*
            如果两次poll操作间隔超过了这个时间，broker就会认为这个consumer处理能力太弱，
            会将其踢出消费组，将分区分配给别的consumer消费
        */
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);

        //key,value的序列化：这里把 key和 value从字符串序列化为字节数组
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        return props;
    }

关于 Kafka客户端：

通过使用 Properties来创建连接 kafka server的客户端AdminClient，它是一个抽象类，只有一个 KafkaAdminClient实现类。通过客户端AdminClient ，我们可以进行 Topic的CRUD等操作。

使用比较简单，可查看参考文章，博主写的很详细。

参考文章：

• kafka客户端操作之Admin API：https://blog.csdn.net/qq_37774171/article/details/122719003

– 求知若饥，虚心若愚。