超过这个时间的消息将被删除

发布时间：2025-06-24 13:49:59  作者：北方职教升学中心  阅读量：915

Properties props =new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");// 禁用自动提交KafkaConsumer<String, String>consumer =new KafkaConsumer<>(props);consumer.subscribe(Collections.singletonList("topic"));while(true){ConsumerRecords<String, String>records =consumer.poll(Duration.ofMillis(100));for(ConsumerRecord<String, String>record :records){// 处理消息		System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());}consumer.commitSync();// 手动提交偏移量}

五、分区越多，能够并行处理的能力越强，但过多的分区也会增加管理开销。

• 主副本（Leader）：每个分区有一个主副本，负责处理所有的读写请求。可以通过 retention.ms 配置项来指定消息在主题中保留的最大时间（以毫秒为单位）。

• 数据持久化
  Kafka 将数据持久化到磁盘，确保即使在系统崩溃后，数据仍然可以恢复。以下是 Kafka 性能瓶颈的几个主要方面：

  1. 硬件限制
     • 磁盘 I/O：Kafka 的性能在很大程度上依赖于磁盘的读写速度。副本的存在确保了数据的冗余和可用性。

       Properties props =new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("acks", "all");// 设置为 all 以确保消息不丢失KafkaProducer<String, String>producer =new KafkaProducer<>(props);producer.send(new ProducerRecord<>("topic", "key", "value"));producer.close();

  2. 消息重试机制：设置retries参数，允许生产者在发送失败时进行重试。通过设置 enable.idempotence=true，可以避免因网络问题导致的重复消息。以下是一些关键的策略和配置：

     1. 消息持久化

        • 设置合适的副本因子：在创建主题时，设置副本因子（replication factor）为大于1的值，以确保消息在多个Broker上都有备份。超过这个时间的消息将被删除。
     2. 主题和分区设计
        • 分区数量：Kafka 的性能与主题的分区数量密切相关。如何保证 Kafka 消息不被重复消费？

在使用 Apache Kafka 进行消息处理时，重复消费是一个常见的问题。Kafka 的日志文件是顺序写入的，这样可以提高写入性能。可以使用数据库或内存存储（如 Redis）来存储已处理的消息 ID。因此，为了保证消息的顺序性，生产者需要将相关的消息发送到同一个分区。Kafka 如何防止消息丢失？

Kafka 通过多种机制来防止消息丢失，确保消息的可靠性和持久性。Kafka 如何保证消息的顺序消费？

Kafka 可以通过以下几种机制来保证消息的顺序性：

1. 使用同一分区
   Kafka 将主题（Topic）划分为多个分区（Partition）。这个过程由 ZooKeeper 管理，确保系统的高可用性。相同的键会被路由到同一个分区，从而保证了同一键的消息在同一分区内的顺序性。

六、

从副本（Follower）：其他副本作为从副本，跟随主副本进行数据复制。每个分区是一个有序的消息序列，Kafka 保证同一分区内的消息是有序的。Kafka 的数据保留策略是什么？

Kafka 的数据保留策略主要通过以下几个方面来管理消息的存储和过期：

1. 时间保留策略：
   • Kafka 允许用户设置消息的保留时间。如果网络带宽不足，可能会导致消息传输延迟。

     # 设置默认的副本因子default.replication.factor=2

2. 生产者配置

   • 消息确认机制：Kafka 允许生产者在发送消息时设置确认级别。当主题的大小超过这个限制时，Kafka 会删除最旧的消息以释放空间。

     Properties props =new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("enable.idempotence", "true");// 启用幂等性KafkaProducer<String, String>producer =new KafkaProducer<>(props);

四、

// 发送带有键的消息for(int i =0;i <10;i++){producer.send(new ProducerRecord<>("my-topic", "my-key", "message-"+ i));}

同一消费者组（Consumer Group）
确保主题只被一个消费者组订阅，这样每个分区只会被 该消费者组中的一个消费者实例消费，就可以避免多个消费者同时处理同一分区的消息，从而保持消息的顺序性。为了有效地处理 Kafka 消息的重复消费，可以采取以下几种策略：

1. 消费端消息去重
   • 唯一标识符：为每条消息分配一个唯一的标识符（如 UUID），在消费时记录已处理的消息 ID。
2. 消费者性能
   • 消费速率：消费者的处理能力直接影响 Kafka 的性能。使用传统的机械硬盘（HDD）可能会成为瓶颈，而使用固态硬盘（SSD）可以显著提高性能。
   • 并发消费：增加消费者的数量可以提高消费速率，但需要合理配置消费者组，以避免重复消费和负载不均。常见的设置有：
     • acks=1：只需主副本确认。

       // 创建生产者配置Properties props =new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");// 创建生产者KafkaProducer<String, String>producer =new KafkaProducer<>(props);// 发送消息到同一个分区for(int i =0;i <10;i++){producer.send(new ProducerRecord<>("my-topic", "key", "message-"+ i));}producer.close();

     • 使用消息键
       在发送消息时，生产者可以指定一个键（Key）。Kafka 使用这个键来决定消息应该发送到哪个分区。处理消息前，先检查该标识符是否已经存在。

配置示例：

# 设置消息保留时间为 1 天retention.ms=86400000# 设置主题的最大存储大小为 1GBretention.bytes=1073741824# 设置日志段大小为 100MBsegment.bytes=104857600