（图片来源于网络，侵删）
一、Producer 参数
对于Producer，我们可以根据情况设置一些不同的参数，以此来达到对应的效果，总结了如下参数👇
1）buffer.memory
- 用来设置
生产者内存缓冲区的大小，生产者用它缓冲要发送到broker的消息。如果应用程序发送消息到生产者的速度超过了生产者发送数据到服务器的速度，会导致生产者空间不足。这个时候，send()方法调用要么被阻塞，要么抛出异常，取决于如何设置block.on.buffer参数
2）compression.type
- 默认情况下，
消息发送时不会被压缩。该参数可以设置为snappy、gzip或lz4，它指定了消息被发到broker 之前使用哪一种压缩算法进行压缩 - snappy压缩算法由Google发明，它占用较少的CPU ，却能提供较好的性能和相当可观的压缩比，如果比较关注性能和网络带宽，可以使用这种算法
- gzip压缩算法一般会占用较多的CPU ，但会提供更高的压缩比，所以如果网络带宽比较有限，可以使用这种算法。使用压缩可以降低网络传输开销和存储开销，而这往往是向Kafka 发送消息的瓶颈所在
3）retries
- 生产者从服务器收到的错误有可能是临时性的错误，在这种情况下， retries 参数的值决定了生产者可以
重发消息次数，如果达到这个次数，生产者会放弃重试并返回错误
4）batch.size
- 当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里，该参数指定了一个批次可以使用的
内存大小，按照字节数计算
5、linger.ms
- 该参数指定了生产者在发送批次之前等待更多消息加入批次的时间
6）client.id
- 该参数可以是任意的字符串，服务器会用它来识别消息的来源，还可以用在日志和配额指标里
7）max.in.flight.requests.per.connection
- 该参数指定了生产者在收到服务器晌应之前可以发送多少个消息
8）timeout.ms 、request.timeout.ms 和metadata.fetch.timeout.ms
request.timeout.ms指定了生产者在发送数据时等待服务器返回响应的时间metadata.fetch.timeout.ms指定了生产者在获取元数据（比如目标分区的Leader是谁）时等待服务器返回响应的时间。如果等待响应超时，那么生产者要么重试发送数据，要么返回一个错误（抛出异常或执行回调）timeout.ms指定了broker 等待同步副本返回消息确认的时间，与asks 的配置相匹配，如果在指定时间内没有收到同步副本的确认，那么broker 就会返回一个错误
9）max.block.ms
- 该参数指定了在调用
send()方法或使用partitionsFor()方法获取元数据时生产者的阻塞时间
10）max.request.size：
- 该参数用于控制生产者发送的
请求大小，可以指能发送的单个消息的最大值，也可以指单个请求里面所有消息总的大小
11）receive.buffer.bytes 和 send.buffer.bytes
- 这两个参数分别指定了TCP socket 接收和发送数据包的缓冲区大小，如果它们被设为
-1，就使用操作系统的默认值
二、Consumer 参数
对于Consumer ，同样的我们可以根据情况设置一些不同的参数，以此来达到对应的效果，总结了如下参数👇
1）fetch.min.bytes
- 该属性指定了消费者从服务器消费数据的
最小字节数
2）fetch.max.wait.ms
- 我们通过
fetch.min.byte告诉Kafka ，等到有足够的数据时才把它返回给消费者
而fetch.max.wait.ms则用于指定 broker 的等待时间
3）max.partition.fetch.bytes
- 默认值是
1MB,该属性指定了服务器从每个分区里返回给消费者的最大字节数.
4）session.timeout.ms
默认3s,该属性指定了消费者在被认为死亡之前可以与服务器断开连接的时间
5）auto.offset.reset
- 该属性指定了消费者在不同的情况下的
消费策略
6）enable.auto.commit
- 该属性指定了消费者是否自动提交偏移量，默认值是true
7）partition.assignment.strategy
- 分区分配给消费者群组的
分配策略，有如下两种策略：- Range：该策略会把主题的若干个连续的分区分配给消费者
- RoundRobin：该策略把主题的所有分区轮询分配给消费
8）client.id
- 该属性可以是任意字符串， broker 用它来标识从客户端发送过来的消息，通常被用在日志、度量指标和配额里
9）max.poll.records
- 该属性用于控制单次调用
call ()方法能够返回的记录数量，可以帮你控制在轮询里需要处理的数据量
10）receive.buffer.bytes 和 send.buffer.bytes
- socket 在读写数据时用到的TCP 缓冲区也可以设置大小。如果它们被设为-1，就使用操作系统的默认值。如果生产者或消费者与broker处于不同的数据中心内，可以适当增大这些值，因为跨数据中心的网络一般都有比较高的延迟和比较低的带宽
三、Zookeeper
Zookeeper 在 Kafka 中的作用：
Kafka基于Controller的Leader Election
原理是在整个集群中选举出一个Broker作为Controller，Controller为所有Topic的所有Partition指定Leader及Follower，Kafka通过在Zookeeper 上创建/controller临时节点来实现leader选举，并在该节点中写入当前broker的信息，例如： {“version”:1,”brokerid”:1,”timestamp”:”1512018424988”}
利用Zookeeper的强一致性特性，broker争抢创建/controller临时节点，最先创建成功的broker即为leader，即先到先得原则，leader也就是集群中的controller，负责集群中所有大小事务。 当leader和zookeeper失去连接时，临时节点会删除，而其他broker会监听该节点的变化，当节点删除时，其他broker会收到事件通知，重新发起leader选举
这样做极大缓解Herd Effect问题，减轻Zookeeper负载，Controller与Leader副本及Follower副本之间通过RPC通信，高效且实时，但是由于引入Controller增加了复杂度，同时需要考虑Controller的Failover（容错）
以下为 Partition 的 Leader 选举过程👇
四、Kafka 事务
Kafka 从 0.11 版本开始引入了事务支持。事务可以保证 Kafka 在 Exactly Once 语义的基
础上，生产和消费可以跨分区和会话，要么全部成功，要么全部失败
Producer 事务
为了实现跨分区跨会话的事务，需要引入一个全局唯一的 Transaction ID，并将 Producer
获得的PID 和Transaction ID 绑定。这样当Producer 重启后就可以通过正在进行的 TransactionID 获得原来的 PID
为了管理 TransactionID，Kafka 引入了一个新的组件 Transaction Coordinator。Producer 就是通过和 Transaction Coordinator 交互获得 Transaction ID 对应的任务状态。TransactionCoordinator 还负责将事务所有写入 Kafka 的一个内部 Topic，这样即使整个服务重启，由于事务状态得到保存，进行中的事务状态可以得到恢复，从而继续进行
Consumer 事务
上述事务机制主要是从 Producer 方面考虑，对于 Consumer 而言，事务的保证就会相对
较弱，尤其时无法保证 Commit 的信息被精确消费。这是由于 Consumer 可以通过 offset 访
问任意信息，而且不同的 Segment File 生命周期不同，同一事务的消息可能会出现重启后被删除的情况
都看到这里了，点赞评论一下吧！！！
点击查看👇
【Kafka】Kafka入门解析（五）
转载：https://blog.csdn.net/qq_43733123/article/details/105688143