Kafka Producer 入门

 

一些重要的参数：

1.acks指定了在给producer发送响应前,leader broker必须要确保已成功写入该消息的副本数.当前acks有3个取值,0,1,和all

2.buffer.memory：producer启动时会首先创建一块内存缓冲区用于保存待发送的消息，然后由另一个专属的线程负责从缓冲区中读取消息执行真正的发送。默认32MB。生产过快会抛异常

3.compression.type：是否压缩消息

4.retries：可重试异常充实次数，默认为0不进行重试。

5.batch.size：默认16384即16KB producer会将发往同一分区的多条消息封装进一个batch中。当batch满了之后，producer会发送batch中的所有消息。不过，producer并不总是等待batch满了才发送消息，很有可能当batch还有很多空间时producer就发送该batch。

6.linger.ms：控制消息发送延时行为，默认为0表示消息需要立即发送，无须关心batch是否已被填满

7.max.request.size：能够发送的最大消息大小.默认1048576字节 

8.request.timeout.ms当producer发送请求给broker后，broker需要在规定的时间范围内将处理结果返回给producer。默认是30秒。超时回调函数显示抛出TimeoutException异常交由用户处理。

无消息丢失配置：

producer端

block.on.full=true缓冲区填满时producer处于阻塞状态并停止接受新的消息而不是抛出异常。否则producer生产速度过快会耗尽缓冲区。新版本0.10。0.0之后可以不用理会这个参数，转而设置max.block.ms即可

ack=all所有isr同步

retries=Integer.MAX_VALUE可重试异常无限重试

max.in.flight.requests.per.connection=1设置该参数为1主要是为了防止topic同分区下的消息乱序问题。这个参数实际效果其实限制了producer在单个broker连接上能够发送的未响应请求的数量，因此，如果设置成1，则producer在某个broker发送响应之前将无法再给该broker发送producer请求。

使用带回调机制的send，

在Callback的失败处理逻辑中显式调用KafkaProducer.close(0),这样做的目的是为了处理消息的乱序问题。若不使用close(0)，默认情况下producer会被允许将未完成的消息发送出去，这样就可能造成消息乱序。但是这样还怎么重试？？？？

broker配置

unclean.leader.election.enable=false关闭unclean leader选举，即不允许非isr中的副本被选举为leader,从而避免broker端因日志水位截断而造成的消息丢失。

replication.factor>=3一致性算法三备份原则

min.insync.replicas>1用于控制某条消息至少被写入到isr中的多少个副本才算成功，设置成大于1是为了提升producer端发送语义的持久性。只有在acks被设置成all或-1时这个参数才有意义。

确保replication.factor>min.insync.replicas若2者相等，那么只要有一个副本挂掉，分区就无法正常工作，虽然有很高的持久性但可用性被极大地降低了，推荐配置成replication.factor=min.insync.replicas+1

Kafka配置实例：

private KafkaTemplate<String, String> createKafkaTemplate() {
    Map<String, Object> properties = new HashMap();
    properties.put("bootstrap.servers", this.servers);
    properties.put("acks", "0");
    properties.put("retries", 1);
    properties.put("batch.size", this.batchSize);
    properties.put("linger.ms", 1000);
    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    return new KafkaTemplate(new DefaultKafkaProducerFactory(properties));
}