Kafka生产者参数

落日映苍穹つ 2023-06-06 12:12 7阅读 0赞

**参数实例**  
ack 确认机制  
buffer.memory 缓存大小  
compression.type 压缩类型  
retries 尝试次数  
batch.size 批次大小  
linger.ms等待时间  
client.id客户端Id

**生产者demo**

final String kafkazk="localhost:9092";
       String topic="testAPI";
        
       Properties properties = new Properties() {
       {
           put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkazk);
           put(ProducerConfig.ACKS_CONFIG, "all");
           put(ProducerConfig.RETRIES_CONFIG, 0);
           put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
           put(ProducerConfig.LINGER_MS_CONFIG, 1);
           put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
           put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
           put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
       }};
       KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        for(int i=0;i<10000000;i++){
                   producer.send(new ProducerRecord<>(topic, UUID.randomUUID().toString(), String.valueOf(i)));
           }
       }

**1.ack**  
ack=0， 生产者在成功写入消息之前不会等待任何来自服务器的相应。如果出现问题生产者是感知不到的，消息就丢失了。不过因为生产者不需要等待服务器响应，所以它可以以网络能够支持的最大速度发送消息，从而达到很高的吞吐量。  
ack=1，只要集群的首领节点收到消息，生产这就会收到一个来自服务器的成功响应。如果消息无法达到首领节点（比如首领节点崩溃，新的首领还没有被选举出来），生产者会收到一个错误响应，为了避免数据丢失，生产者会重发消息。但是，这样还有可能会导致数据丢失，如果收到写成功通知，此时首领节点还没来的及同步数据到follower节点，首领节点崩溃，就会导致数据丢失。  
ack=-1， 只有当所有参与复制的节点都收到消息时，生产这会收到一个来自服务器的成功响应，这种模式是最安全的，它可以保证不止一个服务器收到消息。

**2.buffer.memory**  
该参数用来设置生产者内存缓冲区的大小，生产者用它缓冲要发送到服务器的消息。如果应用程序发送消息的速率比写入kafka的速度要快，会导致生产者空间不足。这个时候，send()方法调用要么被阻塞，要么抛出异常，取决于如何设置max.block.ms,表示在抛出异常之前可以阻塞一段时间。

**3.compression.type**  
默认情况下，消息发送时不会被压缩。该参数可以设置成snappy,gzip或lz4，它指定了消息发送给broker之前使用哪一种压缩算法。snappy占用较少的CPU，却能提供较好的性能和相当可观的压缩比，如果比较关注性能和网络带宽，可以使用这种压缩算法。gzip压缩算法一般会占用比较多的CPU，但会提供更高的压缩，如果网络带宽有限，可以使用这种算法。使用压缩可以降低网络传输开销和存储开销，而这往往是向kafka发送消息的瓶颈所在。

**4.retires**  
生产者从服务器收到的错误有可能是临时性的错误（比如分区找不到首领）。在这种情况下，如果达到了retires设置的次数，生产者会放弃重试并返回错误。默认情况下，生产者会在每次重试之间等待100ms，可以通过retry.backoff.ms参数来修改这个时间间隔。建议在设置重试次数和等待间隔之前，测一下恢复一个崩溃节点需要多久时间（比如所有分区选举出首领需要多长时间），让总的重试时间比kafka集群从崩溃中恢复的时间长，否则生产者会过早的放弃重试。不过有些错误不是临时性错误，没办法通过重试来解决，比如消息太大错误，一般情况下，因为生产者会自动重试，所以没必要在代码逻辑里处理那些可重试的错误，只需要处理那些不可能重试错误或重试次数超过上限的情况。

**5.batch.size**  
当有多个消息要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算，而不是消息个数。当批次被填满，批次里的所有消息会被发送出去。不过生产者并不一定都会等到批次被填满才发送，半满的批次，甚至只包含一个消息的批次也可能被发送。所以就算把batch.size设置的很大，也不会造成延迟，只会占用更多的内存而已，如果设置的太小，生产者会因为频繁发送消息而增加一些额外的开销。

