Kafka 新版消费者 API(二):提交偏移量

1. 自动提交

最简单的提交方式是让消费者自动提交偏移量。如果 enable.auto.commit 被设为 true,那么每过 5s,消费者会自动把从 poll() 方法接收到的最大偏移量提交上去。提交时间间隔由 auto.commit.interval.ms 控制,默认值是5s。消费者每次获取新数据时都会先把上一次poll()方法返回的最大偏移量提交上去。

可能造成的问题:数据重复读

假设我们仍然使用默认的 5s 提交时间间隔,在最近一次提交之后的 3s 发生了再均衡,再均衡之后,消费者从最后一次提交的偏移量位置开始读取消息。这个时候偏移量已经落后了 3s,所以在这 3s内到达的消息会被重复处理。可以通过修改提交时间间隔来更频繁地提交偏移量,减小可能出现重复消息的时间窗,不过这种情况是无法完全避免的。

2. 手动提交

(1) 同步提交

// 把auto.commit.offset设为false,让应用程序决定何时提交偏移量
props.put("auto.commit.offset", false);

try{
    while(true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for(ConsumerRecord<String, String> record : records) {
            // 假设把记录内容打印出来就算处理完毕
            System.out.println("value = " + record.value() + ", topic = " + record.topic() + 
                    ", partition = " + record.partition() + ", offset = " + record.offset());
        }
        
        try{
            // 只要没有发生不可恢复的错误,commitSync() 方法会一直尝试直至提交成功
            // 如果提交失败,我们也只能把异常记录到错误日志里
            consumer.commitSync();
        }catch(CommitFailedException e) {
            System.err.println("commit  failed!" + e.getMessage());
        }
    }
}finally {
    consumer.close();
}

(2) 异步提交

手动提交有一个不足之处,在 broker 对提交请求作出回应之前,应用程序会一直阻塞,这样会限制应用程序的吞吐量。我们可以通过降低提交频率来提升吞吐量,但如果发生了再均衡,会增加重复消息的数量。

这个时候可以使用异步提交,只管发送提交请求,无需等待 broker 的响应。

// 把auto.commit.offset设为false,让应用程序决定何时提交偏移量
props.put("auto.commit.offset", false);

try{
    while(true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for(ConsumerRecord<String, String> record : records) {
            System.out.println("value = " + record.value() + ", topic = " + record.topic() + 
                    ", partition = " + record.partition() + ", offset = " + record.offset());
        }
        // 提交最后一个偏移量,然后继续做其他事情。
        consumer.commitAsync();
    }
}finally {
    consumer.close();
}

在成功提交或碰到无法恢复的错误之前,commitSync()会一直重试,但是commitAsync()不会,这也是commitAsync()不好的一个地方。它之所以不进行重试,是因为在它收到服务器响应的时候,可能有一个更大的偏移量已经提交成功。

假设我们发出一个请求用于提交偏移量2000,这个时候发生了短暂的通信问题,服务器收不到请求,自然也不会作出任何响应。与此同时,我们处理了另外一批消息,并成功提交了偏移量3000。如果commitAsync()重新尝试提交偏移量2000,它有可能在偏移量3000之后提交成功。这个时候如果发生再均衡,就会出现重复消息。

commitAsync()也支持回调,在broker作出响应时会执行回调:

// 把auto.commit.offset设为false,让应用程序决定何时提交偏移量
props.put("auto.commit.offset", false);

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "
                    + record.partition() + ", offset = " + record.offset());
        }
        consumer.commitAsync(new OffsetCommitCallback() {
            @Override
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                if(offsets != null) {
                    System.out.println("commit offset successful!");
                }
                if(exception != null) {
                    System.out.println("commit offset fail!" + exception.getMessage());
                }
            }
        });
    }
} finally {
    consumer.close();
}

可以在回调中重试失败的提交,以下为思路: 使用一个单调递增的序列号来维护异步提交的顺序。在每次提交偏移量之后或在回调里提交偏移量时递增序列号。在进行重试前,先检查回调的序列号和即将提交的偏移量是否相等,如果相等,说明没有新的提交,那么可以安全地进行重试。如果序列号比较大,说明有一个新的提交已经发送出去了,应该停止重试。

(3) 同步和异步组合提交

一般情况下,针对偶尔出现的提交失败,不进行重试不会有太大问题,因为如果提交失败是因为临时问题导致的,那么后续的提交总会有成功的。但如果这是发生在关闭消费者或再均衡前的最后一次提交,就要确保能够提交成功。

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "
                    + record.partition() + ", offset = " + record.offset());
        }
        // 如果一切正常,我们使用 commitAsync() 方法来提交
        // 这样速度更快,而且即使这次提交失败,下一次提交很可能会成功
        consumer.commitAsync();
    }
}catch (Exception e) {
    e.printStackTrace();
}finally {
    try {
        // 使用 commitSync() 方法会一直重试,直到提交成功或发生无法恢复的错误
        // 确保关闭消费者之前成功提交了偏移量
        consumer.commitSync();
    }finally {
        consumer.close();
    }
}

