Kafka消费者

消费者与消费者组

消费者（consumer）需要订阅Kafka中的主题，并且从主题里拉取消息。Kafka的消费者通过拉模式来消费消息。与其他消息中间件不同的是，Kafka的消费者还有一层消费组（consumer group)的概念。每个消费者会归属一个消费者组，每条消息只会投递到消费组中的某个消费者。
消费者与消费者组：

消费者与消费组这种模型可以让整体的消费能力具备伸缩性，我们可以增加（减少）消费者的个数来提高（或降低）系统的消费能力。对于固定数量的分区而言，不停的增加消费者并不能一直提升消费能力。
Kafka默认的分配策略下：如果消费者个数太多大于分区的个数，则会出现某些消费者分配不到分区，此时增加消费者无意义（可通过参数partition.assignment.strategy配置）

组内有3个消费者：

组内消费者过多：
**
对于消费中间件而言，一般有2中消息投递模式：点对点（Point-to-Point,P2P)模式与发布订阅模式（Pub/Sub).
点对点是基于队列的，消息生产者发送消息到队列，消费者从队列接收消息。
发布/订阅模式定义了如何小一个内容节点发布和订阅消息，这个内容节点成为主题（Topic），主题可以认为是消息传递的中介，消息发布者将消息发布到某个主题，而消息订阅者从主题中订阅消息.主题使得消息的订阅者与发布者互相保持独立，不需要耦合，不需要接触即可保证消息的传递。发布/订阅模式在消息的一对多广播时采用。
Kafka通过消费者与消费组同时支持两种消息投递模式。

• 如果所有的消费者都属于同一个消费组，那么所有的消息都会被均衡的投递给每个消费者，即每条消息只会被一个消费者处理，这相当于的点对点模式。
• 如果所有的消费者都隶属于不同的消费组，那么所有的消息都会被广播给所有的消费组，即每条消息会被所有的消费组处理，这就相当于发布订阅模式。

消费组是一个逻辑上的概念，他将旗下的消费组归为一类，每个消费者只隶属与一个消费组。每个消费组都会有一个固定的名称，消费者在进行消费前需要配置group.id来指定消费组的名称，默认为空字符串。

消费者并非逻辑上的概念，是实际的应用实例（如JVM实例），可以是一个线程或一个进程。同一个消费者组内的消费者一般部署到不同的机器上，当然也可以部署到同一台物理机上，通过Docker来隔离。

客户端开发

一般步骤

一般正常的消费者，其消费逻辑分为几个步骤：
1）配置消费者客户端参数即创建相应的消费者实例
2）订阅主题
3）拉取消息并消费
4）提交消费位移(手动提交或自动提交）
5）关闭消费者实例（可以跟随应用来关闭）

    /**
     * 手动同步提交
     * 同步提交：commitSync(); 阻塞等待-影响消费吞吐量，提交失败会抛出异常,自动重试
     * 异步提交：commitAsync(); 异步可能会失败，导致消费出现问题
     * @throws InterruptedException
     */
    public static void manCommitOffsetWithSync() throws InterruptedException {
        // 1. 消费者配置
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 自动提交offset
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 提交offset的时间，单位ms，即1秒钟提交一次
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // 指定k-v反序列化规则
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 指定消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        // 指定clientId
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "my-client-demo");
        // 2. 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        try {
            // 订阅主题
            consumer.subscribe(Collections.singletonList("quickstart-events"));
            while (isRunning.get()){
                // 拉取数据，指定轮询时间为1秒
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    System.out.println(consumerRecord.toString());
                }
                // 同步提交
                consumer.commitSync();
                TimeUnit.SECONDS.sleep(1);
            }
        } catch (Exception e) {
            log.error("consumer error", e);
        } finally {
            consumer.close();
        }
    }

其他消费者参数

除了必需设置的消费者参数，如bootstrap.servers, group.id, key.deserializer, value.deserializer 等，一般还需要知晓一些额外的消费者参数，这些参数大多都有合适的默认值，通常不需要修改。

