Kafka【付诸实践 02】消费者和消费者群组+创建消费者实例+提交偏移量(自动、手动)+监听分区再平衡+独立的消费者+消费者其他属性说明(实例源码粘贴可用)【一篇学会使用Kafka消费者】

简介: 【2月更文挑战第21天】Kafka【付诸实践 02】消费者和消费者群组+创建消费者实例+提交偏移量(自动、手动)+监听分区再平衡+独立的消费者+消费者其他属性说明(实例源码粘贴可用)【一篇学会使用Kafka消费者】

1.消费者和消费者群组

在 Kafka 中,消费者通常是消费者群组的一部分,多个消费者群组共同读取同一个主题时,彼此之间互不影响。Kafka 之所以要引入消费者群组这个概念是因为 Kafka 消费者经常会做一些高延迟的操作,比如把数据写到数据库或HDFS ,或者进行耗时的计算,在这些情况下,单个消费者无法跟上数据生成的速度。此时可以增加更多的消费者,让它们分担负载,分别处理部分分区的消息,这就是 Kafka 实现横向伸缩的主要手段。

在这里插入图片描述
需要注意的是:同一个分区只能被同一个消费者群组里面的一个消费者读取,不可能存在同一个分区被同一个消费者群里多个消费者共同读取的情况,如图:

在这里插入图片描述

可以看到即便消费者 Consumer5 空闲了,但是也不会去读取任何一个分区的数据,这同时也提醒我们在使用时应该合理设置消费者的数量,以免造成闲置和额外开销。

1.1 分区再均衡

因为群组里的消费者共同读取主题的分区,所以当一个消费者被关闭或发生崩溃时,它就离开了群组,原本由它读取的分区将由群组里的其他消费者来读取。同时在主题发生变化时 , 比如添加了新的分区,也会发生分区与消费者的重新分配,分区的所有权从一个消费者转移到另一个消费者,这样的行为被称为再均衡。正是因为再均衡,所以消费费者群组才能保证高可用性和伸缩性。

消费者通过向群组协调器所在的 broker 发送心跳来维持它们和群组的从属关系以及它们对分区的所有权。只要消费者以正常的时间间隔发送心跳,就被认为是活跃的,说明它还在读取分区里的消息。消费者会在轮询消息或提交偏移量时发送心跳。如果消费者停止发送心跳的时间足够长,会话就会过期,群组协调器认为它已经死亡,就会触发再均衡。

2. 创建Kafka消费者

在创建消费者的时候以下以下三个选项是必选的:

  • bootstrap.servers :指定 broker 的地址清单,清单里不需要包含所有的 broker 地址,生产者会从给定的 broker里查找 broker 的信息。不过建议至少要提供两个 broker 的信息作为容错;
  • key.deserializer :指定键的反序列化器;
  • value.deserializer :指定值的反序列化器。

除此之外你还需要指明你需要想订阅的主题,可以使用如下两个 API :

  • consumer.subscribe(Collection topics) :指明需要订阅的主题的集合;
  • consumer.subscribe(Pattern pattern) :使用正则来匹配需要订阅的集合。

最后只需要通过轮询 API( poll ) 向服务器定时请求数据。一旦消费者订阅了主题,轮询就会处理所有的细节,包括群组协调、分区再均衡、发送心跳和获取数据,这使得开发者只需要关注从分区返回的数据,然后进行业务处理。 示例如下:

public class SimpleConsumer {
   
   
    public static void main(String[] args) {
   
   
        String topic = "Hello-Kafka";
        String group = "group1";
        Properties props = new Properties();
        props.put("bootstrap.servers", "tcloud:9092");
        /*指定分组 ID*/
        props.put("group.id", group);
        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);
        /*订阅主题 (s)*/
        consumer.subscribe(Collections.singletonList(topic));
        try {
   
   
            while (true) {
   
   
                /*轮询获取数据*/
                ConsumerRecords<String, String> records = consumer.poll(Duration.of(100, ChronoUnit.MILLIS));
                for (ConsumerRecord<String, String> record : records) {
   
   
                    System.out.printf("topic = %s,partition = %d, key = %s, value = %s, offset = % d,\n ", record.topic(), record.partition(), record.key(), record.value(), record.offset());
                }
            }
        } finally {
   
   
            consumer.close();
        }
    }
}

