kafka知识体系-消费者编程实践

本文主要实际编程讲解kafka生产者相关内容，版本kafka_2.11-0.10.1.0。
kafka消费者相对生产者来说，要复杂一些，主要是涉及到消息的拉取，消息的处理，偏移量offset的提交

kafka安装

前面博文中均已详细介绍，不再赘述。
linux集群安装参考大数据平台搭建-kafka集群的搭建
window单机版安装参考kafka知识体系-生产者编程实践

实践

依赖

kafka 0.10.1.0版本中采用KafkaConsumer对象用来向kafka broker集群拉取消息。
编写代码前先引入相关依赖包：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.11</artifactId>
    <version>0.10.1.0</version>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.10.1.0</version>
</dependency>

基本概念

消费者位置和偏移量
消费者的位置表示下一条消息的偏移量，其比消费者在分区中已经消费的消息的偏移量还要大1。
提交的偏移量表示消费完成后，安全记录保存的偏移量，也是消费进程失败或者重启后的恢复点。可以同步提交commitSync或者异步提交偏移量commitAsync，也可以让消费者通过线程定时提交偏移量。

消费组和订阅主题
kafka允许多个消费者订阅同一个主题并共享同一个消费组，来实现多个进程来消费和处理消息。其中这些进程可以可以在同一台机器上运行，也可以分布在多台机器上，为处理提供可扩展性和容错能力。
组中的每个消费者都可以通过其subscribe方法动态设置要订阅的主题列表 。Kafka将订阅主题中的每条消息传递给每个消费者组中的一个进程。这是通过平衡消费者组中所有成员之间的分区来实现的，这样每个分区就只分配给该组中的一个使用者。因此，如果存在具有四个分区的主题和具有两个进程的使用者组，则每个进程将使用两个分区。
当消费者失败或者新增消费者，以及新增分区都会发生消费组内均衡操作。同时当均衡操作发生时，可以通过ConsumerRebalanceListener去监听，以便完成必要的应用程序级逻辑，例如状态清理，手动偏移提交等。

消费者故障检测
消费者在订阅主题后，调用poll方法时将自动加入组，poll方法旨在确保消费者的活力。只要继续调用poll方法，消费者将留在组中并继续从分配的分区中接收消息。在底层，是消费者会定期向服务端发送心跳。当消费者崩溃或者无法再持续的时间内(session.timeout.ms)发送心跳，则消费者被视为死亡，其占用的分区也会被重新分配。
此外，消费者可能遇到活锁的情况，即仍在发送心跳，但没有消费消息。
为了防止消费者在这种情况下无限期地占用其分区，我们使用该max.poll.interval.ms设置提供了活跃度检测机制。基本上，如果您不至少与配置的最大间隔一样频繁地调用轮询，则客户端将主动离开该组，以便其他使用者可以接管其分区。发生这种情况时，您可能会看到偏移提交失败（如CommitFailedException异常）。这是一种安全机制，可确保只有组中的活动成员才能提交偏移量。
消费者提供两个配置设置来控制轮询循环的行为：

• max.poll.interval.ms：通过增加预期轮询之间的间隔，您可以为消费者提供更多时间来处理从中返回的一批记录poll(long)。缺点是增加此值可能会延迟组重新平衡，因为消费者将仅在轮询调用内加入重新平衡。
• max.poll.records：使用此设置可限制从单个调用返回到poll的总记录数。这可以更容易地预测每个轮询间隔内必须处理的最大值。通过调整此值，您可以减少轮询间隔，这将减少组重新平衡的影响。

编码实践

单线程自动提交偏移量

package com.molyeo.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Properties;

public class SingleKafkaConsumerDemo {
    static Logger logger= LoggerFactory.getLogger(SingleKafkaConsumerDemo.class.getName());
    public static void main(String[] args){
        Properties props = new Properties();
        props.put("bootstrap.servers","LAPTOP-2CBRDCI0:9092");
        props.put("group.id","mygroup");
        props.put("enable.auto.commit","true");
        props.put("auto.commit.interval.ms","1000");
        props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer consumer=new KafkaConsumer(props);
        consumer.subscribe(Arrays.asList("TEST"));
        try{
            while (true){
                ConsumerRecords<String,String> records=consumer.poll(200);
                for(ConsumerRecord<String,String> record:records){
                    logger.info("topic={},offset = {}, key = {}, value = {}", record.topic(), record.offset(), record.key(), record.value());
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            consumer.close();
        }
    }
}

程序中设置配置enable.auto.commit=true,auto.commit.interval.ms=1000即每秒钟一次的频率自动提交offset的值，这种方式最简单方便。存在的问题是如果我们拉取消息后，如果在处理消息的过程中出现异常，而此时offset值已经更新提交了，会导致消息没有正确处理即已丢失。

手工提交偏移量
针对自动提交偏移量的缺陷，kafka消费者允许我们手工控制何时将记录视为已消耗，并提交其偏移量，这能将消息的拉取和消息的处理过程解耦。

package com.molyeo.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class ManualOffsetControlDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "LAPTOP-2CBRDCI0:9092");
        props.put("group.id", "mygroup");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        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("TEST"));
        final int minBatchSize = 20;
        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);
            }
            if (buffer.size() >= minBatchSize) {
                handle(buffer);
                consumer.commitSync();
                buffer.clear();
            }
        }
    }

    public static void handle(List<ConsumerRecord<String, String>> recordList) {
        //do something
    }
}