该策略会把主题的若干个连续的分区分配给消费者。假设消费者 C1 和消费者 C2 同时 订阅了主题 T1 和主题 T2，并且每个主题有 3 个分区。那么消费者 C1 有可能分配到这 两个主题的分区 0 和分区 1，而消费者 C2 分配到这两个主题的分区 2。因为每个主题 拥有奇数个分区，而分配是在主题内独立完成的，第一个消费者最后分配到比第二个消 费者更多的分区。只要使用了 Range 策略，而且分区数量无法被消费者数量整除，就会 出现这种情况。

- RoundRobin
该策略把主题的所有分区逐个分配给消费者。如果使用 RoundRobin 策略来给消费者 C1 和消费者 C2 分配分区，那么消费者 C1 将分到主题 T1 的分区 0 和分区 2 以及主题 T2 的分区 1，消费者 C2 将分配到主题 T1 的分区 1 以及主题 T2 的分区 0 和分区 2。一般 来说，如果所有消费者都订阅相同的主题(这种情况很常见)，RoundRobin 策略会给所 有消费者分配相同数量的分区(或最多就差一个分区)。

可以通过设置 partition.assignment.strategy 来选择分区策略。默认使用的是 org. apache.kafka.clients.consumer.RangeAssignor，这个类实现了 Range 策略，不过也可以 把它改成 org.apache.kafka.clients.consumer.RoundRobinAssignor。我们还可以使用自定 义策略，在这种情况下，partition.assignment.strategy 属性的值就是自定义类的名字 | | client.id | 该属性可以是任意字符串，broker 用它来标识从客户端发送过来的消息，通常被用在日志、 度量指标和配额里。 | | max.poll.records | 该属性用于控制单次调用 call() 方法能够返回的记录数量，可以帮你控制在轮询里需要处 理的数据量 | | receive.buffer.bytes 和 send.buffer.bytes | socket 在读写数据时用到的 TCP 缓冲区也可以设置大小。如果它们被设为 -1，就使用操 作系统的默认值。如果生产者或消费者与 broker 处于不同的数据中心内，可以适当增大这 些值，因为跨数据中心的网络一般都有比较高的延迟和比较低的带宽。 |

位移提交

对于Kafka的分区而言，写入消息成功后每条消息都会一个偏移量（offset), 用来表示消息具体的位置。而对于Kafka消费者而言，也有一个消费者offset, 一般称为位移，表示当前消费者消费分区消息的位置。

Kafka通过poll方法每次拉取一批消息，返回未消费过的消息集，因此Kafka需要针对每个分区的消费者记录它上次消费确认过的位移(consumer-offset)，并且这个上次消费位置需要持久化存储，便于某个消费者下线后重启，或再均衡时同一分区分配给其他消费者，都需要重新拉取该分区上次消费的位移。

为了能持久化这个消费位移，Kafka消费者需要显示告诉Broker，这个过程称为位移提交。在旧版本的Kafka中，提交的位移是保存在ZooKeeper中，由于ZK不适合做高并发、大数量的消息消息写，因此在新版的Kafka中，消费位移存储在Kafka内部主题：__consumer_offsets中，消费者提交位移时，直接向这个主题发送提交消息，其可靠性保证通过Kafka的分区与副本机制来保证。

提交消息位移

由于消费者是批量拉取消息，消费完成后，会找到该批次中最后一条消息的位置然后取出它的偏移量，增加1后并做提交，表示下一条要开始消费的消息位移。但是在提交消费位移的过程中，如果处理不当，会出现重复消费或消息的丢失的情况。
**
如下图所示，x表示某一次拉取分区消息最大的偏移量，当前已消费到x这个位置，表示当前消费者的消费位移为x

但是当前消费者需要提交的位移为：x+1, 对应与图中的position。在Kafka中的API中，用如下方法来获取下一条消息的位置

Consumer#position(org.apache.kafka.common.TopicPartition)

1）上次提交的位移后，消费者拉取新的批次消息后，还未处理完发生宕机或再均衡，则消息会被重复投递

2）上次提交位移后，消费者拉取新的批次消息后，还未消费完就提交了本次批次消息，但是发生宕机或再均衡，则会丢失消息

所以，处理位移的方式对消息可靠性也有很大影响，但是Kafka 提供了多种API来提交位移。

自动提交
消费者设置enable.auto.commit=true 时，Kafka会在一定的间隔时间周期内，自动提交本次拉取批次消息。间隔时间通过设置auto.commit.interval.ms ,默认为5s.
注意：这种方式存在消息重复的情况，消费者还未提交消息时发生了宕机或再均衡，则消息会被重复拉取。