(4) 提交特定的偏移量

不管是自动提交还是使用commitAsync()或者commitSync()来提交偏移量,提交的都是 poll() 方法返回的那批数据的最大偏移量,想要自定义在什么时候提交偏移量可以这么做:

Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>(); 
int count = 0;

......

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "
                    + record.partition() + ", offset = " + record.offset());
            
            currentOffsets.put(new TopicPartition(record.topic(), record.partition()), 
                    new OffsetAndMetadata(record.offset() + 1, "no metadata"));
                     
            if (count % 1000 == 0) {
                //  这里调用的是 commitAsync(),不过调用 commitSync() 也是完全可以的
                // 当然,在提交特定偏移量时,仍然要处理可能发生的错误
                consumer.commitAsync(currentOffsets, null);
            }
            
            count++;         
        }
    }
}finally {
    consumer.close();
}

3. 分区再均衡监听器

消费者在退出和进行分区再均衡之前,应该做一些正确的事情:

  • 提交最后一个已处理记录的偏移量(必须做)
  • 根据之前处理数据的业务不同,你可能还需要关闭数据库连接池、清空缓存等

程序如何能得知集群要进行"分区再均衡"了?消费者 API 提供了再均衡监听器,以下程序可以做到 kafka 消费数据的 Exactly Once 语义:

package com.bonc.rdpe.kafka110.consumer;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
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.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

/**
 * @Title RebalanceListenerConsumer.java 
 * @Description 再均衡监听器
 * @Author YangYunhe
 * @Date 2018-06-27 17:35:05
 */
public class RebalanceListenerConsumer {
    
    public static void main(String[] args) {
        
        Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>(); 

        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.42.89:9092,192.168.42.89:9093,192.168.42.89:9094");
        // 把auto.commit.offset设为false,让应用程序决定何时提交偏移量
        props.put("auto.commit.offset", false);
        props.put("group.id", "dev3-yangyunhe-group001");
        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(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {
            
            /*
             * 再均衡开始之前和消费者停止读取消息之后被调用
             * 如果在这里提交偏移量,下一个接管分区的消费者就知道该从哪里开始读取了
             */
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                // 如果发生再均衡,我们要在即将失去分区所有权时提交偏移量
                // 要注意,提交的是最近处理过的偏移量,而不是批次中还在处理的最后一个偏移量
                System.out.println("Lost partitions in rebalance. Committing current offsets:" + currentOffsets);
                consumer.commitSync(currentOffsets);
            }
            
            /*
             * 在重新分配分区之后和新的消费者开始读取消息之前被调用
             */
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                
                long committedOffset = -1;
                for(TopicPartition topicPartition : partitions) {
                    // 获取该分区已经消费的偏移量
                    committedOffset = consumer.committed(topicPartition).offset();
                    // 重置偏移量到上一次提交的偏移量的下一个位置处开始消费
                    consumer.seek(topicPartition, committedOffset + 1);
                }
                
            }
        });

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "
                            + record.partition() + ", offset = " + record.offset());
                    
                    currentOffsets.put(new TopicPartition(record.topic(), record.partition()), 
                            new OffsetAndMetadata(record.offset() + 1, "no metadata"));
                }
                consumer.commitAsync(currentOffsets, null);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try{
                consumer.commitSync(currentOffsets);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                consumer.close();
                System.out.println("Closed consumer successfully!");
            }
        }
    }
}

当然你也可以选择再均衡后从头开始消费:

consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        System.out.println("starting partitions rebalance...");
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        consumer.seekToBeginning(partitions);
    }
});

以上代码与 props.put("auto.offset.reset", "earliest");是等效的。

设置从最新消息开始消费:

consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        System.out.println("starting partitions rebalance...");
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        consumer.seekToEnd(partitions);
    }
});

以上代码与props.put("auto.offset.reset", "latest");等效。

4. 涉及到数据库的 Exactly Once 语义的实现思路

当处理 Kafka 中的数据涉及到数据库时,那么即使每处理一条数据提交一次偏移量,也可以造成数据重复处理或者丢失数据,看以下为伪代码:

Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>(); 
......

