kafka发送超大消息设置

最近开发一cdc应用,为了测试极端情况,需要kafka传递100万条数据过去,1个G左右,由于其他环节限制,不便进行拆包(注:测下来,大包走kafka不一定性能更好,甚至可能更低)。

测试百万以上的变更数据时,报消息超过kafka broker允许的最大值,因此需要修改如下参数,保证包能够正常发送:

socket.request.max.bytes=2147483647    # 设置了socket server接收的最大请求大小
log.segment.bytes=2147483647              # kafka数据文件的大小,确保这个数值大于一个消息的长度。一般说来使用默认值即可(一般一个消息很难大于1G,因为这是一个消息系统,而不是文件系统)。
message.max.bytes=2147483647             # 设置了kafka server接收的最大消息大小,应小于等于socket.request.max.bytes
replica.fetch.max.bytes=2147483647         #每个分区试图获取的消息字节数。要大于等于message.max.bytes,否则broker会接收此消息,但无法将此消息复制出去,从而造成数据丢失。
fetch.message.max.bytes=2147483647      #每个提取请求中为每个主题分区提取的消息字节数。要大于等于message.max.bytes,否则broker就会因为消费端无法使用这个消息而挂起。

生产者可以如下设定:

kafkaProps.put("max.request.size", 2147483647);    # 要小于 message.max.bytes,也可以设置在producer.properties配置文件中kafkaProps.put("buffer.memory", 2147483647);kafkaProps.put("timeout.ms", 3000000);kafkaProps.put("request.timeout.ms", 30000000);

消费者设定如下:

props.put("request.timeout.ms", 30000000);props.put("session.timeout.ms", "3000000");props.put("fetch.max.wait.ms", "3000000");

各参数的含义可以参考kafka官方文档https://kafka.apache.org/documentation/#configuration。

kafka基础知识体系,请参考LZ学习笔记kafka学习指南(总结版)

注,各参数对内存的影响如下:Brokers会为每个分区分配replica.fetch.max.bytes参数指定的内存空间,假设replica.fetch.max.bytes=1M,且有1000个分区,则需要差不多1G的内存,确保 分区数*最大的消息不会超过服务器的内存,否则会报OOM错误。同样地,消费端的fetch.message.max.bytes指定了最大消息需要的内存空间,同样,分区数*最大需要内存空间 不能超过服务器的内存。所以,如果你有大的消息要传送,则在内存一定的情况下,只能使用较少的分区数或者使用更大内存的服务器。

  虽然上面的方法可以奏效,但是并不推荐。Kafka设计的初衷是迅速处理短小的消息,一般10K大小的消息吞吐性能最好(可参见LinkedIn的kafka性能测试)。但有时候,我们需要处理更大的消息,比如XML文档或JSON内容,一个消息差不多有10-100M,这种情况下,Kakfa应该如何处理?

针对这个问题,有以下几个建议:

  1. 最好的方法是不直接传送这些大的数据。如果有共享存储,如NAS, HDFS, S3等,可以把这些大的文件存放到共享存储,然后使用Kafka来传送文件的位置信息。
  2. 第二个方法是,将大的消息数据切片或切块,在生产端将数据切片为10K大小,使用分区主键确保一个大消息的所有部分会被发送到同一个kafka分区(这样每一部分的拆分顺序得以保留),如此以来,当消费端使用时会将这些部分重新还原为原始的消息。
  3. 第三,Kafka的生产端可以压缩消息,如果原始消息是XML,当通过压缩之后,消息可能会变得不那么大。在生产端的配置参数中使用compression.codec和commpressed.topics可以开启压缩功能,压缩算法可以使用GZip或Snappy。

原文地址:https://www.cnblogs.com/zhjh256/p/11369165.html

时间: 2024-10-10 03:31:14

kafka发送超大消息设置的相关文章

kafka 发送数据,如何确定partition源码解释