手动提交**

KafkaConsumer#commitSync() // 同步提交
KafkaConsumer#commitAsync() // 异步提交

再均衡(Rebalance)

Kafka通过消费者组来管理一组消费者实例，这种机制具有很好的扩展性且具有很好的容错性。Kafka的消费组通过再均衡协议来平衡组内的消费者，Rebalance的触发条件有3个:

1. 组成员数发生变更。比如有新的Consumer实例加入组或者离开组，抑或是有Consumer实例崩溃被“踢出”组。
2. 订阅主题数发生变更。Consumer Group可以使用正则表达式的方式订阅主题，比如consumer.subscribe(Pattern.compile(“t.*c”))就表明该Group订阅所有以字母t开头、字母c结尾的主题。在Consumer Group的运行过程中，你新创建了一个满足这样条件的主题，那么该Group就会发生Rebalance。
3. 订阅主题的分区数发生变更。Kafka当前只能允许增加一个主题的分区数。当分区数增加时，就会触发订阅该主题的所有Group开启Rebalance。

再均衡让消费者具备高可用与扩展性，使得主题可以很安全的增加消费者或减少消费者。但是再均衡发生期间，组内的所有消费者都会暂停消费，也就是说在暂停这个时间内，消费组变得不可用。另外当某个分区被重新分配给其他消费者时，之前的状态也会丢失，例如上个消费者的消费位移，导致消息可能重复消费。一般情况下，我们希望能尽量避免发生再均衡。

消费者客户端订阅某个主题时，可指定一个再均衡监听器。再均衡监听器会在再均衡生命周期中回调相应的动作，用以执行一些准备或收尾的动作。如果容忍再均衡带来的重复消费问题，可以通过再均衡监听器实现里把当前消费位移做持久化或手动提交一下。

public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener);
public interface ConsumerRebalanceListener {
    void onPartitionsRevoked(Collection<TopicPartition> partitions);
    void onPartitionsAssigned(Collection<TopicPartition> partitions);
}

onPartitionsRevoked(Collection<TopicPartition> partitions) 这个方法会在再均衡发生之前和消费者停止读取消息之后被调用，参数partitions 表示再均衡前所分配的分区信息。可以用这个方法来处理消费位移的提交或保存，用来避免一些不必要的重复消费。

void onPartitionsAssigned(Collection<TopicPartition> partitions) 这个方法会在重新分配分区和消费者开始新消费之前被调用。partitions表示再均衡后分配的分区信息。

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(...);
        Map<TopicPartition, OffsetAndMetadata> currentOffsets =new HashMap<>() ;
        consumer.subscribe(Arrays.asList("demo-topic"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                // 处理方式1：手动提交一次位移
                consumer.commitSync(currentOffsets);
                currentOffsets .clear();
                // 处理方式2：持久化到DB
                // save(currentOffsets);
            }
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                for (TopicPartition tp : partitions) {
                    consumer.seek(tp, getOffsetFromDB(tp));
                }
            }
        });
        try {
            while (isRunning.get()) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));// 1秒轮询一次
                for (ConsumerRecord<String, String> record : records) {
                    // 处理消费逻辑
                    System.out.println(record);
                    // 这里记录每个分区主题的下一个提交位移
                    // 由于Key是TopicPartition,因此相同分区的消息的Offset会被覆盖,直到取最后一条消息的提交位移
                    currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1)) ;
                }
                consumer.commitAsync(currentOffsets, null);
            }
        } finally {
            consumer.close();
        }

消费者拦截器

与生产者类似，消费者可通过配置参数：org.apache.kafka.clients.consumer.ConsumerConfig#INTERCEPTOR_CLASSES_CONFIG 来指定消费者拦截器，实现的接口为：

public interface ConsumerInterceptor<K, V> extends Configurable, AutoCloseable {
    /**
    This is called just before the records are returned by poll
    This method is allowed to modify consumer records, in which case the new records will be
     * returned. There is no limitation on number of records that could be returned from this
     * method. I.e., the interceptor can filter the records or generate new records.
     Any exception thrown by this method will be caught by the caller, logged, but not propagated to the client
    */
    public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records);
    /**
     * This is called when offsets get committed.
     * <p>
     * Any exception thrown by this method will be ignored by the caller.
     *
     * @param offsets A map of offsets by partition with associated metadata
     */
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets);
}

