生产者和消费者

生产者API

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;


public class ProducerMain {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "10.4.121.218:9092");
        /**
         * acks 控制着认定请求已经完成的标准,all是最慢却最耐用的设置
         */
        props.put("acks", "all");
        /**
         * 如果请求失败了,生产者可以自动重新发送请求的次数
         */
        props.put("retries", 0);
        /**
         * 生产者为每一个partition还没发送出去的记录设置一个缓存区,此配置设置了缓存区的大小,
         * 调大参数会导致更大的批处理,但是需要更多的内存,因为会为每一个激活的partition创建
         * 这么大的缓存区
         */
        props.put("batch.size", 16384);
        /**
         * 默认情况下,即使缓存区中还有空间,也可以立即发送请求。如果要减少请求次数,可以设置
         * linger.ms > 0,这样生产者再发送请求之前会等待该毫秒数,希望有更多的记录被添加到缓
         * 存区以填满这个批次的缓存。但是如果1ms之后,仍然没有记录被添加进来,每个请求就会白
         * 白增加1ms的延迟。所以在高负载下,可以用一小部分延迟来换取更少、但更高效的请求。另
         * 外,即便linger.ms = 0,在一段时间内到达的记录也是会被同一批次处 理的
         */
        props.put("linger.ms", 1);
        /**
         * 生产者可以使用的缓存空间的总量。当请求的发送速度比请求发送到服务器的速度还快,那么缓
         * 存空间很快就会被耗尽,这时其他的请求就会被阻塞。阻塞的时长由max.block.ms决定,如果
         * 阻塞的时间超过max.block.ms的值,会抛出TimeoutException
         */
        props.put("buffer.memory", 33554432);
        /**
         * serializer指定了如何将ProducerRecord的键值对对象转换成字节。可以使用
         * ByteArraySerializer 或者 StringSerializer
         */
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        /**
         * 生产者包含一个缓冲池,它保存还未传送到集群上的记录,以及一个后台I/O线程,负责将记录转
         * 换成请求,并将其传输到集群。使用后没能正常关闭生产者会泄露缓冲池中的资源。
         */
        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        for (int i = 0; i < 100; i++) {
            /**
             *  send()方法是异步的,调用时会将记录添加到待处理记录的缓冲池中,然后立即返回
             *  这使得生产者可以对缓冲池中的各个记录进行批处理,从而提高效率
             */
            producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i)));
        }
        producer.close();
    }
    
}

消费者API

自动提交

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

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

