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
处于不同的数据中心,那么可以适当增大这些值,因为跨数据中心的网络一般都有比较高的延迟和比较低的带宽。