Kafka Consumer 拉取到消息后，会在调用poll方法之前调用拦截器的onConsume()方法来对消息做定制处理，比如修改消息的内容，过滤消息。如果抛出异常，则会被捕获，不会在继续传递出来。

多线程消费

KafkaProducer是线程安全的，然而KafkaConsumer却是非线程安全的。KafkaConsumer内部通过acquire方法来检测是否只有一个线程在操作，如有其它线程调用KafkaConsuemr的方法，则抛出异常，acquire方法定义：

// currentThread holds the threadId of the current thread accessing KafkaConsumer
// and is used to prevent multi-threaded access
private static final long NO_CURRENT_THREAD = -1L
private final AtomicLong currentThread = new AtomicLong(NO_CURRENT_THREAD);
// refcount is used to allow reentrant access by the thread who has acquired currentThread
private final AtomicInteger refcount = new AtomicInteger(0);
private void acquire() {
        long threadId = Thread.currentThread().getId();
        if (threadId != currentThread.get() && !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId))
            throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access");
        refcount.incrementAndGet();
}

acuire方法实现了一个JVM级别的轻量级锁的功能，内部通过一个原子变量AtomicLong对象操作计数的方式来检测是否发生了并发，以此用来保证只有一个线程在操作。accquire方法与release方法组合使用，用以实现加锁与解锁操作。release方法如下：

    /**
     * Release the light lock protecting the consumer from multi-threaded access.
     */
    private void release() {
        if (refcount.decrementAndGet() == 0)
            currentThread.set(NO_CURRENT_THREAD);
    }

虽然KafkaConsumer对象是非线程安全的，但是仍然可以采用多线程方案来消费消息。

第一种:线程封闭

线程封闭即每个线程实例化一个KafkaConsumer对象

参考：指定的线程数量可通过获取分区数量，一般不建议大于分区数量，获取分区数量可通过如下方法获取：

List partitionsFor(String topic)

简要实现如下：

public void oneKafkaConsumerPerThread() {
        Properties props = initConfig(true);
        int consumerThreadNum = 4; 
        for(int i=0;i<consumerThreadNum;i++) {
            new KafkaConsumerPerThread(props, "demo-topic").start();
        }
    }
    private static class KafkaConsumerPerThread extends Thread {
        KafkaConsumer<String, String> consumer;
        public KafkaConsumerPerThread(Properties props, String topic) {
            this.consumer = new KafkaConsumer<>(props);
            this.consumer.subscribe(Arrays.asList(topic));
        }
        @Override
        public void run() {
            try {
                while (isRunning.get()) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));// 1秒轮询一次
                    for (ConsumerRecord<String, String> record : records) {
                        // 处理消费逻辑
                        System.out.println(record);
                    }
                    consumer.commitAsync();
                }
            } finally {
                consumer.commitSync();
                consumer.close();
            }
        }
    }

第二种：线程池消费

简要实现如下：

public void threadPoolConsume() {
        new MultiThreadConsumePool(initConfig(false),
                Runtime.getRuntime().availableProcessors(),
                "demo-topic").start();;
    }
    private static class MultiThreadConsumePool extends Thread {
        private ExecutorService executor;
        private KafkaConsumer<String, String> consumer;
        private int consumerThreadNum;
        public MultiThreadConsumePool(Properties props, int consumerThreadNum, String topic) {
            consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Collections.singleton(topic));
            this.consumerThreadNum = consumerThreadNum;
            executor =new ThreadPoolExecutor(consumerThreadNum, consumerThreadNum, 0L, TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<>(1000) , new ThreadPoolExecutor. CallerRunsPolicy() ) ;
        }
        @Override
        public void run() {
            while (isRunning.get()) {
                while (isRunning.get()) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));// 1秒轮询一次
                    if (!records.isEmpty()) {
                        executor.submit(new Runnable() {
                            @Override
                            public void run() {
                                // 消费Records
                            }
                        });
                    }
                }
            }
        }
    }

对比