| 如何往 Kafka 发送大消息？ | 您所在的位置：网站首页 › kafka大量topic怎么解决 › 如何往 Kafka 发送大消息？ |
如何往 Kafka 发送大消息？
|
默认情况下，Kafka topic 中每条消息的默认限制为 1MB。这是因为在 Kafka 中，非常大的消息被认为是低效和反模式的。然而，有时候你可能需要往 Kafka 中发送大消息。在本文中我们将研究在 Kafka 中处理大消息的两种方法。 选项 1：使用外部存储将大消息（例如视频文件）发送到外部存储，在 Kafka 中只保存这些文件的引用，例如文件的 URL。外部存储可以是云存储，也可以是网络存储（NAS）或者 HDFS 等本地大型文件存储系统。  选项 2：修改 Kafka 消息大小限制（适用于大于 1MB 小于 10 MB 的消息）这里我们需要修改 broker, consumer, producer 3 个部分的配置，以允许处理更大的消息。 Broker 服务端在 broker 端有两种修改最大消息大小的方式： 1.message.max.bytes 静态参数在 broker 级别生效，影响所有的 topic，需要修改 server.properties 文件，并重启 Kafka 集群。2.max.message.bytes 动态参数在 topic 级别生效，只影响指定的 topic，修改后立即生效，无需重启 Kafka 集群。建议保留 broker 级别最大消息大小的默认值（1MB），仅在 topic 级别覆盖此设置。 可以在创建 topic 的时候指定动态配置参数，例如创建一个名叫 large-message 的 topic，指定 max.message.bytes 为 10MB。 代码语言：javascript复制kafka-topics.sh --bootstrap-server localhost:9092 \ --create --topic large-message \ --config max.message.bytes=10485880也可以在已创建的 topic 上修改该配置参数。 代码语言：javascript复制kafka-configs.sh --bootstrap-server localhost:9092 \ --alter --entity-type topics \ --entity-name large-message \ --add-config max.message.bytes=10485880查看 large-message topic 的动态配置参数。 代码语言：javascript复制kafka-configs.sh --bootstrap-server localhost:9092 \ --entity-type topics --entity-name large-message --describe # 返回结果 Dynamic configs for topic large-message are: # DYNAMIC_TOPIC_CONFIG 是此时生效的配置 10MB，DEFAULT_CONFIG 是默认配置 1M。 max.message.bytes=10485880 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:max.message.bytes=10485880, DEFAULT_CONFIG:message.max.bytes=1048588}现在我们已经修改了 topic 的最大消息大小，但这还不够，我们还需要设置 replica.fetch.max.bytes=10485880（默认也是 1MB），以便大消息可以正常复制到 broker 的副本中。 该参数是静态配置，只能在 server.properties 配置文件中修改，并且需要重启 Kafka 集群才能生效。 如果没有修改 replica.fetch.max.bytes 参数，当往 leader replica 写入大消息时，follower replica 会因为无法复制该消息产生如下报错。 代码语言：javascript复制[2022-06-17 09:15:08,717] ERROR [ReplicaManager broker=1] Error processing fetch with max size 1048576 from replica [2] on partition large-message-0: PartitionData(fetchOffset=410683670, logStartOffset=395947464, maxBytes=1048576, currentLeaderEpoch=Optional[0], lastFetchedEpoch=Optional.empty) (kafka.server.ReplicaManager) org.apache.kafka.common.errors.CorruptRecordException: Found record size 0 smaller than minimum record overhead (14) in file /data/large-message-0/00000000000410481778.log. Consumer 消费者在 consumer 端需要修改 max.partition.fetch.bytes 参数的值，以便可以消费大消息，需要确保该值大于等于 broker 上配置的 message.max.bytes，否则一旦消息大于max.partition.fetch.bytes 的值，消费者将无法拉取到这条消息，从而导致消费进度卡住。 在 CLI 中可以使用可以使用 --consumer-property 参数进行设置。 代码语言：javascript复制kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic large-message \ --from-beginning \ --consumer-property max.partition.fetch.bytes=10485880在 Java 代码中可以这样设置。 代码语言：javascript复制properties.setProperty(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, "10485880");如果你使用 Logstash 作为消费者，可以这样设置。需要注意的是，在 Logstash 中 max_partition_fetch_bytes 参数的类型在不同的版本中是不一样的，例如在 7.7 版本中是 STRING 类型，而在 7.8 版本开始变为 NUMBER 类型。 代码语言：javascript复制input { kafka { bootstrap_servers => "localhost:9092" topics => ["large-message"] max_partition_fetch_bytes => "10485880" # 设置最大消费消息大小 } } Producer 生产者在 producer 端需要修改 max.request.size 参数的值，以便可以发送大消息，要确保该值小于等于 broker 上配置的 message.max.bytes。 在 CLI 中可以使用可以使用 --consumer-property 参数进行设置。 代码语言：javascript复制kafka-console-producer.sh --bootstrap-server localhost:9092 \ --topic large-message \ --producer-property max.request.size=10485880在 Java 代码中可以这样设置。 代码语言：javascript复制properties.setProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "10485880");如果使用 Filebeat 作为生产者，可以这样设置。大于 max_message_bytes 的消息将会被丢弃，不会发送给 Kafka。 代码语言：javascript复制output: kafka: hosts: ["localhost:9092"] topic: large-message max_message_bytes: 10485880 # 设置最大生产消息大小 参考资料[1] How to send Large Messages in Apache Kafka: https://www.conduktor.io/kafka/how-to-send-large-messages-in-apache-kafka[2] Kafka input plugin: https://www.elastic.co/guide/en/logstash/7.7/plugins-inputs-kafka.html#plugins-inputs-kafka-max_partition_fetch_bytes |
【本文地址】