Apache Kafka 发布与订阅消息系统
kafka介绍：
Kafka就是一款基于发布与订阅的梢息系统。它一般被称为 “分布式提交日志”或者“分布式流平台”。文件系统或数据库提交日志用来提供所有事务的持久记录 ， 通过重放这些日志可以重建系统的状态。同样地， Kafka 的数据是按照 一定 顺序持久化保存的，可以按需读取 。 此外， Kafka 的数据分布在整个系统里，具备数据故 障保护和性能伸缩能力。
1，传统的消息队列
在了解kafka之前我们先可以与传统的MQ做对比，传统的MQ大多数是基于内存做的处理，并且消费即清除掉，而kafka则是基于硬盘存储且有序的保存。
这里讲到redis，基于内存的消息队列。当服务器宕机之后，队列内部消息是不能做到及时的持久化的，redis基于他的场景，个人觉得更适用于短队列，和缓存的用处。
2，kafka的场景
在与传统的MQ组件的对比，如果你有初步了解，其实已经很明白他们的差异，kafka被称为一个日志提交系统，用于记录用户行为以及一系列日志消息，并且支持流处理。那这里当做消息队列有什么好处呢？
如果我们需要把一个场景的数据分配给多个系统或业务进行分析或处理，那么这里kafka可以完美实现，kafka的数据基于硬盘存储，而且提供了 Consumer Group 的概念，每个系统基于自己的Consumer Group 去消费数据并记录自己的消息偏移量（offset），与其他的消费者互不干涉。
这里基于我当前项目的场景也使用到kafka作为一个日志系统
基于自己的业务场景还可以有很多使用到的地方，这里要看各自项目的需要了。
3，Kafka基本概念
topic 主题
partition 分区
offset 偏移量（位置）
broker 服务的节点
topic
主题，发布记录的消息类别，比如上图的用户行为记录类的数据都在一个主题内，一个主题会有个消费者
partition
分区，一个topic会有若干个数据分区
offset
偏移量，标识分区每条记录的位置
贴个图领会一下
每个分区都是有序的，不可变的序列记录，分区中的每条记录都分配了一个offset编号，也是分区当中每条记录的唯一标识。
使用到kafka的两个角色，生产者和消费者
kafka的生产者
从创建一个ProducerRecord对象开始（基于java语言），包含了要发送到的topic，和partition，key，value。这里如果不指定partition，则根据key值生成hash散列，再根据分区的个数做取模（类似hashmap做分桶），达到消息均衡，如果没有填key则是轮询写。接着这条记录会被加到一个批次内（batch），独立的线程会把独立的批次发送到对应的broker上。
下面演示下如果创建一个kafka的producer
Properties props = new Properties();
props.put("bootstrap.servers", "url:port");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);
三个必填属性
bootstrap.servers kafka集群地址
key.deserializer 键序列化
value.deserializer 值序列化
发送消息到kafka
ProducerRecord<String, String> producerRecord = new ProducerRecord<>("topic1","key1","value ...");
kafkaProducer.send(producerRecord);
生产者三种发送消息方式：
- 发送-并忘记
把消息发送给kafka之后不关心是否到达，大多数情况下是正常的，就算有问题，生产者也会自动开始重试，但是这种可能会出现丢失数据 - 同步发送
是用send发送返回一个Future对象，调用get方法开始等待kafka返回结果，此方法会降低kafka吞吐量
ProducerRecord<String, String> producerRecord = new ProducerRecord<>("topic1","key1","value ...");
kafkaProducer.send(producerRecord).get();
- 异步发送
调用send方法指定回调函数，服务器响应时会调用该函数
class KafkaProducerCallback implements Callback {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e == null) {
System.out.println(recordMetadata.offset());
System.out.println(recordMetadata.partition());
} else {
e.printStackTrace();
}
}
}
ProducerRecord<String, String> producerRecord = new ProducerRecord<>("topic1", "key1", "value ...");
kafkaProducer.send(producerRecord, new KafkaProducerCallback());
kafka的消费者
消费者订阅消息，并且按照partition内消息生产的顺序读取，消费者通过offset来区分是否已经读取过，消费者会把每个分区的最后读取的offset保存在 “_consumer_offset ”这个topic上，这时候消费者如果宕机，不会丢失掉读取的记录。
消费组是由若干个消费者组成的，主题当中每一个分区只会被消费组当中一个消费者消费，如果一个消费者宕机之后会有另外一个消费者接替消费，这种分区对应消费者的关系在这里叫作所有权。
一个分区只会分配给一个消费者当消费者的数量大于分区数，那么这个时候多余的消费者则会闲置。
消费者的再均衡
分区的所有权转交给另外一个消费者叫作再均衡（比如一个消费者宕机了，另外一个消费者就会继续接替他的分区，继续消费）以此机制消费者有更好的伸缩性，高可用，消费者通过像群组协调器的broker发送心跳来维持他们和消费组的从属关系，和分区所有权关系，如果消费者的心跳断掉之后，那么群组协调器（后面的文章会讲到）会触发一次再均衡
4，集群分区
集群由多个broker组成，主题分为多个分区，一个分区会在多个broker上，多个broker上的分区会有一个分区leader，该分区leader负责处理生产者和消费者对于该分区的请求。除了这个分区leader，其他的broker上的该分区称为follower（追随者），follower的主要工作是从leader分区当中复制消息，当分区中的leader节点宕机之后，会由若干个follower当中选举一个新的分区leader，依照消费者图改造一下如下图。
5，kafka属性配置
1，生产者的配置
acks：指定了有多少个副本分区收到消息之后，生产者才算是消息写入成功
- acks=0 ，生产者在成功写入消息之后是不会等待来自服务器的任何问题，也就是说中间不管出现任何问题生产者都不知道，消息也丢失了，但是这种话方式可以提高kafka的吞吐量
- acks = 1 只要集群分区leader收到消息那么生产者就会收到一个kafka的成功响应，为了避免消息丢失生产者会重试，但是如果这个时候分区leader宕机了，在分区leader选举出来之前，消息依然会丢失
- acks = all，当所有的分区follower和分区leader都收到消息之后就才会给到成功响应，他保证不止一台服务器收到消息，尽管此时有broker宕机，也能保证整个服务是可用的，并且数据一致
timeout.ms： 和acks对应如果在设定时间内没有收到所有的分区follower和leader的响应，则会抛出超时
buffer.memory：缓存区大小，如果生产者写入速度超过发送到服务器的速度，会导致生产者空间不足。这个时候要么抛出一个异常要么send方法被阻塞
这里可以设置阻塞的时间block.on.buffer.full （0.9 之前），max.block.ms（0.9之后）
compression.type：发送消息时是否压缩，默认是不会压缩的。这里提供选择压缩算法
- snappy cpu占用少
- gzip 压缩比更高
- lz4
retries：生产者收到临时性错误，此参数决定了生产者重试次数，达到这个次数会放弃重试
默认情况下重试间隔是100ms，可以通过retry.backoff.ms来设置间隔时间大小
出现临时性错误比如，leader未选举出来。
batch.size：多个消息发送到同一个分区的时候生产者，会把他们放在同一个批次内，此参数指定批次使用内存大小，批次满了之后，会把所有的消息发送出去。如果设置太小，这里就会更频繁的发送消息
linger.ms：发送消息之前等待更多的消息加入批次的时间，增加更大，每次多送一些消息，增大吞吐量，会稍有延时
max.in.flight.requests.per.connection：此参数决定了生产者收到响应之前可以发送多少个消息，值越高内存越高，吞吐量随之提升设置为1，可以保证所有消息都是按照发送的顺序写入服务，无论中间是否出现重试
request.timeout.ms：指定生产者发送数据时等待服务器的响应时间
metadata.fetch.timeout.ms：
生产者获取元数据时候等待服务器的响应时间，（比如找到分区的leader是哪台broker）
mac.block.ms：
发送消息时候获取元数据的阻塞时间，到达时间抛出超时异常
max.request.size：
此参数设置生产者发送请求的大小，与broker节点的配置message.max.bytes匹配
2，消费者的配置
fetch.min.bytes：broker在收到消费者的拉取数据请求时，可用的数据量少于fetch.min.bytes的指定大小，那么他会等到有足够的大小的时候才返回给消费者，降低消费这和broker工作负载如果当在整个topic不活跃时段，就不需要一直处理消息，可以调整稍大降低工作负载
fetch.max.wait.ms：通过fetch.min.bytes这个属性设置要足够的数据返回给消费者，而此属性，则指定broker的等待时间，默认500ms，如果没有足够的数据大小，最终导致500ms的延迟。要么满足等待时间，要么满足设定的大小。
max.partition.fetch.bytes：此属性指定了返回给消费者的最大字节数，默认1MB。
poll方法从每个分区拉取数据最多不超过max.partition.fetch.bytes 指定的字节
max.partition.fetch.bytes 能够比broker的max.message.size属性大，
否则消费者可能无法读取这些消息。
session.timeout.ms：指定在消费者死亡之前与服务器断开连接的时间，默认3s
如果消费者没有在指定的时间内发送心跳给群组协调器，就被认为死亡，此时会触发再均衡
此属性与heartbeat.interval.ms 相关，heartbeat.interval.ms指定了poll方法向协调器发送心跳的频率，heartbeat.interval.ms必须要比session.timeout.ms小，一般1:3。
此属性设置时间断可以更快的检测宕机消费者，不过可能导致意外过多的再均衡。
auto.offset.reset：该属性指定了消费者在一个没有偏移量分区，或者偏移量无效的情况下如果处理latest在偏移量无效的情况下，从最新的开始读取，earliest 在偏移量无效时候从最新的开始读取
enable.auto.commit：是否自动提交偏移量，如果是true 可以通过auto.commit.interval.ms 来控制提交偏移量的提交频率
partition.assignment.strategy：指定默认的分区的分配策略
- Range： 分配连续的分区给消费者
- RoundRobin ：均衡分配
max.poll.records
单次返回的记录数量
本章叙述一些kafka基本概念，后面会继续深入讲到消费者分区的分配以及如果解决kafka的重复消费，如有槽点欢迎指点~
