1：fetch.min.bytes，指定消费者从broker获取消息的最小字节数，即等到有足够的数据时才把它返回给消费者

2：fetch.max.wait.ms，等待broker返回数据的最大时间，默认是500ms。fetch.min.bytes和fetch.max.wait.ms哪个条件先得到满足，就按照哪种方式返回数据

3：max.partition.fetch.bytes，指定broker从每个partition中返回给消费者的最大字节数，默认1MB

4：session.timeout.ms，指定消费者被认定死亡之前可以与服务器断开连接的时间，默认是3s

5：auto.offset.reset，消费者在读取一个没有偏移量或者偏移量无效的情况下（因为消费者长时间失效，包含偏移量的记录已经过时并被删除）该作何处理。默认是latest（消费者从最新的记录开始读取数据）。另一个值是 earliest（消费者从起始位置读取partition的记录）

6：enable.auto.commit，指定消费者是否自动提交偏移量，默认为true

7：partition.assignment.strategy，指定partition如何分配给消费者，默认是Range。Range：把Topic的若干个连续的partition分配给消费者。RoundRobin：把Topic的所有partition逐个分配给消费者

8：max.poll.records，单次调用poll方法能够返回的消息数量

前面我们提到了消费者组的重平衡，kafka 也为我们提供了这样的组件，让我们可以去发生Rebalance的时候做一些操作，实现ConsumerRebalanceListener 接口，然后在订阅kafka topic 的时候传入,一个常见的常见就是对offset 的管理。因为Rebalance 可能导致数据重复消费。

对于Kafka而言，从poll方法返回消息的那一刻开始这条消息已经算是“消费”完成了，这个时候如果发生了重平衡，你的offset 没有提交的话，重平衡之后会重复消费。所以我们希望在重平衡之前进行offset 的提交

public class ConsumerRebalance {
    private static KafkaConsumer<String, String> consumer;
    private static Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
     *  初始化消费者
    static {
        Properties configs = initConfig();
        consumer = new KafkaConsumer<String, String>(configs);
        consumer.subscribe(Arrays.asList("flink_json_source_4"), new RebalanceListener(consumer));
     * 初始化配置
    private static Properties initConfig() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", true);
        props.put("auto.commit.interval.ms", 1000);
        props.put("session.timeout.ms", 30000);
        props.put("max.poll.records", 1000);
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        return props;
    public static void main(String[] args) {
        while (true) {
            // 这里的参数指的是轮询的时间间隔，也就是多长时间去拉一次数据
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(3000));
            records.forEach((ConsumerRecord<String, String> record) -> {
                System.out.println("revice: key ===" + record.key() + " value ====" + record.value() + " topic ===" + record.topic());
                currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1, "no matadata"));
            });
            consumer.commitAsync();
    static class RebalanceListener implements ConsumerRebalanceListener {
        KafkaConsumer<String, String> consumer;
        public RebalanceListener(KafkaConsumer consumer) {
            this.consumer = consumer;
        // 在再均衡开始之前和消费者停止读取消息之后被调用。如果在这里提交偏移量，下一个接管partition的消费者就知道该从哪里开始读取了。
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            //用于跟踪偏移量的map
            consumer.commitSync(currentOffsets);
        // 在重新分配partition之后和消费者开始读取消息之前被调用。
        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            consumer.commitSync(currentOffsets);
分区是kafka 高吞吐高扩展的重要基石，也是kafka 负载均衡的基本单位，所以我们要从三个方面去理解分区 
  
是从Topic 的角度
从生产者的角度
从消费者的角度
consumer group 一方面通过优秀的设计达到了两种消息模型的实现，也实现了可扩展的消费机制
重平衡会严重影响消费者的性能，但是没有重平衡的话就不能做到消费者的高扩展。
                            
python计算男女生比例 计算男女比例程序python
                            
  1、安装扩展包cmd：pip install itchat 1. 微信好友男女比例想统计下自己微信里好友的性别比例，当然也是很简单，先获取好友列表，统计列表里性别计数import itchat
# 先登录
itchat.login()
# 获取好友列表
friends = itchat.get_friends(update=True)[0:]
# 初始化计数器，有男有女