无法向 Kafka 发送大消息答案

【问题标题】：Not able to send large messages to Kafka无法向 Kafka 发送大消息
【发布时间】：2019-01-16 23:16:31
【问题描述】：

我想从生产者向 Kafka 发送一条大消息，所以我更改了以下属性。

代理（server.properties）

replica.fetch.max.bytes=317344026
message.max.bytes=317344026
max.message.bytes=317344026
max.request.size=317344026

生产者（producer.properties）

max.request.size=3173440261

消费者（consumer.properties）

max.partition.fetch.bytes=327344026
fetch.message.max.bytes=317344026

当我使用 python Popen 和 kafta 的 cli 命令运行生产者时，仍然出现如下错误。

代码：

def producer(topic_name, content):
    p = subprocess.Popen(['/opt/kafka/kafka_2.11-0.9.0.0/bin/kafka-console-producer.sh', '--broker-list', 'localhost:9092', '--topic', 'Hello-Kafka'], stdout=subprocess.PIPE, stdin=subprocess.PIPE)
    p.stdin.write(content)
    out, err = p.communicate()
    print out

错误：

ERROR Error when sending message to topic Hello-Kafka with key: null, value: 1677562 bytes with error: The message is 1677588 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)

当我为 kafka 使用 python 模块时出现以下错误 (https://github.com/dpkp/kafka-python)

代码：

def producer(topic_name, content):
    p = KafkaProducer(bootstrap_servers='localhost:9092')
    a = p.send(topic_name, content).get()
    print a    
    p.flush()
    p.close()

错误：

kafka.errors.MessageSizeTooLargeError: [Error 10] MessageSizeTooLargeError: The message is 217344026 bytes when serialized which is larger than the maximum request size you have configured with the max_request_size configuration

我成功尝试过的一件事是将内容分成块，但如果有人有任何解决方案可以在不分割内容的情况下做到这一点。

【问题讨论】：

你用的是什么序列化？
Kafka 代理并非设计用于处理 300MB 的消息。除非您拥有大量可用内存并且是 Linux/Java 内存管理方面的专家，否则您会发现自己的性能很差。最好的策略是打破它。也就是说，在您的示例中，您没有将生产者属性文件传递给控制台生产者，因此您没有设置配置（至少按照书面规定）。
@AbhishekN，当我从文件中读取它时，我直接在字符串中发送它。
@dawsaw，“你没有将生产者属性文件传递给控制台生产者”是什么意思 - 我该怎么做？你能提供任何参考吗？

标签： python-2.7 apache-kafka max kafka-producer-api kafka-python

【解决方案1】：

kafka-console-producer.sh

调用 kafka-console-producer.sh 时，您没有使用您的 producer.properties 文件。
使用--producer.config 标志。

KafkaProducer

您的 KafkaProducer 正在使用默认值。调用时必须设置max_request_size。
见KafkaProducer doc

KafkaProducer(bootstrap_servers='localhost:9092', max_request_size=3173440261)

【讨论】：

【解决方案2】：

您的字符串大小确实很大，它并不是真正要在基于队列的系统中使用的消息，请重新考虑您的平台架构。话虽如此，您可以尝试压缩配置，看看它们是否有帮助。

Kafka数据压缩：Kafka数据压缩有两种方式，producer端和broker端。两者各有利弊，我发现（我认为其他人也推荐）生产者端压缩更好，因为它可以提供更好的批量优化。

"compression.codec"="2"
"compressed.topics"="<your-topic-name>"

（0：无压缩，1：GZIP 压缩，2：Snappy 压缩，3：LZ4 压缩）

进一步阅读： Compression ideas

【讨论】：