在上面的代码中，我们拉取消息，并将消息放入到缓存中，当消息累计到一定数量后，调用handle方法去处理数据，处理完成后再去提交偏移量。如果在handle处理后，偏移量提交前进程失败了，则下次启动后由于偏移量还是之前的，我们还能再重复消费一次数据。

上面的代码中，是将已接收的消息标记为已提交，此外我们还可以明确指标偏移量来更好的控制已提交的记录。如下我们将一个分区的消息接收处理完成后，按照分区去提交偏移量。

 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());
             }
             long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
             consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
         }
     }
 } finally {
   consumer.close();
 }

多线程并发消费
多线程并发消费的时候，尤其得注意KafkaConsumer不是线程安全的。这样每个线程不能共享KafkaConsumer实例，常用的方法是每个线程包含一个KafkaConsumer实例，这样offset的控制则只需要关注本线程，不需要考虑其他线程的情况。

package com.molyeo.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by zhangkh on 2018/7/9.
 * auto commit offset
 * multi thread
 */
public class MultiKafkaConsumerDemo {
    Logger logger = LoggerFactory.getLogger(MultiKafkaConsumerDemo.class.getName());

    public static void main(String[] args) {
        String topic = "TEST";
        String groupId = "mygrouop";
        int nThreads = 3;
        ConsumerGroup consumerGroup = new ConsumerGroup(topic, groupId, nThreads);
        consumerGroup.run();
    }
}

class ConsumerGroup {
    Logger logger = LoggerFactory.getLogger(ConsumerGroup.class.getName());
    private ExecutorService executorService;
    private List<ConsumerTask> consumerTaskList;

    ConsumerGroup(String topic, String groupId, int nThreads) {
        executorService = Executors.newFixedThreadPool(nThreads);
        consumerTaskList = new ArrayList<>(nThreads);
        for (int i = 0; i < nThreads; i++) {
            consumerTaskList.add(new ConsumerTask(topic, groupId, i));
        }
    }

    public void run() {
        for (ConsumerTask task : consumerTaskList) {
            executorService.submit(task);
        }
    }

    public void shutdown() {
        if (executorService != null) {
            executorService.shutdown();
        }
        try {
            if (!executorService.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
                logger.info("Timed out waiting for consumer threads to shut down, exiting uncleanly");
            }
        } catch (InterruptedException e) {
            logger.error("Interrupted during shutdown, exiting uncleanly");
            e.printStackTrace();
        }
    }
}

class ConsumerTask<K, V> implements Runnable {
    private final AtomicBoolean closed = new AtomicBoolean(false);

    Logger logger = LoggerFactory.getLogger(ConsumerTask.class.getName());
    private final KafkaConsumer consumer;
    private String topic;
    private String groupId;
    private int threadNo;

    public ConsumerTask(String topic, String groupId, int threadNo) {
        this.topic = topic;
        this.groupId = groupId;
        this.threadNo = threadNo;
        Properties props = getConsumerConfig(groupId);
        consumer = new KafkaConsumer(props);
        consumer.subscribe(Arrays.asList(topic));
    }

    public Properties getConsumerConfig(String groupId) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaCommonConfig.BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KafkaCommonConfig.KEY_DESERIALIZER_CLASS);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaCommonConfig.VALUE_DESERIALIZER_CLASS);
        return props;
    }

    @Override
    public void run() {
        try{
            while (!closed.get()) {
                ConsumerRecords<K, V> records = consumer.poll(100);
                for (ConsumerRecord record : records) {
                    //TODO do something
                    logger.info("{},threadNo={},topic={},groupId={},offset = {}, key = {}, value = {}", Thread.currentThread().getName(),threadNo, record.topic(),groupId, record.offset(), record.key(), record.value());
                }
            }
        }catch (WakeupException e){
            if (!closed.get()) throw e;
        }finally {
            consumer.close();
        }
    }

    public void shutdown() {
        closed.set(true);
        consumer.wakeup();
    }
}

程序输出如下：

