Kafka发送消息大小问题

⚠️ 本文实验的Kafka版本为2.11版本.

消息概述

kafka中的消息指的就是一条ProducerRecord,里面除了携带发送的数据之外,还包含:

• topic 发往的Topic
• partition 发往的分区
• headers 头信息
• key 数据
• value 数据
• timestamp-long 时间戳

Producer生产消息过长

在生产者发送消息的时候,并不是上面所有的信息都算在发送的消息大小.详情见下面代码.

上面的代码会将value序列化成字节数组,参与序列化的有topic,headers,key. 用来验证value是否超出长度的是ensureValidRecordSize(serializedSize);方法.

ensureValidRecordSize从两个方面验证,一个是maxRequestSize(max.request.size),另一个是totalMemorySize(buffer.memory), 只有当value的长度同时小于时,消息才可以正常发送.

private void ensureValidRecordSize(int size) {
    if (size > this.maxRequestSize)
        throw new RecordTooLargeException("The message is " + size +
                " bytes when serialized which is larger than the maximum request size you have configured with the " +
                ProducerConfig.MAX_REQUEST_SIZE_CONFIG +
                " configuration.");
    if (size > this.totalMemorySize)
        throw new RecordTooLargeException("The message is " + size +
                " bytes when serialized which is larger than the total memory buffer you have configured with the " +
                ProducerConfig.BUFFER_MEMORY_CONFIG +
                " configuration.");
}

单条消息过长或产生如下错误.

这里有个注意的点,如果只是单纯的发送消息,没有用Callback进行监控或者用Future进行获得结果,在消息过长的情况下,不会主动发出提示,

使用Future接收结果

Future<RecordMetadata> send = kafkaProducer.send(new ProducerRecord<>("topic", "key", "value"));
RecordMetadata recordMetadata = send.get();
System.out.println(recordMetadata);

Future类中get()方法, @throws ExecutionException 如果计算抛出异常,该方法将会抛出该异常.

/**
 * Waits if necessary for the computation to complete, and then
 * retrieves its result.
 *
 * @return the computed result
 * @throws CancellationException if the computation was cancelled
 * @throws ExecutionException if the computation threw an
 * exception
 * @throws InterruptedException if the current thread was interrupted
 * while waiting
 */
V get() throws InterruptedException, ExecutionException;

使用Callback进行监控

先看Kafka专门为回调写的接口.

// 英文注释省略,总的来说: 用于异步回调,当消息发送server已经被确认之后,就会调用该方法
// 该方法中的肯定有一个参数不为null,如果没有异常产生,则metadata有数据,如果有异常则相反
public void onCompletion(RecordMetadata metadata, Exception exception);

kafkaProducer.send(new ProducerRecord<>("topic", "key", "value"), new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            exception.printStackTrace();
        }
    }
});

日志Level=DEBUG

将日志的消息级别设置为DEBUG,也会给标准输出输出该警告信息.

Future和Callback总结

通过上面两种比较,不难发现Future是Java并发标准库中,并不是专门为kafka而设计,需要显示捕获异常,而Callback接口是kafka提供标准回调措施,所以应尽可能采用后者.

服务端接收消息限制

在生产者有一个限制消息的参数,而在服务端也有限制消息的参数,该参数就是
message.max.bytes,默认为1000012B (大约1MB),服务端可以接收不到1MB的数据.(在新客户端producer,消息总是经过分批group into batch的数据,详情见RecordBatch接口).

/**
 * A record batch is a container for records. In old versions of the record format (versions 0 and 1),
 * a batch consisted always of a single record if no compression was enabled, but could contain
 * many records otherwise. Newer versions (magic versions 2 and above) will generally contain many records
 * regardless of compression.
 * 在旧版本不开启消息压缩的情况下,一个batch只包含一条数据
 * 在新版本中总是会包含多条消息,不会去考虑消息是否压缩
 */
public interface RecordBatch extends Iterable<Record>{
    ...
}

设置Broker端接收消息大小

修改broker端的可以接收的消息大小,需要在broker端server.properties文件中添加message.max.bytes=100000. 数值可以修改成自己想要的,单位是byte.

生产端消息大于broker会发生什么

如果生产者设置的消息发送大小为1MB,而broker端设置的消息大小为512KB会发生什么?
答案就是broker会拒绝该消息,生产者会返回一个RecordTooLargeException. 该消息是不会被消费者消费.提示的信息为: org.apache.kafka.common.errors.RecordTooLargeException: The request included a message larger than the max message size the server will accept.

消费者消息的限制

消费者也会进行消息限制,这里介绍有关三个限制消费的参数

• fetch.max.bytes 服务端消息合集(多条)能返回的大小
• fetch.min.bytes 服务端最小返回消息的大小
• fetch.max.wait.ms 最多等待时间

如果fetch.max.wait.ms设置的时间到达,即使可以返回的消息总大小没有满足fetch.min.bytes设置的值,也会进行返回.

fetch.max.bytes设置过小

如果fetch.max.bytes设置过小会发生什么? 会是不满足条件的数据一条都不返回吗? 我们可以根据文档来查看一下.

The maximum amount of data the server should return for a fetch request. Records are fetched in batches by the consumer, and if the first record batch in the first non-empty partition of the fetch is larger than this value, the record batch will still be returned to ensure that the consumer can make progress.

英文的大意就是: fetch.max.bytes 表示服务端能返回消息的总大小. 消息是通过分批次返回给消费者. 如果在分区中的第一个消息批次大于这个值,那么该消息批次依然会返回给消费者,保证流程运行.

可以得出结论: 消费端的参数只会影响消息读取的大小.

实践fetch.max.bytes设置过小

properties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 1024);
properties.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024);
properties.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 1);
...
while (true) {
    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(Integer.MAX_VALUE));
    System.out.println(records.count());
}

启动消费者,添加上面三个参数. 指定消息批次最小最大返回的大小以及允许抓取最长的等待时间. 最后将返回的消息总数输出到标准输出.

实验结果: 因为每次发送的消息都要大于1024B,所以消费者每个批次只能返回一条数据. 最终会输出1...