Kafka生产者各种启动参数说明 - HelloWorld开发者社区

首先是启动一个生产者

 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)));
        }
    }

启动配置参数如下所示：

2018-06-19 17:34: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

acks参数：

在考虑请求完成之前，生产者要求leader收到的确认数量，这将控制发送的记录的持久性。

acks=0如果设置为零，则生产者不会等待来自服务器的任何确认。该记录将被立即添加到套接字缓冲区并被视为已发送。在这种情况下，retries不能保证服务器已经收到记录，并且配置不会生效（因为客户端通常不会知道任何故障）。为每个记录返回的偏移量将始终设置为-1。

acks=1这意味着领导者会将记录写入其本地日志中，但会在未等待所有追随者完全确认的情况下作出响应。在这种情况下，如果领导者在承认记录后但在追随者复制之前立即失败，那么记录将会丢失。

acks=all这意味着领导者将等待全套的同步副本确认记录。这保证只要至少有一个同步副本保持活动状态，记录就不会丢失。这是最强有力的保证。这相当于acks = -1设置。

batch.size

只要有多个记录被发送到同一个分区，生产者就会尝试将记录一起分成更少的请求。这有助于客户端和服务器的性能。该配置以字节为单位控制默认的批量大小。不会尝试批量大于此大小的记录。发送给brokers的请求将包含多个批次，每个分区有一个可用于发送数据的分区。

小批量大小将使批次不太常见，并可能降低吞吐量（批量大小为零将完全禁用批次）。一个非常大的批量大小可能会更浪费一点使用内存，因为我们将始终为预期的额外记录分配指定批量大小的缓冲区。

block.on.buffer.full

当我们的内存缓冲区耗尽时，我们必须停止接受新记录（块）或抛出错误。默认情况下，此设置为false，生产者不再抛出BufferExhaustException，而是使用该max.block.ms值来阻塞，之后将抛出TimeoutException。将此属性设置为true会将其设置max.block.ms为Long.MAX_VALUE。此外，如果此属性设置为true，则参数metadata.fetch.timeout.ms不再有效。

此参数已被弃用，并将在未来版本中删除，应该使用参数max.block.ms。

bootstrap.servers

用于建立到Kafka集群的初始连接的主机/端口对列表。客户端将使用所有服务器，而不管在此指定哪些服务器用于引导 - 此列表仅影响用于发现全套服务器的初始主机。该列表应该以表格的形式出现host1:port1,host2:port2,...。由于这些服务器仅用于初始连接以发现完整的群集成员资格（可能会动态更改），因此此列表不需要包含全套服务器（但您可能需要多个服务器）。

buffer.memory

生产者可用于缓冲等待发送到服务器的记录的总内存字节数。如果记录的发送速度比发送到服务器的速度快，那么生产者将会阻止max.block.ms它，然后它会抛出异常。

这个设置应该大致对应于生产者将使用的总内存，但不是硬性限制，因为不是所有生产者使用的内存都用于缓冲。一些额外的内存将用于压缩（如果启用了压缩功能）以及维护正在进行的请求。

client.id

发出请求时传递给服务器的id字符串。这样做的目的是通过允许将逻辑应用程序名称包含在服务器端请求日志中，从而能够跟踪ip / port之外的请求源，如果不手动指定，代码中会自动生成一个id。

compression.type

指定给定主题的最终压缩类型。该配置接受标准压缩编解码器（'gzip'，'snappy'，'lz4'）。它另外接受'未压缩'，这相当于没有压缩，这意味着保留制片人设置的原始压缩编解码器，也可以修改源码，自定义压缩类型。

connections.max.idle.ms

在此配置指定的毫秒数后关闭空闲连接。

interceptor.classes

用作拦截器的类的列表。通过实现ProducerInterceptor接口，您可以在生产者发布到Kafka集群之前拦截（并可能会改变）生产者收到的记录。默认情况下，没有拦截器，可自定义拦截器。

key.serializer

用于实现Serializer接口的密钥的串行器类。

linger.ms

生产者将在请求传输之间到达的任何记录归入单个批处理请求。通常情况下，这只会在记录到达速度快于发送时才发生。但是，在某些情况下，即使在中等负载下，客户端也可能希望减少请求的数量。此设置通过添加少量人工延迟来实现此目的 - 即不是立即发送记录，而是生产者将等待达到给定延迟以允许发送其他记录，以便发送可以一起批量发送。这可以被认为与TCP中的Nagle算法类似。这个设置给出了批量延迟的上限：一旦我们得到batch.size值得记录的分区，它将被立即发送而不管这个设置如何，但是如果我们为这个分区累积的字节数少于这个数字，我们将在指定的时间内“等待”，等待更多的记录出现。该设置默认为0（即无延迟）。linger.ms=5例如，设置可以减少发送请求的数量，但会对在无效负载中发送的记录添加高达5毫秒的延迟。