18/07/12 14:57:24 INFO kafka.ConsumerTask: pool-1-thread-1,threadNo=0,topic=TEST,groupId=mygrouop,offset = 73, key = 0, value = 1531378643884
18/07/12 14:57:24 INFO kafka.ConsumerTask: pool-1-thread-2,threadNo=1,topic=TEST,groupId=mygrouop,offset = 82, key = 1, value = 1531378644918
18/07/12 14:57:25 INFO kafka.ConsumerTask: pool-1-thread-3,threadNo=2,topic=TEST,groupId=mygrouop,offset = 45, key = 2, value = 1531378645919
18/07/12 14:57:26 INFO kafka.ConsumerTask: pool-1-thread-1,threadNo=0,topic=TEST,groupId=mygrouop,offset = 74, key = 3, value = 1531378646919
18/07/12 14:57:27 INFO kafka.ConsumerTask: pool-1-thread-2,threadNo=1,topic=TEST,groupId=mygrouop,offset = 83, key = 4, value = 1531378647920
18/07/12 14:57:28 INFO kafka.ConsumerTask: pool-1-thread-3,threadNo=2,topic=TEST,groupId=mygrouop,offset = 46, key = 5, value = 1531378648920
18/07/12 14:57:29 INFO kafka.ConsumerTask: pool-1-thread-1,threadNo=0,topic=TEST,groupId=mygrouop,offset = 75, key = 6, value = 1531378649920
18/07/12 14:57:30 INFO kafka.ConsumerTask: pool-1-thread-2,threadNo=1,topic=TEST,groupId=mygrouop,offset = 84, key = 7, value = 1531378650921
18/07/12 14:57:31 INFO kafka.ConsumerTask: pool-1-thread-3,threadNo=2,topic=TEST,groupId=mygrouop,offset = 47, key = 8, value = 1531378651921
18/07/12 14:57:32 INFO kafka.ConsumerTask: pool-1-thread-1,threadNo=0,topic=TEST,groupId=mygrouop,offset = 76, key = 9, value = 1531378652922
18/07/12 14:57:33 INFO kafka.ConsumerTask: pool-1-thread-2,threadNo=1,topic=TEST,groupId=mygrouop,offset = 85, key = 10, value = 1531378653922
18/07/12 14:57:34 INFO kafka.ConsumerTask: pool-1-thread-3,threadNo=2,topic=TEST,groupId=mygrouop,offset = 48, key = 11, value = 1531378654923
18/07/12 14:57:35 INFO kafka.ConsumerTask: pool-1-thread-1,threadNo=0,topic=TEST,groupId=mygrouop,offset = 77, key = 12, value = 1531378655923
18/07/12 14:57:36 INFO kafka.ConsumerTask: pool-1-thread-2,threadNo=1,topic=TEST,groupId=mygrouop,offset = 86, key = 13, value = 1531378656923
18/07/12 14:57:37 INFO kafka.ConsumerTask: pool-1-thread-3,threadNo=2,topic=TEST,groupId=mygrouop,offset = 49, key = 14, value = 1531378657924
18/07/12 14:57:38 INFO kafka.ConsumerTask: pool-1-thread-1,threadNo=0,topic=TEST,groupId=mygrouop,offset = 78, key = 15, value = 1531378658924
18/07/12 14:57:39 INFO kafka.ConsumerTask: pool-1-thread-2,threadNo=1,topic=TEST,groupId=mygrouop,offset = 87, key = 16, value = 1531378659925
18/07/12 14:57:40 INFO kafka.ConsumerTask: pool-1-thread-3,threadNo=2,topic=TEST,groupId=mygrouop,offset = 50, key = 17, value = 1531378660925
18/07/12 14:57:41 INFO kafka.ConsumerTask: pool-1-thread-1,threadNo=0,topic=TEST,groupId=mygrouop,offset = 79, key = 18, value = 1531378661926
18/07/12 14:57:42 INFO kafka.ConsumerTask: pool-1-thread-2,threadNo=1,topic=TEST,groupId=mygrouop,offset = 88, key = 19, value = 1531378662926

从输出看，三个线程各自处理自己分配的分区数据，pool-1-thread-1 ->partition 0,pool-1-thread-2 ->partition 1,pool-1-thread-3 -> partition 2。
各位可以去尝试将MultiKafkaConsumerDemo类中的消费者线程数改为4，可以看到有个线程不能消费到数据；如果将线程数改为2，可以看到一个线程消费了两个分区的数据，另一个线程消费了一个分区的数据。这是因为kafka是以分区作为并发单元的。
如果要增加消费速度，一方面可以考虑通过增加分区数，进而增加拉取线程的数量；另一方面可以通过缓存机制将消息拉取和消息处理解耦，这样则可以自定义处理线程数量，从而增加吞吐量。
此外如果要严格保证有且仅有一次的消费语义，可以通过保存offset到外部存储去实现，如保存到zookeeper或者mysql等。

本文主要讲述了消息的拉取，消息的处理，偏移量offset的提交，多线程并发消费等。

---

关于作者
爱编程、爱钻研、爱分享、爱生活
关注分布式、高并发、数据挖掘
如需捐赠，请扫码