从根上讲解原理，不会深入代码细节（不建议一上来就陷入细节中），根据自己的理解，找到切入点类，抽丝剥茧，如履薄冰。

PS:本系列文章要看懂需要一些基础哈，最起码都都使用过，以下框架都是为了简化开发者的使用，框架帮助我们写了大量的样板代码。

Spring Kafka

用于简化Kafka Client的使用，避免大量样板代码，处理大量错误重试，异常重启兜底等业务之外的逻辑，让业务专注于在回调中处理消息。

Kafka

两种消费方式

• KafkaConsumer#assign：指定Topic的点对点消费，不支持消费组的模式，也就是Topic内的某个Partition能被组内的多个Consumer消费，支持指定特定的Offset消费。
• KafkaConsuemr#subscribe：订阅Topic消费，存在消费组的概念，Topic内的某个Partition仅仅允许同一个组内的某个consumer消费，topic分区的增加、减少、consumer宕机、consumer的增加和减少，都会导致consumer的rebalance，不支持指定特定的offset消费。

consumer端重要参数

• consumer.group.id：消费组ID，订阅消费的模式下，consumer的并发度应该和partition的数量一致，保持最大的消费效率，否则单个consumer会拉去多个partition，极大影响性能，topic中partiton数量的设置非常的重要。
• auto.offset.rest：消费的起始位置
  earliest：当前group.id下各分区含有offset，则从offset消费；否则从头开始消费，针对消息绝对不能丢失的场景，应该使用这种。
  latest：当前group.id下各分区含有offset，则从offset消费；否则根据时间戳消费该分区下新产生的数据，这种模式使用最多。
  none：当前group.id下各分区含有offset，则从offset消费；否则抛出异常。
• enable.auto.commit：手动或自动提交前面的谈到的offset。若true则自动提交则需要设置auto.commit.interval.ms用于控制consumer提交的频率；false则手动提交，在消息不允许丢失的场景下必须设置成手动提交。
• max.poll.interval.ms：两次poll的时间间隔，如果超过这个时间，当前consumer会被踢出组，并rebalance，server觉得你消费能力不行，可以放弃了，所以batch消费大批次的场景，必须注意这个参数。
• max.poll.records：单次poll的最大消息数量。

消息丢失

在日志大数据使用场景下，消息丢失部分影响不太大，但是金融等场景，消息必须不能丢失，所以需要业务方自行处理消息丢失的情况，鉴于此，先了解下为啥会有消息丢失。

为啥会有消息丢失？

1. producer生产导致消息丢失，概率小，但是producer的requet.required.ack参数注意
   all：表示Server端的Lead和Follower都接收才表示消息发送成功，如果server没有返回ack，会自动重发，在这种情况下，返回ack表示lead和follower都有这条消息，任何挂了，都无所谓。
   0：server端不需要相应，producer发完立即返回，不校验ack。
   1：默认值，表示当Leader返回ack则表示发送成功，如果恰好这个Lead挂了，follow没有同步成功，那么消息丢失了。
2. consumer手动提交，宁愿重复消费也避免消费不到
   #1、先commit再处理消息，如果处理消息异常或者rebalance，但是offset提交了，这个批次的消息对于消费者而言就是丢失了再也消费不到了。
   #2、先处理再commit，如果在commit之前发生异常，则会导致重复消费。
   #3、在微服务升级的时候，会大概率会打断当前的消费者，如果commit未提交，则会重复消费，已经提交，则会丢失消息。

鉴于上Kafka如何避免消息丢失？

1. producer必须等待ack全量响应再返回。
2. 必须设置成手动消息提交，当前消费者消费成功再手动提交commit，唯一的问题就是大概率有重复消费，每条消息增加唯一UUID，存储最近消费的消息。

如何去重？非常重要
使用redis的list实现有序有限队列，针对每条消息做幂等处理。

1. key为topicName:partition，存储topic和partition信息。
2. value存储最近消息的uuid，大小一般定义为max.poll.records的多少倍。

可使用RedissonList，addBefore：LPUSH，trim：LTRIM，contain实现LRANGE+Lua脚本原子化

@Overridepublic RFuture<Boolean> containsAsync(Object o) {return indexOfAsync(o, new BooleanNumberReplayConvertor(-1L));}// 如下优雅的Lua脚本实现，先range全量元素，然后foreach判断是否含有
// 如果List数量巨大，则可能存在耗损Redis性能的问题。public <R> RFuture<R> indexOfAsync(Object o, Convertor<R> convertor) {return commandExecutor.evalReadAsync(getName(), codec, new RedisCommand<R>("EVAL", convertor),"local key = KEYS[1] " +"local obj = ARGV[1] " +"local items = redis.call('lrange', key, 0, -1) " +"for i=1,#items do " +"if items[i] == obj then " +"return i - 1 " +"end " +"end " +"return -1",Collections.<Object>singletonList(getName()), encode(o));}

去重伪代码

while(true) {list<ConsumerRecord> lists = consumer.poll(timeout)lists.forearch {uuid = lists.get(index).getUUID; // 消息体uidtopicName = lists.get(index).topicName; // topic名字partition = lists.get(index).partition; // 分区redisKey = topicName:uuid; // 组合KeyRedissonList list = Redisson.getList(redisKey);// 非重复if(!RedissonList.contain(uuid)) {process();list.add(uuid);list.trim(0, max.poll.records); // 始终保留最近的条数}// 重复则直接丢弃}
}

框架原理

原始伪代码
发现没有，如果自己使用KafkaClientApi处理Msg会面临以下问题

1. consumerThread消费Topic的Partition的并发度需要自行控制。
2. 如果Thead被Interrupt异常，需要自行兜底重启。
3. 每个Consumer初始化都有大量的样板代码。
4. 如果需要监听某些Client的Event，则需要自行注册进去。

Spring Kafka框架正是用来解决以下样板代码()过多的痛点

new Thread() {while(true) {kafkaClint.init(proprity); // 初始化Server信息List<Recod> msgs = kafkaClient.poll(timeOut); // Poll消息precess(msgs); // 处理消息kafkaClient.commit(); // 提交当前偏移量}
}

使用要点:

Spring Redis

用于简化Redis Client的使用，避免大量样板代码。

框架原理:

使用要点:

Spring

Spring mvc

用于简化Tomcat的使用，避免大量样板代码。
仅考虑前后端分离的场景，jsp等过时的技术就不作讨论了。
mvc

模型层又称为数据bean层，视图层使用模型bean数据+模板引擎渲染出视图如jsp页面，控制层就是rest请求路径。在当前的前后端分离架构下，并不需要jsp，mustache这种反人类的mv，所以一般不会配置mv对应的hander，因此当前mvc仅仅使用了control简化Tomcat的使用。

框架原理
自动装配embed tomcat，然后通过contex拉起。