每一个record 都有一个key 其作用在于 1 作为一个record的元数据 2 用于分区,以便确定record进入到哪个,那么具体代码是如何实现的呢?我们查看下源码 partition 如果所有的message是同一个key,将会被放入同一个partition kafka发送一个消息的流程 1 判断有没有拦截器,如果有拦截器就会执行拦截器的send 2 拿到cluster 3 对key和value序列化 4 计算此消息发送到哪个partition partition计算过程具体如下: 第二

在服务端处理同步发送小消息的性能上Kafka>RocketMQ>RabbitMQ

在发送小消息的场景中,三个消息中间件的表现区分明显: Kafka的吞吐量高达17.3w/s,远超其他两个产品.这主要取决于它的队列模式保证了写磁盘的过程是线性IO.此时broker磁盘IO已达瓶颈. RocketMQ也表现不俗,吞吐量在11.6w/s,磁盘IO %util已接近100%.RocketMQ的消息写入内存后即返回ack,由单独的线程专门做刷盘的操作,所有的消息均是顺序写文件. RabbitMQ的吞吐量5.95w/s,CPU资源消耗较高.它支持AMQP协议,实现非常重量级,为了保证消息

java客户端向单机版kafka发送消息没有接收到

kafka版本:kafka_2.11-0.10.0.0 在kafka服务器命令发送消息,消费者可以接受到, 但是在java客户端向kafka发送消息时消费者接受不到, 在kafka/config/sever.properties把这个注解打开 advertised.listeners=PLAINTEXT://ip.137:9092        #本机服务器ip 意思就是说:hostname.port都会广播给producer.consumer.如果你没有配置了这个属性的话,则使用listene

Kafka发布订阅消息

Maven <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.11.0.0</version> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <

Kafka 自定义指定消息partition策略规则及DefaultPartitioner源码分析

Kafka 自定义指定消息partition策略规则及DefaultPartitioner源码分析 一.概述 kafka默认使用DefaultPartitioner类作为默认的partition策略规则,具体默认设置是在ProducerConfig类中(如下图) 二.DefaultPartitioner.class 源码分析 1.类关系图 2.源码分析 public class DefaultPartitioner implements Partitioner { //缓存map key->to

ZeroMQ接口函数之 :zmq_msg_send – 从一个socket发送一个消息帧

ZeroMQ 官方地址 :http://api.zeromq.org/4-0:zmq_msg_send zmq_msg_send(3) ØMQ Manual - ØMQ/3.2.5 Name zmq_msg_send – 从一个socket发送一个消息帧 Synopsis int zmq_msg_send (zmq_msg_t *msg, void *socket, int flags); Description zmq_msg_send函数和zmq_sendmsg(3)函数是完全相同的,只是z

微信发送模板消息代码示例

最近一个微信的项目里需要发送微信模板消息给卖家或者供应商等,微信开发其实也就按照微信的官方接口要求组装起来即可,下面简单介绍一下我的微信模板发送代码. 1.获取access token,至于access token是什么,大家可以自行微信接口文档看一下,这边不多说 获取access token我这边主要是用定时器没大概2分钟获取一次,每天获取的次数是100000次,用法如下: 1 #region 2 3 using System; 4 using System.Timers; 5 6 #endr

脱离微信客户端发送微信消息(二)

Python版本:使用微信API发送微信消息 本文代码借用朋友编写的成品代码,使用Python3编写,配合上一篇文章:<脱离微信客户端发送微信消息(一)>经过试验完全可以发送微信消息. 文件:BaseData.py Python3代码: 1 # -*- coding: utf-8 -*- 2 3 corpid="XXXXXXX" # 设置-权限设置-部门-查看CorpID 4 corpsecret="YYYYYYYYYY" # 设置-权限设置-部门-查看

实现发送模板消息的五大步骤 | 小程序

一.获取模板id 登录小程序微信公众平台(https://mp.weixin.qq.com),获取模板(模板消息->模板库) 二.页面表单组件属性设置 form组件:设置report-submit为true button组件:设置form-type为submit 示例: <form report-submit bindsubmit="submit"> <button  form-type="submit" >下单</button&