3.提交偏移量

3.1 偏移量的重要性

Kafka 的每一条消息都有一个偏移量属性,记录了其在分区中的位置,偏移量是一个单调递增的整数。消费者通过往一个叫作_consumer_offset的特殊主题发送消息,消息里包含每个分区的偏移量。 如果消费者一直处于运行状态,那么偏移量就没有什么用处。不过,如果有消费者退出或者新分区加入,此时就会触发再均衡。完成再均衡之后,每个消费者可能分配到新的分区,而不是之前处理的那个。为了能够继续之前的工作,消费者需要读取每个分区最后一次提交的偏移量,然后从偏移量指定的地方继续处理。 因为这个原因,所以如果不能正确提交偏移量,就可能会导致数据丢失或者重复出现消费,比如下面情况:

  • 如果提交的偏移量小于客户端处理的最后一个消息的偏移量 ,那么处于两个偏移量之间的消息就会被重复消费
  • 如果提交的偏移量大于客户端处理的最后一个消息的偏移量,那么处于两个偏移量之间的消息将会丢失

3.2.自动提交偏移量

Kafka 支持自动提交和手动提交偏移量两种方式。这里先介绍比较简单的自动提交:

只需要将消费者的enable.auto.commit属性配置为true即可完成自动提交的配置。 此时每隔固定的时间,消费者就会把 poll() 方法接收到的最大偏移量进行提交,提交间隔由auto.commit.interval.ms属性进行配置,默认值是5s

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

3.3.手动提交偏移量

用户可以通过将enable.auto.commit设为false,然后手动提交偏移量。基于用户需求手动提交偏移量可以分为两大类:

  • 手动提交当前偏移量:即手动提交当前轮询的最大偏移量;
  • 手动提交固定偏移量:即按照业务需求,提交某一个固定的偏移量。

而按照 Kafka API,手动提交偏移量又可以分为同步提交和异步提交。

3.3.1 同步提交

通过调用consumer.commitSync()来进行同步提交,不传递任何参数时提交的是当前轮询的最大偏移量。

public class SimpleConsumerSyn {
   
   
    public static void main(String[] args) {
   
   
        String topic = "Hello-Kafka";
        String group = "group1";
        Properties props = new Properties();
        props.put("bootstrap.servers", "tcloud:9092");
        /*指定分组 ID*/
        props.put("group.id", group);
        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);
        /*订阅主题 (s)*/
        consumer.subscribe(Collections.singletonList(topic));
        try {
   
   
            while (true) {
   
   
                /*轮询获取数据*/
                ConsumerRecords<String, String> records = consumer.poll(Duration.of(100, ChronoUnit.MILLIS));
                for (ConsumerRecord<String, String> record : records) {
   
   
                    System.out.printf("topic = %s,partition = %d, key = %s, value = %s, offset = % d,\n ", record.topic(), record.partition(), record.key(), record.value(), record.offset());
                }
                /*同步提交*/
                consumer.commitSync();
            }
        } finally {
   
   
            consumer.close();
        }
    }
}

如果某个提交失败,同步提交还会进行重试,这可以保证数据能够最大限度提交成功,但是同时也会降低程序的吞吐量。基于这个原因,Kafka 还提供了异步提交的 API。

3.3.2 异步提交

异步提交可以提高程序的吞吐量,因为此时你可以尽管请求数据,而不用等待 Broker 的响应。代码如下:

public class SimpleConsumerAsyn {
   
   
    public static void main(String[] args) {
   
   
        String topic = "Hello-Kafka";
        String group = "group1";
        Properties props = new Properties();
        props.put("bootstrap.servers", "tcloud:9092");
        /*指定分组 ID*/
        props.put("group.id", group);
        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);
        /*订阅主题 (s)*/
        consumer.subscribe(Collections.singletonList(topic));
        try {
   
   
            while (true) {
   
   
                /*异步提交并定义回调*/
                ConsumerRecords<String, String> records = consumer.poll(Duration.of(100, ChronoUnit.MILLIS));
                for (ConsumerRecord<String, String> record : records) {
   
   
                    System.out.printf("topic = %s,partition = %d, key = %s, value = %s, offset = % d,\n ", record.topic(), record.partition(), record.key(), record.value(), record.offset());
                }
                consumer.commitAsync(new OffsetCommitCallback() {
   
   
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
   
   
                        if (exception != null) {
   
   
                            System.out.println("错误处理");
                        } else {
   
   
                            offsets.forEach((x, y) -> System.out.printf("topic = %s,partition = %d, offset = %s", x.topic(), x.partition(), y.offset()));
                        }
                    }
                });
            }
        } finally {
   
   
            consumer.close();
        }
    }
}