**6.linger.ms**  
该参数指定了生产者在发送批次之前等待更多消息加入批次的时间。KafkaProducer会在批次填满或linger.ms达到上限时把批次发送出去。默认情况下，只要有可用的线程，就算批次里只有一个消息，生产者也会把消息发送出去。把linger.ms设置成大于0的数，让生产者在发送批次前等一会，使更多的消息加入这个批次，这样做会增加延迟，但也会提高吞吐量。

**7.client.id**  
该参数可以是任意的字符串，服务器会用它识别消息的来源，还可以用在日志和配额指标里（限速）。

**8.max.in.flight.requests.per.connection**  
该参数指定了生产者在收到服务器响应之前可以发送多少消息。它的值越高，就会占用越多的内存，不过也会提升吞吐量。把它设置成1可以保证消息是按照发送的顺序写入服务器的，即使发生了重试。

**9.timeout.ms、request.timeout.ms、metadata.fetch.timeout.ms**  
**request.timeout.ms**  
指定了生产者在发送数据时等待服务器返回响应的时间，**metadata.fetch.timeout.ms**  
指定了生产这在获取元数据（比如目标分区的leader是谁）时等待服务器返回响应的时间。如果等待响应超时，那么生产这要么重试，要么返回一个错误（抛出异常或者执行回调）。  
**timeout.ms**  
指定了broker等待同步副本返回消息确认的时间，与ack的配置相匹配——如果在指定时间内没有收到同步副本的确认，那么broker会返回一个错误。

**10.max.block.ms**  
该参数指定了在调用send()方法或使用partitionsFor()方法获取元数据时生产者的阻塞时间，当生产者的缓冲区已满，或没有可用的元数据时，这些方法就会阻塞。在阻塞时间达到max.block.ms时，生产者抛出超时异常。

**11.max.request.size**  
该参数用于控制生产者发送的请求大小，它可以指定能发送的单个消息的最大值，也可以指单个请求里所有消息的总大小。broker对可接收的消息最大值也有自己的限制（message.max.size），所以两边的配置最好匹配，避免生产者发送的消息被broker拒绝。

**12.receive.buffer.bytes和send.buffer.bytes**  
这两个参数分别指定了TCP socket接收和发送数据包的缓冲区大小，如果它们被设置成-1，就使用操作系统的默认值。如果生产者或消费者与broker 处于不同的数据中心，那么可以适当增大这些值，因为跨数据中心的网络一般都有比较高的延迟和比较低的带宽。

2019-10-11 17:04:23,636 - ProducerConfig values:
        acks = all
        batch.size = 16384
        block.on.buffer.full = false
        bootstrap.servers = [localhost:9092]
        buffer.memory = 33554432
        client.id =
        compression.type = none
        connections.max.idle.ms = 540000
        interceptor.classes = null
        key.serializer = class org.apache.kafka.common.serialization.StringSerializer
        linger.ms = 1
        max.block.ms = 60000
        max.in.flight.requests.per.connection = 5
        max.request.size = 1048576
        metadata.fetch.timeout.ms = 60000
        metadata.max.age.ms = 300000
        metric.reporters = []
        metrics.num.samples = 2
        metrics.sample.window.ms = 30000
        partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
        receive.buffer.bytes = 32768
        reconnect.backoff.ms = 50
        request.timeout.ms = 30000
        retries = 0
        retry.backoff.ms = 100
        sasl.jaas.config = null
        sasl.kerberos.kinit.cmd = /usr/bin/kinit
        sasl.kerberos.min.time.before.relogin = 60000
        sasl.kerberos.service.name = null
        sasl.kerberos.ticket.renew.jitter = 0.05
        sasl.kerberos.ticket.renew.window.factor = 0.8
        sasl.mechanism = GSSAPI
        security.protocol = PLAINTEXT
        send.buffer.bytes = 131072
        ssl.cipher.suites = null
        ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
        ssl.endpoint.identification.algorithm = null
        ssl.key.password = null
        ssl.keymanager.algorithm = SunX509
        ssl.keystore.location = null
        ssl.keystore.password = null
        ssl.keystore.type = JKS
        ssl.protocol = TLS
        ssl.provider = null
        ssl.secure.random.implementation = null
        ssl.trustmanager.algorithm = PKIX
        ssl.truststore.location = null
        ssl.truststore.password = null
        ssl.truststore.type = JKS
        timeout.ms = 30000
        value.serializer = class org.apache.kafka.common.serialization.StringSerializer```