while (true) { 
    ConsumerRecords<String, String> records = consumer.poll(100); 
    for (ConsumerRecord<String, String> record : records) { 

    currentOffsets.put(new TopicPartition(record.topic(), record.partition()), 
 new OffsetAndMetadata(record.offset() + 1); 

    // 处理数据
    processRecord(record); 
    // 把数据存储到数据库中
    storeRecordInDB(record); 

    // 提交偏移量
    consumer.commitAsync(currentOffsets); 
 } 
}

假设把数据存储到数据库后,没有来得及提交偏移量程序就因某种原因挂掉了,那么程序再次启动后就会重复处理数据,数据库中会有重复的数据。

如果把存储到数据库和提交偏移量在一个原子操作里完成,就可以避免这样的问题,但数据存到数据库,偏移量保存到kafka是无法实现原子操作的,而如果把数据存储到数据库中,偏移量也存储到数据库中,这样就可以利用数据库的事务来把这两个操作设为一个原子操作,同时结合再均衡监听器就可以实现 Exactly Once 语义,以下为伪代码:

consumer.subscribe(Collections<String> topics, new ConsumerRebalanceListener() {
    
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 发生分区再均衡之前,提交事务
        commitDBTransaction();
    }
    
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 再均衡之后,从数据库获得消费偏移量
        for(TopicPartition topicPartition : partitions) {
            consumer.seek(topicPartition, getOffsetFromDB(topicPartition));
        }
    }
});

/**
 * 消费之前调用一次 poll(),让消费者加入到消费组中,并获取分配的分区
 * 然后马上调用 seek() 方法定位分区的偏移量
 * seek() 设置消费偏移量,设置的偏移量是从数据库读出来的,说明本次设置的偏移量已经被处理过
 * 下一次调用 poll() 就会在本次设置的偏移量上加1,开始处理没有处理过的数据
 * 如果seek()发生错误,比如偏移量不存在,则会抛出异常
 */
consumer.poll(0);
for(TopicPartition topicPartition : consumer.assignment()) {
    consumer.seek(topicPartition, getOffsetFromDB(topicPartition));
}

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(1000);
    for (ConsumerRecord<String, String> record : records) {
        // 处理数据
        processRecord(record); 
        // 把数据存储到数据库中
        storeRecordInDB(record); 
        // 把偏移量存储到数据库中
        storeOffsetInDB(record.topic(), record.partition(), record.offset()); 
    }
    // 以上3步为一个事务,提交事务,这里在每个批次末尾提交一次事务,是为了提高性能
    commitDBTransaction();
}

把偏移量和记录保存到用一个外部系统来实现 Exactly Once 有很多方法,但核心思想都是:结合 ConsumerRebalanceListener 和 seek() 方法来确保能够及时保存偏移量,并保证消费者总是能够从正确的位置开始读取消息。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏PHP在线

拒绝重复造轮子,用composer搞自己的框架(2)

久负盛名的 CodeIgniter 框架是很多人的 PHP 开发入门框架,同样也是我开始学习如何从头构建一个网站的框架。在 CI中我学到了很多,其中对 MVC ...

40290
来自专栏Golang语言社区

Golang语言社区--【游戏服务器知识】多线程并发

引言:上篇文章说到了多进程并发式的服务端模型,如上一篇文章所述,进程的频繁创建会导致服务器不堪负载,那这一篇博客主要讲述的是线程模型和线程池的方式来提高服务端的...

35840
来自专栏小夜博客

LNMP1.3 安装 fileinfo 组件

40090
来自专栏Java后端技术

解决jetty7.0.pre5启动时报ClassNotFoundException: javax.interceptor.InvocationContext异常的问题

一.背景介绍:最近项目在使用maven命令行运行jetty服务器时出现ClassNotFoundException: javax.interceptor.Inv...

8110
来自专栏IT技术精选文摘

Nginx模块之Filter解析

过滤模块简介 执行时间和内容 过滤(filter)模块是过滤响应头和内容的模块,可以对回复的头和内容进行处理。它的处理时间在获取回复内容之后,向用户发送响...

29590
来自专栏有趣的django

10.socket网络编程

套接字工作流程 ? 先从服务器端说起。服务器端先初始化Socket,然后与端口绑定(bind),对端口进行监听(listen),调用accept阻塞,等待客户端...

40270
来自专栏difcareer的技术笔记

使用AndroidStudio调试AOSP源码

当我们在AOSP中加入了自己的代码,或者当我们阅读源码时,一个迫切的需求就是调试,下面讲述如何利用AndroidStudio调试Java部分的代码。

14630
来自专栏coder修行路

Python 并发编程(一)之线程

常用用法 t.is_alive() Python中线程会在一个单独的系统级别线程中执行(比如一个POSIX线程或者一个Windows线程) 这些线程将由操作系统...

30260
来自专栏php

Composer常见错误解决

执行composer install遇到错误:Your requirements could not be resolved to an installable...

35000
来自专栏高性能分布式系统设计

Go的PCRE包装在启用JIT的时候遇到的问题和解决方案

用CGO包装C的库, 如果启用了JIT的话,会有一些问题. 例如pcre用了JIT动态生成代码并JIT代码引用里自己线程的JIT Stack数据. 这时会有一个...

31570

扫码关注云+社区

领取腾讯云代金券