近期遇到问题与总结

项目概况：

云边协同混合部署

上行 边缘-》中心 连接方式 http restful

下行 中心-》边缘 连接方式 websocket长连接

中心包含中心同步服务、管理控制台服务

边缘包括边缘协议解析，边缘引擎、边缘同步服务

问题

边缘端通常部署在园区或私有服务器内，通过网络与部署在公有云的中心端相连接。当边缘端与中心端因为网络等因素断开连接后，会产生数据丢失，这对某些场景是难以接受的。因此同步特性需要解决数据如何续传的问题。在实现这一特性的期间，遇到一些问题，例如 如何在同步失败后，再次消费边缘传递来的消息数据。

解决思路过程：

第一阶段：

上游消息通过kafka传递，原先通过自动提交方式实现消费数据，因此考虑修改成通过手动提交方式实现。通过修改 enable.auto.commit为false，并在调用http返回体中解析状态码，若请求成功则commit当前consumer的三元组（topic，partition，offset）。

自测方式：1、保证上行通道畅通，在边缘端监听的kafka中produce一条正在监听的topic的消息，中心端能获取到对应日志打印，并且在中心端转发到的kafka中能consume到对应的消息。2、kill掉中心同步服务（若多实例需将kill干净）3、再向边缘端监听的kafka中发送第1步发送的消息（可以修改一些特殊关键字方便后续搜索）4、重启第2步kill掉的中心同步服务 5、查看正在consume中心端转发到的kafka有没有消费到第3步发送的消息。结果是没有收到。

原因：正在起着的消费者不会再次消费没有commit的消息，只有触发rebalance 时（重启consumer时会触发），会从offset开始消费。

第二阶段：

在上述实现的基础上加入kafka的seek方法，seek方法可以通过（topic，partition，offset）三元组，调整consumer的本地缓存中的offset，在执行过该方法后，在下一次poll消息时，会从对应offset进行消费。因此当 请求成功则commit当前consumer的三元组（topic，partition，offset），失败时则调用seek方法，重置offset。自测后发现可以解决问题。但在转测试后，在多topic以及同一topic有多条消息时，出现丢失消息问题。仍在定位中。

并且由于边缘与中心端服务的总体架构是通过vertx实现，因此kafka的api都是vertx的api，分析丢失数据可能的几种情况：1、vertx kafka api实现的与原生kafka api不太一致，是封装了原生kafka的实现，因此可能使用方式有误，需要阅读vertx kafka的源码 2、vertx框架间调用都是采用异步调用，可能产生同一个consumer被多线程seek 或commit的问题

考虑到上述因素，考虑替换为原生kafka api 进行消费消息，commit和seek也使用对应的api，将同步这块逻辑从边缘同步服务中拆分出来，作为一个独立的服务（之前边缘引擎和边缘同步服务在一个工程里）。在拆分的过程中，遇到的问题，由于之前http调用采用的也是vertx的httpclient api，拆分后使用原生http client访问中心同步服务，在kill掉中心同步服务进程后，再启动，调用http接口返回的response仍一直是null。由于上行访问http的方式都是通过nginx代理进行https的访问，因此尝试将访问地址url修改为对应服务的ip 并采用http访问，没有上述问题。目前考虑是采用的原生httpclient问题，因此准备换一个okhttp api尝试一下。

标签：总结,同步,http,遇到,近期,kafka,边缘,api,offset
来源： https://www.cnblogs.com/cosmocosmo/p/15967126.html