异步提交存在的问题是,在提交失败的时候不会进行自动重试,实际上也不能进行自动重试。假设程序同时提交了 200 和 300 的偏移量,此时 200 的偏移量失败的,但是紧随其后的 300 的偏移量成功了,此时如果重试就会存在 200 覆盖 300 偏移量的可能。同步提交就不存在这个问题,因为在同步提交的情况下,300 的提交请求必须等待服务器返回 200 提交请求的成功反馈后才会发出。基于这个原因,某些情况下,需要同时组合同步和异步两种提交方式。

注:虽然程序不能在失败时候进行自动重试,但是我们是可以手动进行重试的,可以通过一个Map<TopicPartition, Integer> offsets来维护提交的每个分区的偏移量,然后当失败时候,可以判断失败的偏移量是否小于维护的同主题同分区的最后提交的偏移量,如果小于则代表已经提交了更大的偏移量请求,此时不需要重试,否则就可以进行手动重试。

3.3.3 同步加异步提交

下面这种情况,在正常的轮询中使用异步提交来保证吞吐量,但是因为在最后即将要关闭消费者了,所以此时需要用同步提交来保证最大限度的提交成功。

public class SimpleConsumerSynAndAsyn {
   
   
    public static void main(String[] args) {
   
   
        String topic = "Hello-Kafka";
        String group = "group1";
        Properties props = new Properties();
        props.put("bootstrap.servers", "tcloud:9092");
        /*指定分组 ID*/
        props.put("group.id", group);
        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);
        /*订阅主题 (s)*/
        consumer.subscribe(Collections.singletonList(topic));
        try {
   
   
            while (true) {
   
   
                /*轮询获取数据*/
                ConsumerRecords<String, String> records = consumer.poll(Duration.of(100, ChronoUnit.MILLIS));
                for (ConsumerRecord<String, String> record : records) {
   
   
                    System.out.printf("topic = %s,partition = %d, key = %s, value = %s, offset = % d,\n ", record.topic(), record.partition(), record.key(), record.value(), record.offset());
                }
                /*异步提交*/
                consumer.commitAsync();
            }
        } finally {
   
   
            try {
   
   
                // 因为即将要关闭消费者,所以要用同步提交保证提交成功
                consumer.commitSync();
            } finally {
   
   
                consumer.close();
            }
        }
    }
}

3.3.4 提交特定偏移量

在上面同步和异步提交的 API 中,实际上我们都没有对 commit 方法传递参数,此时默认提交的是当前轮询的最大偏移量,如果你需要提交特定的偏移量,可以调用它们的重载方法。需要注意的是,因为可以订阅多个主题,所以 offsets 中必须要包含所有主题的每个分区的偏移量,示例代码如下:

/*同步提交特定偏移量*/
commitSync(Map<TopicPartition, OffsetAndMetadata> offsets)
/*异步提交特定偏移量*/  
commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)

需要注意的是,由于可以订阅多个主题,所以 offsets 中必须要包含所有主题的每个分区的偏移量,示例代码如下:

public class SimpleConsumerSpecific {
   
   
    public static void main(String[] args) {
   
   
        String topic = "Hello-Kafka";
        String group = "group1";
        Properties props = new Properties();
        props.put("bootstrap.servers", "tcloud:9092");
        /*指定分组 ID*/
        props.put("group.id", group);
        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);
        /*订阅主题 (s)*/
        consumer.subscribe(Collections.singletonList(topic));
        try {
   
   
            while (true) {
   
   
                /*轮询获取数据*/
                ConsumerRecords<String, String> records = consumer.poll(Duration.of(100, ChronoUnit.MILLIS));
                Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(16);
                for (ConsumerRecord<String, String> record : records) {
   
   
                    System.out.printf("topic = %s,partition = %d, key = %s, value = %s, offset = % d,\n ", record.topic(), record.partition(), record.key(), record.value(), record.offset());
                    /*记录每个主题的每个分区的偏移量*/
                    TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
                    OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(record.offset() + 1, "no metaData");
                    /*TopicPartition 重写过 hashCode 和 equals 方法,所以能够保证同一主题和分区的实例不会被重复添加*/
                    offsets.put(topicPartition, offsetAndMetadata);
                }
                /*提交特定偏移量*/
                consumer.commitAsync(offsets, null);
            }
        } finally {
   
   
            consumer.close();
        }
    }
}

