深入解析message.max.bytes

发布时间：2023-05-19

一、message.max.bytes是什么

在Kafka中，message.max.bytes是一个非常重要的参数。它指定了broker接收或者发送的消息的最大字节数。默认值为1000000（即1MB） 当Kafka Producer发送的消息大小超过message.max.bytes设置的值，broker会拒绝接收该消息。当Kafka Consumer读取的消息大小超过message.max.bytes设置的值，broker也会拒绝发送该消息。因此，正确设置message.max.bytes可以避免生产者和消费者出现丢失数据或者其他问题

二、如何设置message.max.bytes

Kafka可以在全局级别和主题级别上设置message.max.bytes。如果在全局级别设置了message.max.bytes，则所有主题的message.max.bytes都将设置为该值。如果在主题级别设置了message.max.bytes，则该主题的message.max.bytes将优先使用 在Kafka配置文件server.properties中设置全局级别message.max.bytes参数：

message.max.bytes=10000000

在创建主题时，可以通过Kafka Topic命令指定该主题的message.max.bytes参数：

bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 3 --topic test --config max.message.bytes=5000000

三、message.max.bytes的影响

1.生产者

当Kafka Producer发送的消息大小超过message.max.bytes设置的值，会触发Producer端的RecordTooLargeException异常。生产者此时可以选择修改消息大小或者增加message.max.bytes大小，以避免数据丢失。同时，生产者还可以通过调用Producer.send()方法中的max.block.ms参数，设置最大的阻塞时间

try {
  producer.send(record).get();
} catch (ExecutionException e) {
  if (e.getCause() instanceof RecordTooLargeException) {
    // handle
  }
}

2.消费者

当Kafka Consumer读取的消息大小超过message.max.bytes设置的值，会触发Consumer端的RecordTooLargeException异常。消费者此时可以选择减小读取的消息大小或者增加message.max.bytes大小，以避免数据丢失。同时，消费者还可以通过调用Kafka Consumer API中的max.poll.records参数，限制一次poll()返回的最大记录数

while(true) {
  ConsumerRecords<k, v> records = consumer.poll(Duration.ofMillis(100));
  if (records.count() == 0) {
      continue;
  }
  try {
    for (ConsumerRecord<k, v> record : records) {
      // Process the record.
    }
  } catch (Exception e) {
    if (e instanceof RecordTooLargeException) {
      // handle
    }
  }
}

四、结论

在Kafka消息系统中，message.max.bytes是一个非常重要的参数。除了用于限制消息大小以外，还可以用于避免因数据过大导致的通信问题。在生产环境中，合理的设置message.max.bytes值可以让整个Kafka消息系统更加健壮可靠。

推荐文章