kafka producer 生产者客户端参数配置

我会带着你远行 2023-08-17 16:16 152阅读 0赞

在生产者向broker发送消息时，需要配置不同的参数来确保发送成功。

acks = all　　　　　　　　　#指定分区中有多少副本必须收到这条消息，生产者才认为这条消息发送成功
        
            acks = 0　　　　　　#生产者发送消息之后不需要等待任何服务端的响应
            acks = 1     　　　#只要分区的leader副本成功写入消息，那么它就会收到服务端的成功响应
            acks = -1 或all 　 #生产者在发送消息之后，需要等待ISR中的所有副本都成功写入消息之后，才能够收到来自服务端的成功响应。
    
    batch.size = 16384        #ProducerBatch 最大缓存空间，默认16KB
        
    bootstrap.servers =[192.1.1.2:9092]    #kafka集群
        
    buffer.memory = 33554432            #RecordAccumulator消息累加器最大存储空间，默认32MB 
        
    client.id =                         #客户端id
        
    compression.type = none             #设置消息的压缩格式("gzip,snappy,lz4")\
    　　　　　　对消息压缩可以极大的减少网络传输、降低网络IO，从而提高整体性能。这是一种时间换空间的优化方式，如果对延时性要求高，则不推荐对消息进行压缩
            
    connections.max.idle.ms = 540000    #设置多久之后关闭闲置连接，默认9分钟
    
    enable.idempotence = false        
        
    interceptor.classes = []            #拦截器配置
        
    key.serializer = class org.apache.kafka.common.serialization.IntegerSerializer    #key的序列化器
        
    linger.ms = 0                        #生产者在客户端发送ProducerBatch被填满或等待时间超过linger.ms值是发出去
                                            #指定生产者发送producerBatch 之前等待更多消息加入producerRecord加入ProducerBatch的时间
        
    max.block.ms = 60000                #用来控制KafkaProducer中send()方法和partitionsFor()方法的阻塞时间，当生产者的发送缓冲区
                                            #已满，或者没有可用的元数据时，这些方法就会阻塞。
        
    max.in.flight.requests.per.connection = 5        #客户端与broker端连接最多缓存5个未响应的请求(即发送到broker端，没来得及收到响应)，
                                                        #如果超过了就不能再发送请求，可以通过这个参数大小来判断是否有消息堆积
        
    max.request.size = 1048576                        #生产者客户端能发送的消息的最大值，默认1M(不建议改，会引起联动)
        
    metadata.max.age.ms = 300000                    #元数据更新时间，5分钟
    
    partitioner.class = class com.zpb.partitioner.CustomPartition 　　#自定义key分区器,可以根据指定的key来作特殊的的相关业务
    
    retries = 0                                        #生产者重试次数，默认0，在发生异常时不进行任何的重试。在发送数据时会遇到2种异常，一种是可恢复的，一种是不可
                                                        #恢复的，如：leader的选举，网络抖动等这些异常是可以恢复的，这个时候设置retries大于0就可以进行重试，在网络稳定
                                                        #或者leader选举完后，这种异常就会消失，数据在重发时就会正常，在不可恢复异常时，如超过了max.request.size最大值
                                                        #时，这种错误是不可恢复的
        
    retry.backoff.ms = 100                            #重试之间的时间间隔，最好预估一下异常恢复的时间间隔，让重试时间大于异常恢复时间，
    
    value.serializer = class org.apache.kafka.common.serialization.StringSerializer        #value序列化器

转载于:https://www.cnblogs.com/MrRightZhao/p/11355991.html