public class AutomaticOffsetConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();

        /**
         * 通过bootstrap.servers指定要连接的brokers地址,这个列表仅用于发现集群中的其他代理,
         * 并不一定是集群中所有服务器的详尽地址列表。尽管你可能想要指定多个以防客户端连接的时候
         * 存在服务器故障。
         */
        props.put("bootstrap.servers", "10.4.121.218:9092,10.4.121.219:9092");
        /**
         * kafka用consumer group的概念,允许一组进程对消费和处理记录的工作进行划分。这些
         * 进程要么运行在同一台机器上,要么运行在高可扩展和容错的分布式集群上。所有group.id相同
         * 的消费者实例属于同一个consumer group。
         * <p>
         * 同组中的consumer可以通过 subscribe API 来动态订阅话题列表。然后,kafka会把被订阅
         * 话题中的每一条消息分发给每个consumer group的一个进程。这是通过平衡同组所有成员的分区来
         * 实现的,这样每个分区正好能分到组中的一个消费者。如果一个主题有4个分区,一个consumer group
         * 有2个进程,那么一个进程就消费两个分区。
         * <p>
         * consumer group中的成员资格是动态维护的:如果进程失败了,原本分配给它的partitions
         * 将重新分给同组中的其他进程。类似的,如果一个新的consumer被添加到consumer group中,
         * partitions将会被重新分配。这叫group rebalance。当新的partition被添加到被订阅的topic
         * 中或新建与订阅的正则表达式匹配的topic时,也会发生group rebalance。group通过定期的元数
         * 据刷新来检测是否有新的partitions,并且将新的partitions分配给组中的成员。
         * <p>
         * 在概念上,可以将consumer group看作是由多个进程组成的逻辑订阅者。作为一个多订阅用户
         * 系统,就可以支持任意多个consumer group,对于不同的consumer group,可以重复消费消息,
         * 只要订阅主题就可以了。
         */
        props.put("group.id", "test");
        
        /**
         * 每隔`auto.commit.interval.ms`,自动提交已消费的`offset`,消费者每次获取新数据时都会先把上
         * 一次poll()方法返回的最大偏移量提交上去。
         * <p>
         * 自动提交的问题是会重复消费数据,如果使用默认的 5s 提交时间间隔,在最近一次提交之后的 3s 发生了
         * rebalance,rebalance之后,消费者从最后一次提交的偏移量位置开始读取消息,又回到了 3s 之前的 
         * offset,所以在这 3s内到达的消息会被重复处理。可以通过修改提交时间间隔来更频繁地提交偏移量,
         * 减小可能出现重复消息的时间窗,但这种情况是无法完全避免的。
         */
        props.put("enable.auto.commit", "true");
        // 设置`auto.commit.interval.ms`来控制提交`offset`的频率
        props.put("auto.commit.interval.ms", "1000");
        /**
         *  当 kafka topic 对应的 partition 没有初始`offset`或如果当前`offset`在服务器上不再存在时,
         *  共支持三种消费策略。
         *  1. earliest: 当无提交的`offset`时,从分区的最开始消费
         *  2. latest(default): 当无提交的`offset`时,从分区下最新生产的的数据开始消费 
         *  3. none: 只要有一个分区不存在已提交的`offset`,则抛出异常
         *  <p>
         *  但是不论哪种策略,如果有当前`group.id`在`topic`的`partition`下有已提交的`offset`,
         *  则从提交的`offset`开始消费
         */
        props.put("auto.offset.reset", "earliest");
        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<String, String>(props);
        // 订阅两个主题 foo bar
        consumer.subscribe(Arrays.asList("foo", "bar"));
        System.out.println("主题订阅:foo bar");
        while (true) {
            /**
             * 订阅完主题之后,当调用poll方法时,consumer会被自动添加到group中。poll()是用来确保消费者
             * 存活的方法,只要你一直调用poll方法,consumer就会一直存在group中,接收分配给这个 consumer 的
             * partitions中的消息。consumer 会每隔一段时间就发送一次心跳,如果这个 consumer 崩溃了或者超过
             * session.timeout.ms设置的时间没有过发送心跳,那么这个consumer就被认为挂了,分配给这个consumer
             * 的 partitions 也会被重新分配。
             * <p>
             * 也有可能consumer一直发送心跳,但是不调用poll去处理消费信息,这样consumer会永远占用分配给它的
             * partitions,为了防止这种情况,kafka提供了一种存活检测机制, max.poll.interval.ms。即如果你
             * 调用poll()的时间间隔超过了这个值,客户端就会主动将consumer移除出group,然后重新分配partitions。
             * 这个机制确保了只有活跃的consumer才能够提交offset,所以为了consumer能保留在group中,你需要一直
             * 调用poll方法。
             * <p>
             * 有两种方法控制poll轮询
             * 1. max.poll.interval.ms:提高调用poll的最大间隔时间,这样就可以给consumer更多时间去处理从
             * poll中获取的这一批记录。这样做的缺点是会延迟组的rebalance,因为consumer只有在调用poll的时候
             * 才被添加到group,然后rebalance。你可以通过这个参数来控制rebalance的频率,但是当实际调用poll
             * 的频率达不到这个参数的要求,就会被移出group,这样就会使得处理速度会很慢。
             * 2. max.poll.records:调用一次poll最多返回的记录数。这样就能更容易的去预测一次poll间隔时间内
             * 要处理多少条记录。调小这个参数值,就可以减少间隔时间,从而减少rebalance的影响。
             * <p>
             * 对于批次消息处理时间不确定的情况,这些选项都不够。推荐方法是将消息处理移动到另一个线程(异步),
             * 这样消费者就能在处理器处理记录的同时调用poll方法。但是这样就需要禁用自动提交,在线程处理完数据
             * 后手动提交记录的offset。需要注意的是提交的offset不能领先实际的position。另外还需要暂停分区,
             * 以便在线程处理完之前的记录之后再接受从poll返回的新纪录。
             */
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }
}

手动提交

同步提交

// 把`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();
}

异步提交

手动提交有一个不足之处,在 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());
        }
        /**
         * 提交最后一个偏移量,然后继续做其他事情。
         * <p>
         * 在成功提交或碰到无法恢复的错误之前,commitSync()会一直重试,但是commitAsync()不会,
         * 它之所以不进行重试,是因为在它收到服务器响应的时候,可能有一个更大的偏移量已经提交成功。
         * <p>
         * 这也是commitAsync()不好的一个地方。假设我们发出一个请求用于提交偏移量2000,这个时候发
         * 生了短暂的通信问题,服务器收不到请求,自然也不会作出任何响应。与此同时,我们处理了另外一
         * 批消息,并成功提交了偏移量3000。如果commitAsync()重新尝试提交偏移量2000,它有可能在偏
         * 移量3000之后提交成功,这个时候如果发生 rebalance,就会出现重复消费的问题。
         */
        consumer.commitAsync();
        /*
         * commitAsync()也支持回调,在broker作出响应时会执行回调,比如可以在回调中重试失败的提交,
         * 以下为思路:
         * <p> 
         * 使用一个单调递增的序列号来维护异步提交的顺序。在每次提交偏移量之后或在回调里提交偏移量时
         * 递增序列号。在进行重试前,先检查回调的序列号和即将提交的偏移量是否相等,如果相等,说明没
         * 有新的提交,那么可以安全地进行重试。如果序列号比较大,说明有一个新的提交已经发送出去了,
         * 应该停止重试。
         */ 
        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();
}

同步和异步组合提交

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

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();
    }
}

提交特定的偏移量

不管是自动提交还是使用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();
}

ReBalance Listener

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

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

程序如何能得知集群要进行”分区再均衡”了?消费者 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!");
            }
        }
    }
}

当然你也可以选择再均衡后从头开始消费,与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.seekToBeginning(partitions);
    }
});

设置从最新消息开始消费,与props.put("auto.offset.reset", "latest");等效。

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);
    }
});

涉及到数据库的 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() 方法来确保能够及时保存偏移量,并保证消费者总是能够从正确的位置开始读取消息。

打赏一个呗

取消

感谢您的支持,我会继续努力的!

扫码支持
扫码支持
扫码打赏,一毛也是爱

打开支付宝扫一扫,即可进行扫码打赏哦