Kafka Consumer API样例

浅浅的花香味﹌ 2021-08-30 22:46 329阅读 0赞

Kafka Consumer API样例  
1. 自动确认Offset  
说明参照：http://blog.csdn.net/xianzhen376/article/details/51167333

Properties props = new Properties();
    /* 定义kakfa 服务的地址，不需要将所有broker指定上 */
    props.put("bootstrap.servers", "localhost:9092");
    /* 制定consumer group */
    props.put("group.id", "test");
    /* 是否自动确认offset */
    props.put("enable.auto.commit", "true");
    /* 自动确认offset的时间间隔 */
    props.put("auto.commit.interval.ms", "1000");
    props.put("session.timeout.ms", "30000");
    /* key的序列化类 */
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    /* value的序列化类 */
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     /* 定义consumer */
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    /* 消费者订阅的topic, 可同时订阅多个 */
    consumer.subscribe(Arrays.asList("foo", "bar"));
    
     /* 读取数据，读取超时时间为100ms */
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records)
            System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
    }

说明:  
1. bootstrap.servers 只是代表kafka的连接入口，只需要指定集群中的某一broker；  
2. 一旦consumer和kakfa集群建立连接，consumer会以心跳的方式来高速集群自己还活着，如果session.timeout.ms 内心跳未到达服务器，服务器认为心跳丢失，会做rebalence。

2. 手工控制Offset  
如果consumer在获得数据后需要加入处理，数据完毕后才确认offset，需要程序来控制offset的确认。举个栗子：  
consumer获得数据后，需要将数据持久化到DB中。自动确认offset的情况下，如果数据从kafka集群读出，就确认，但是持久化过程失败，就会导致数据丢失。我们就需要控制offset的确认。

Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("group.id", "test");
    /* 关闭自动确认选项 */
    props.put("enable.auto.commit", "false");
    props.put("auto.commit.interval.ms", "1000");
    props.put("session.timeout.ms", "30000");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Arrays.asList("foo", "bar"));
    final int minBatchSize = 200;
    List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records) {
            buffer.add(record);
        }
        /* 数据达到批量要求，就写入DB，同步确认offset */
        if (buffer.size() >= minBatchSize) {
            insertIntoDb(buffer);
            consumer.commitSync();
            buffer.clear();
        }
    }

还可以精细的控制对具体分区具体offset数据的确认：

try {
        while(running) {
            ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
            for (TopicPartition partition : records.partitions()) {
                List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                for (ConsumerRecord<String, String> record : partitionRecords) {
                    System.out.println(record.offset() + ": " + record.value());
                }
                /* 同步确认某个分区的特定offset */
                long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
            }
        }
    } finally {
      consumer.close();
    }

说明：确认的offset为已接受数据最大offset+1。

3. 分区订阅  
可以向特定的分区订阅消息。但是会失去partion的负载分担。有几种场景可能会这么玩：  
1. 只需要获取本机磁盘的分区数据；  
2. 程序自己或者外部程序能够自己实现负载和错误处理。例如YARN/Mesos的介入，当consumer挂掉后，再启动一个consumer。

String topic = "foo";
    TopicPartition partition0 = new TopicPartition(topic, 0);
    TopicPartition partition1 = new TopicPartition(topic, 1);
    consumer.assign(Arrays.asList(partition0, partition1));

说明：  
1. 此种情况用了consumer Group，也不会做负载均衡。  
2. topic的订阅和分区订阅不可以在同一consumer中混用。

4. 外部存储offset  
消费者可以自定义kafka的offset存储位置。该设计的主要目的是让消费者将数据和offset进行原子性的存储。这样可以避免上面提到的重复消费问题。举栗说明：  
订阅特定分区。存储所获得的记录时，将每条记录的offset一起存储。保证数据和offset的存储是原子性的。当异步存储被异常打断时，凡已经存储的数据，都有有相应的offset记录。这种方式可以保证不会有数据丢失，也不会重复的从服务端读取。  
如何配置实现：  
1. 去使能offset自动确认：enable.auto.commit=false；  
2. 从ConsumerRecord中获取offset，保存下来;  
3. Consumer重启时，调用seek(TopicPartition, long)重置在服务端的消费记录。

