参考博客：
https://www.cnblogs.com/qingyunzong/category/1212387.html
https://www.cnblogs.com/haolujun/p/9632835.html (kafka与rabbitmq区别)
https://www.cnblogs.com/alvinscript/p/17407980.html （kafka核心机制，有图）

一、概念

1.1 Broker

• Kafka 集群包含一个或多个服务器，服务器节点成为 Broker
• Broker 存储 Topic ，如果某 Topic 有 N个 Partiton，集群有N个 Broker，那么每个 Broker 存储该 Topic的一个 Partition

1.2 Topic（略）

：主题

1.3 Partition

：每个topic至少有一个partition，当生产者产生数据的时候，根据分配策略(Hash取模)，选择分区，然后将消息追加到指定的分区的末尾。每个partition有多个副本，其中有且仅有一个作为Leader，Leader是当前负责数据的读写的partition。

Partition数据路由规则

1. 指定了partition，则直接使用
2. 未指定partition，但指定key，通过对key的value进行hash，选出一个partition
3. partition和key都未指定，使用轮询选出一个partition

leader与follower同步机制具体逻辑：

1. producer先从zookeeper的 "/brokers/../.../state" 节点找到该 partition 的leader
2. producer将消息发送给该leader
3. followers从leader pull消息，写入本地log后leader发送ACK
4. leader收到所有ISR（in sync replicas）的ACK后，增加HW（hight watermark，最后commit的offset）并向 producer 发送ACK

1.4 Producer（略）

1.5 Consumer（略）

1.6 Consumer Group

• 每个Consumer属于一个特定的 Consumer Group（可为每个Consumer指定group name，若不指定group name则数据默认的group）
• 将多个消费者集中到一起去处理某一个Topic的数据
• 整个消费者组共享一组偏移量(防止数据被重复读取)，因为一个Topic有多个分区

1.6 Offset

• 可以唯一的标识一条消息
• HW（High Watermark）： 也被称为高水位，它标识了一个特定的消息偏移量（offset），消费者只能拉取到这个offset之前的消息。对于每个分区，HW是已知的，并且对消费者而言，只能消费HW之前的消息。
• LEO（Log End Offset）： 是当前日志文件中下一条待写入消息的偏移量。对于每个分区，LEO都会维护在ISR集合中的每个副本中，而ISR集合中最小的LEO即为分区的HW。一般来说，HW的值不大于LEO的值。

二、数据安全

2.1 ISR机制

• AR（（assigner replicas）：用来标识副本的全集，AR = ISR + OSR
• ISR（in-sync replicas）： 加入同步队列的副本；ISR = Leader + 没有落后太多的副本
• OSR（out-sync replicas）： 离开同步队列的副本

2.2 传输保证

Kafka默认保证 At least once，并且允许通过设置Producer异步提交来实现At most once。而Exactly once要求与外部存储系统协作，幸运的是Kafka提供的offset可以非常直接非常容易得使用这种方式。

2.3 生产者方面

Producers可以选择是否为数据的写入接收ack，有以下几种ack的选项：
request.required.acks

• acks=0：producer在ISR中leader已成功收到数据并得到确认后，发送下一条message
• acks=1：producer会等待leader的ack（这意味着producer无需等待来自Broker的确认而继续发送下一批消息）
• acks=2：leader与replicas的ack（producer需要等待ISR中所有的Follower都确认接收到数据后才算一次发送完成）

2.4 消费者方面

• 如果将consumer设置为autocommit，consumer一旦读到数据立即自动commit （Exactly once）
• 读完消息先commit，再处理消息（At most once）
• 读完消息先处理，再commit（At least one）

三、参数配置

kafka中producer中的配置参数linger.ms的含义是什么，一直不太理解 ？ - 知乎 (zhihu.com)
spring boot kafak 配置 - 布咚嘞 - 博客园 (cnblogs.com)

kafka:  
	# 指定kafka 代理地址，可以多个  
	bootstrap-servers: xxx
	# 主题分区  
	partition: 3  
	# 主题副本，副本数量不允许超过broker的数量  
	factor: 3  
	producer:  
		retries: 3  
		# 每次批量发送消息的数量  (字节数，16KB)
		batch-size: 16384  
		# 缓存容量  （32MB）
		buffer-memory: 33554432  
		# 如果消息迟迟没有达到`batch.size`，那么将尝试等待`linger.ms`时间发送。默认等待时间为0，也就是当消息到达之后立即发送
		linger: 5  
		# 指定消息key和消息体的编解码方式  
		key-serializer: org.apache.kafka.common.serialization.StringSerializer  
		value-serializer: org.apache.kafka.common.serialization.StringSerializer  
	consumer:  
		# 指定默认消费者group id  
		group-id: consumer-tutorial  
		# 当Kafka中没有初始偏移量或服务器上不再存在当前偏移量时该怎么办
			# _earliest_：自动将偏移量重置为最早的偏移量；
			# _latest_：自动将偏移量重置为最迟的偏移量；
			# _none_：如果未找到消费者组的先前偏移量，则将异常抛出给消费者；
			# _exception_：向消费者抛出异常；
		auto-offset-reset: earliest  
		# 消费者的消费记录offset是否后台自动提交
		enable-auto-commit: false  
		# 当消费者的消费记录offset是否后台自动提交时，多长时间自动提交一次（ms）
		auto-commit-interval: 100  
		# 指定消息key和消息体的编解码方式  
		key-deserializer: org.apache.kafka.common.serialization.StringDeserializer  
		value-deserializer: org.apache.kafka.common.serialization.StringDeserializer  
		# 批量一次最大拉取数据量  
		max-poll-records: 1000  
		# 指定listener 容器中的线程数，用于提高并发量  
	listener:  
		# 是否开启批量消费，true表示批量消费  
		batch-listener: true  
		concurrency: 3  
	streams:  
		application-id: MessageCenter

# kafka topic  
topic:  
	sms-mac: RXEnv{RX_MC_TOPIC_SMS_MAC}  
	sms-general: RXEnv{RX_MC_TOPIC_SMS_GENERAL}  
	im: RXEnv{RX_MC_TOPIC_IM}  
	mail: RXEnv{RX_MC_TOPIC_MAIL}  
  
# kafka group  
group:  
	sms-mac: RXEnv{RX_MC_GROUP_SMS_MAC}  
	sms-general: RXEnv{RX_MC_GROUP_SMS_GENERAL}  
	im: RXEnv{RX_MC_GROUP_IM}  
	mail: RXEnv{RX_MC_GROUP_MAIL}