4.监听分区再均衡

因为分区再均衡会导致分区与消费者的重新划分,有时候可能希望在再均衡前执行一些操作:比如提交已经处理但是尚未提交的偏移量,关闭数据库连接等。此时可以在订阅主题时候,调用 subscribe 的重载方法传入自定义的分区再均衡监听器。

/*订阅指定集合内的所有主题*/
subscribe(Collection<String> topics, ConsumerRebalanceListener listener)
/*使用正则匹配需要订阅的主题*/  
subscribe(Pattern pattern, ConsumerRebalanceListener listener)

代码示例如下:

public class SimpleConsumerRebalance {
   
   
    public static void main(String[] args) {
   
   
        String topic = "Hello-Kafka";
        String group = "group1";
        Properties props = new Properties();
        props.put("bootstrap.servers", "tcloud:9092");
        /*指定分组 ID*/
        props.put("group.id", group);
        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);
        /*订阅主题 (s)*/
        consumer.subscribe(Collections.singletonList(topic));

        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(16);
        consumer.subscribe(Collections.singletonList(topic), new ConsumerRebalanceListener() {
   
   
            /*该方法会在消费者停止读取消息之后,再均衡开始之前就调用*/
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
   
   
                System.out.println("再均衡即将触发");// 提交已经处理的偏移量
                consumer.commitSync(offsets);
            }

            /*该方法会在重新分配分区之后,消费者开始读取消息之前被调用*/
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
   
   
            }
        });
        try {
   
   
            while (true) {
   
   
                ConsumerRecords<String, String> records = consumer.poll(Duration.of(100, ChronoUnit.MILLIS));
                for (ConsumerRecord<String, String> record : records) {
   
   
                    System.out.println(record);
                    TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
                    OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(record.offset() + 1, "no metaData");
                    /*TopicPartition 重写过 hashCode 和 equals 方法,所以能够保证同一主题和分区的实例不会被重复添加*/
                    offsets.put(topicPartition, offsetAndMetadata);
                }
                consumer.commitAsync(offsets, null);
            }
        } finally {
   
   
            consumer.close();
        }
    }
}

5.退出轮询

Kafka 提供了consumer.wakeup()方法用于退出轮询,它通过抛出WakeupException异常来跳出循环。需要注意的是,在退出线程时最好显式的调用consumer.close(), 此时消费者会提交任何还没有提交的东西,并向群组协调器发送消息,告知自己要离开群组,接下来就会触发再均衡 ,而不需要等待会话超时。下面的示例代码为监听控制台输出,当输入exit时结束轮询,关闭消费者并退出程序

public class SimpleConsumerWakeup {
   
   
    public static void main(String[] args) {
   
   
        String topic = "Hello-Kafka";
        String group = "group1";
        Properties props = new Properties();
        props.put("bootstrap.servers", "tcloud:9092");
        /*指定分组 ID*/
        props.put("group.id", group);
        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);
        /*订阅主题 (s)*/
        consumer.subscribe(Collections.singletonList(topic));
        /*调用 wakeup 优雅的退出*/
        final Thread mainThread = Thread.currentThread();
        new Thread(() -> {
   
   
            Scanner sc = new Scanner(System.in);
            while (sc.hasNext()) {
   
   
                if ("exit".equals(sc.next())) {
   
   
                    consumer.wakeup();
                    try {
   
   
                        /*等待主线程完成提交偏移量、关闭消费者等操作*/
                        mainThread.join();
                        break;
                    } catch (InterruptedException e) {
   
   
                        e.printStackTrace();
                    }
                }
            }
        }).start();
        try {
   
   
            while (true) {
   
   
                ConsumerRecords<String, String> records = consumer.poll(Duration.of(100,
                        ChronoUnit.MILLIS));
                for (ConsumerRecord<String, String> rd : records) {
   
   
                    System.out.printf("topic = %s,partition = %d, key = %s, value = %s, offset = % d, ", rd.topic(), rd.partition(), rd.key(), rd.value(), rd.offset());
                }
            }
        } catch (WakeupException e) {
   
   
            //对于 wakeup() 调用引起的 WakeupException 异常可以不必处理
        } finally {
   
   
            consumer.close();
            System.out.println("consumer 关闭");
        }
    }
}