如果消费分区也是自定义的，这种方式用起来会很爽。如果分区是自动分配的，当分区发生reblance的时候，就要考虑清楚了。如果因为升级等原因，分区漂移到一个不会更新offset的consumer上，那就日了狗了。  
该情况下：  
1. 原consumer需要监听分区撤销事件，并在撤销时确认好offset。接口:ConsumerRebalanceListener.onPartitionsRevoked(Collection)；  
2. 新consumer监听分区分配事件，获取当前分区消费的offset。接口：ConsumerRebalanceListener.onPartitionsAssigned(Collection)；  
3. consumer监听到 ConsumerRebalance事件，还没有处理或者持久化的缓存数据flush掉。

5. 控制消费位置  
大多数情况下，服务端的Consumer的消费位置都是由客户端间歇性的确认。Kafka允许Consumer自己设置消费起点，达到的效果：  
1. 可以消费已经消费过的数据；  
2. 可以跳跃性的消费数据；  
看下这样做的一些场景：  
1. 对Consumer来说，数据具备时效性，只需要获取最近一段时间内的数据，就可以进行跳跃性的获取数据；  
2. 上面自己存offset的场景，重启后就需要从指定的位置开始消费。  
接口上面已经提到过了，用seek(TopicPartition, long)。、  
麻蛋，说指针不就好了，这一小节就是多余的叨叨。

6. 控制消费流Consumption Flow Control  
如果一个consumer同时消费多个分区，默认情况下，这多个分区的优先级是一样的，同时消费。Kafka提供机制，可以让暂停某些分区的消费，先获取其他分区的内容。场景举栗：  
1. 流式计算，consumer同时消费两个Topic，然后对两个Topic的数据做Join操作。但是这两个Topic里面的数据产生速率差距较大。Consumer就需要控制下获取逻辑，先获取慢的Topic，慢的读到数据后再去读快的。  
2. 同样多个Topic同时消费，但是Consumer启动是，本地已经存有了大量某些Topic数据。此时就可以优先去消费下其他的Topic。

调控的手段：让某个分区消费先暂停，时机到了再恢复，然后接着poll。接口：pause(TopicPartition…)，resume(TopicPartition…)

7. 多线程处理模型 Multi-threaded Processing  
Kafka的Consumer的接口为非线程安全的。多线程共用IO，Consumer线程需要自己做好线程同步。  
如果想立即终止consumer，唯一办法是用调用接口：wakeup()，使处理线程产生WakeupException。看砖：

public class KafkaConsumerRunner implements Runnable {
        /* 注意，这俩货是类成员变量 */
        private final AtomicBoolean closed = new AtomicBoolean(false);
        private final KafkaConsumer consumer;
    
        public void run() {
            try {
                consumer.subscribe(Arrays.asList("topic"));
                while (!closed.get()) {
                    ConsumerRecords records = consumer.poll(10000);
                    // Handle new records
                }
            } catch (WakeupException e) {
                // Ignore exception if closing
                if (!closed.get()) throw e;
            } finally {
                consumer.close();
            }
        }
    
        // Shutdown hook which can be called from a separate thread
        public void shutdown() {
            closed.set(true);
            consumer.wakeup();
        }
    }

说明：  
1. KafkaConsumerRunner是runnable的，请自觉补脑多线程运行；  
2. 外部线程控制KafkaConsumerRunner线程的停止；  
3. 主要说的是多线程消费同一topic，而不是消费同一分区；

比较一下两种模型：

Consumer单线程模型  
优点：实现容易；  
优点：没有线程之间的协作。通常比下面的那种更快；  
优点：单分区数据的顺序处理；  
缺点：多个TCP连接，但是关系不大，kafka对自己的server自信满满；  
缺点：太多的Request可能导致server的吞吐降低一丢丢；  
缺点：consumer数量受到分区数量限制，一个consumer一个分区；

Consumer多线程模型  
优点：一个consumer任意多的线程，线程数不用受到分区数限制；  
缺点：如果有保序需求，自己要加控制逻辑；  
缺点：该模型中如果手动offset，自己要加控制逻辑；  
一种可行的解决办法：为每个分区分配独立的存储，获取的数据根据数据所在分区进行hash存储。这样可以解决顺序消费，和offset的确认问题。

后记  
其实对于官网上说的，我是迷惑的：  
对比两种线程模型时，应该是有隐藏地图的。  
1. 单线程模型中，多分区情况下，应该说的是每个Consumer独立去消费一个分区；  
2. 多线程模型中，单Consumer消费一个Topic。如果多个线程同时消费同一分区，也就是要公用连接了，各个线程之间要做好同步；  
3. 对于多线程模型下提出的客户端分区数据分开存储，各个分区之间是如何保序的？