Kafka设计的初衷是迅速处理短小的消息,通常10K大小的消息吞吐性能最好。但有时候,咱们须要处理更大的消息,好比XML文档或JSON内容,一个消息差很少有10-100M,这种状况下,Kakfa应该如何处理?java
须要JAVA Spring Cloud大型企业分布式微服务云构建的B2B2C电子商务平台源码 一零三八七七四六二六算法
针对这个问题,有如下几个建议:spring
最好的方法是不直接传送这些大的数据。若是有共享存储,如NAS, HDFS, S3等,能够把这些大的文件存放到共享存储,而后使用Kafka来传送文件的位置信息。bash
第二个方法是,将大的消息数据切片或切块,在生产端将数据切片为10K大小,使用分区主键确保一个大消息的全部部分会被发送到同一个kafka分区(这样每一部分的拆分顺序得以保留),如此以来,当消费端使用时会将这些部分从新还原为原始的消息。服务器
第三,Kafka的生产端能够压缩消息,若是原始消息是XML,当经过压缩以后,消息可能会变得不那么大。在生产端的配置参数中使用compression.codec和commpressed.topics能够开启压缩功能,压缩算法可使用GZip或Snappy。session
不过若是上述方法都不是你须要的,而你最终仍是但愿传送大的消息,那么,则能够在kafka中设置下面一些参数:mybatis
broker 配置:mvc
message.max.bytes (默认:1000000) –app
broker能接收消息的最大字节数,这个值应该比消费端的fetch.message.max.bytes更小才对,不然broker就会由于消费端没法使用这个消息而挂起。分布式
log.segment.bytes (默认: 1GB) –
kafka数据文件的大小,确保这个数值大于一个消息的长度。通常说来使用默认值便可(通常一个消息很难大于1G,由于这是一个消息系统,而不是文件系统)。
replica.fetch.max.bytes (默认: 1MB) –
broker可复制的消息的最大字节数。这个值应该比message.max.bytes大,不然broker会接收此消息,但没法将此消息复制出去,从而形成数据丢失。
consumer 配置:
fetch.message.max.bytes (默认 1MB) – 消费者能读取的最大消息。这个值应该大于或等于message.max.bytes。
因此,若是你必定要选择kafka来传送大的消息,还有些事项须要考虑。要传送大的消息,不是当出现问题以后再来考虑如何解决,而是在一开始设计的时候,就要考虑到大消息对集群和主题的影响。
性能: 根据前面提到的性能测试,kafka在消息为10K时吞吐量达到最大,更大的消息会下降吞吐量,在设计集群的容量时,尤为要考虑这点。
可用的内存和分区数:Brokers会为每一个分区分配replica.fetch.max.bytes参数指定的内存空间,假设replica.fetch.max.bytes=1M,且有1000个分区,则须要差很少1G的内存,确保 分区数的最大的消息不会超过服务器的内存,不然会报OOM错误。一样地,消费端的fetch.message.max.bytes指定了最大消息须要的内存空间,一样,分区数*最大须要内存空间 不能超过服务器的内存。因此,若是你有大的消息要传送,则在内存必定的状况下,只能使用较少的分区数或者使用更大内存的服务器。
垃圾回收:到如今为止,我在kafka的使用中还没发现过此问题,但这应该是一个须要考虑的潜在问题。更大的消息会让GC的时间更长(由于broker须要分配更大的块),随时关注GC的日志和服务器的日志信息。若是长时间的GC致使kafka丢失了zookeeper的会话,则须要配置zookeeper.session.timeout.ms参数为更大的超时时间。 一切的一切,都须要在权衡利弊以后,再决定选用哪一个最合适的方案。
参考配置以下:
replica.fetch.max.bytes=4194304
message.max.bytes=4000000
compression.codec=snappy
max.partition.fetch.bytes=4194304
复制代码