6.独立的消费者

因为 Kafka 的设计目标是高吞吐和低延迟,所以在 Kafka 中,消费者通常都是从属于某个群组的,这是因为单个消费者的处理能力是有限的。但是某些时候的需求可能很简单,比如可能只需要一个消费者从一个主题的所有分区或者某个特定的分区读取数据,这个时候就不需要消费者群组和再均衡了, 只需要把主题或者分区分配给消费者,然后开始读取消息井提交偏移量即可。

在这种情况下,就不需要订阅主题, 取而代之的是消费者为自己分配分区。 一个消费者可以订阅主题(并加入消费者群组),或者为自己分配分区,但不能同时做这两件事情。 分配分区的示例代码如下:

public class SimpleConsumerStandAlone {
   
   
    public static void main(String[] args) {
   
   
        String topic = "Hello-Kafka";
        String group = "group1";
        Properties props = new Properties();
        props.put("bootstrap.servers", "tcloud:9092");
        /*指定分组 ID*/
        props.put("group.id", group);
        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);
        /*订阅主题 (s)*/
        consumer.subscribe(Collections.singletonList(topic));
        try {
   
   
            List<TopicPartition> partitions = new ArrayList<>();
            List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
            /*可以指定读取哪些分区 如这里假设只读取主题的 0 分区*/
            for (PartitionInfo partition : partitionInfos) {
   
   
                if (partition.partition() == 0) {
   
   
                    partitions.add(new TopicPartition(partition.topic(), partition.partition()));
                }
            }
            // 为消费者指定分区
            consumer.assign(partitions);

            while (true) {
   
   
                ConsumerRecords<String, String> records = consumer.poll(Duration.of(100, ChronoUnit.MILLIS));
                for (ConsumerRecord<String, String> record : records) {
   
   
                    System.out.printf("partition = %s, key = %d, value = %s\n",
                            record.partition(), record.key(), record.value());
                }
                consumer.commitSync();
            }
        } finally {
   
   
            consumer.close();
        }
    }
}

6.消费者其他属性

  1. fetch.min.byte
    消费者从服务器获取记录的最小字节数。如果可用的数据量小于设置值,Broker 会等待有足够的可用数据时才会把它返回给消费者。
  2. fetch.max.wait.ms
    Broker 返回给消费者数据的等待时间,默认是 500ms。
  3. max.partition.fetch.bytes
    该属性指定了服务器从每个分区返回给消费者的最大字节数,默认为 1MB。
  4. session.timeout.ms
    消费者在被认为死亡之前可以与服务器断开连接的时间,默认是 3s。
  5. auto.offset.reset
    该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理:
    latest (默认值) :在偏移量无效的情况下,消费者将从最新的记录开始读取数据(在消费者启动之后生成的最新记录);
    earliest :在偏移量无效的情况下,消费者将从起始位置读取分区的记录。
  6. enable.auto.commit
    是否自动提交偏移量,默认值是 true。为了避免出现重复消费和数据丢失,可以把它设置为 false。
  7. client.id
    客户端 id,服务器用来识别消息的来源。
  8. max.poll.records
    单次调用 poll() 方法能够返回的记录数量。
  9. receive.buffer.bytes & send.buffer.byte
    这两个参数分别指定 TCP socket 接收和发送数据包缓冲区的大小,-1 代表使用操作系统的默认值。
目录
相关文章
|
1月前
|
消息中间件 负载均衡 Kafka
【赵渝强老师】Kafka的主题与分区
Kafka 中的消息按主题分类,生产者发送消息到特定主题,消费者订阅主题消费。主题可分多个分区,每个分区仅属一个主题。消息追加到分区时,Broker 分配唯一偏移量地址,确保消息在分区内的顺序性。Kafka 保证分区有序而非主题有序。示例中,Topic A 有 3 个分区,分区可分布于不同 Broker 上,支持负载均衡和容错。视频讲解及图示详见原文。
|
1月前
|
消息中间件 监控 负载均衡
在Kafka中,如何进行主题的分区和复制?
在Kafka中,如何进行主题的分区和复制?
|
1月前
|
消息中间件 存储 负载均衡
Apache Kafka核心概念解析:生产者、消费者与Broker
【10月更文挑战第24天】在数字化转型的大潮中,数据的实时处理能力成为了企业竞争力的重要组成部分。Apache Kafka 作为一款高性能的消息队列系统,在这一领域占据了重要地位。通过使用 Kafka,企业可以构建出高效的数据管道,实现数据的快速传输和处理。今天,我将从个人的角度出发,深入解析 Kafka 的三大核心组件——生产者、消费者与 Broker,希望能够帮助大家建立起对 Kafka 内部机制的基本理解。
79 2
|
2月前
|
消息中间件 监控 负载均衡
在Kafka中,如何进行主题的分区和复制?
在Kafka中,如何进行主题的分区和复制?
|
1月前
|
消息中间件 Kafka
【赵渝强老师】Kafka分区的副本机制
在Kafka中,每个主题可有多个分区,每个分区有多个副本。其中仅有一个副本为Leader,负责对外服务,其余为Follower。当Leader所在Broker宕机时,Follower可被选为新的Leader,实现高可用。文中附有示意图及视频讲解。
|
2月前
|
消息中间件 存储 运维
为什么说Kafka还不是完美的实时数据通道
【10月更文挑战第19天】Kafka 虽然作为数据通道被广泛应用,但在实时性、数据一致性、性能及管理方面存在局限。数据延迟受消息堆积和分区再平衡影响;数据一致性难以达到恰好一次;性能瓶颈在于网络和磁盘I/O;管理复杂性涉及集群配置与版本升级。
103 1
|
2月前
|
消息中间件 Java Kafka
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
58 1
|
4月前
|
消息中间件 Java Kafka
Kafka不重复消费的终极秘籍!解锁幂等性、偏移量、去重神器,让你的数据流稳如老狗,告别数据混乱时代!
【8月更文挑战第24天】Apache Kafka作为一款领先的分布式流处理平台,凭借其卓越的高吞吐量与低延迟特性,在大数据处理领域中占据重要地位。然而,在利用Kafka进行数据处理时,如何有效避免重复消费成为众多开发者关注的焦点。本文深入探讨了Kafka中可能出现重复消费的原因,并提出了四种实用的解决方案:利用消息偏移量手动控制消费进度;启用幂等性生产者确保消息不被重复发送;在消费者端实施去重机制;以及借助Kafka的事务支持实现精确的一次性处理。通过这些方法,开发者可根据不同的应用场景灵活选择最适合的策略,从而保障数据处理的准确性和一致性。
354 9
|
4月前
|
消息中间件 负载均衡 Java
"Kafka核心机制揭秘:深入探索Producer的高效数据发布策略与Java实战应用"
【8月更文挑战第10天】Apache Kafka作为顶级分布式流处理平台,其Producer组件是数据高效发布的引擎。Producer遵循高吞吐、低延迟等设计原则,采用分批发送、异步处理及数据压缩等技术提升性能。它支持按消息键值分区,确保数据有序并实现负载均衡;提供多种确认机制保证可靠性;具备失败重试功能确保消息最终送达。Java示例展示了基本配置与消息发送流程,体现了Producer的强大与灵活性。
73 3
|
4月前
|
vr&ar 图形学 开发者
步入未来科技前沿:全方位解读Unity在VR/AR开发中的应用技巧,带你轻松打造震撼人心的沉浸式虚拟现实与增强现实体验——附详细示例代码与实战指南
【8月更文挑战第31天】虚拟现实(VR)和增强现实(AR)技术正深刻改变生活,从教育、娱乐到医疗、工业,应用广泛。Unity作为强大的游戏开发引擎,适用于构建高质量的VR/AR应用,支持Oculus Rift、HTC Vive、Microsoft HoloLens、ARKit和ARCore等平台。本文将介绍如何使用Unity创建沉浸式虚拟体验,包括设置项目、添加相机、处理用户输入等,并通过具体示例代码展示实现过程。无论是完全沉浸式的VR体验,还是将数字内容叠加到现实世界的AR应用,Unity均提供了所需的一切工具。
162 0

热门文章

最新文章