Kafka重复消费、Dubbo重复调用问题排查-Toy模板网

这篇具有很好参考价值的文章主要介绍了Kafka重复消费、Dubbo重复调用问题排查。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

一、业务流程简述：

本业务为车机流量充值业务，大致流程为：收到微信、支付宝端用户支付成功回调后，将用户订单信息发送至kafka中；消费者接收到kafka中信息后进行解析，处理用户订单信息，为用户订购相关流量包（调用电信相关接口），订购成功/失败后会通过MQTT发送订购成功/失败消息至车机端，若订购失败则为用户退款。

二、线上问题

8/22 收到400反馈称用户于 8/21购买的流量包未到账

三、问题排查

首先排查用户订单记录及订单状态

org.springframework.kafka.kafkaexception: seek to current after exception; n,kafka,分布式,dubbo

可以看到用户在同一时间有两条处理失败的订单，查看代码后发现此订单记录有两处场景会创建：

1、调用电信套餐订购接口时会根据调用结果创建对应订单记录；

2、调用自身套餐订购服务失败后会创建对应订单记录。

看了下订单记录生成逻辑后，我有点蚌埠住了：在自身套餐订购服务内容内调用了电信套餐订购接口并根据订购结果存储对应订单记录后，还要在外层自身订购服务失败后存储一个失败订单，如果订单处理失败的话岂不是最少都有两条失败订单……

但是我仔细看了下发现这两处创建的订单参数有些许不同，查看数据后发现果然没有这么简单……（竟然有三条订单记录，由于第二种创建的失败订单会有部分参数缺失，页面上不会查询到这类订单）

org.springframework.kafka.kafkaexception: seek to current after exception; n,kafka,分布式,dubbo

其实此时根据订单记录和订单生成逻辑大致可以确定肯定套餐订购服务走了两次，我们查看下服务日志：

org.springframework.kafka.kafkaexception: seek to current after exception; n,kafka,分布式,dubbo

确实流量订购失败了，但是二者时间间隔竟然长达40s，而这个项目是用的dubbo服务，查看项目配置后发现设置的调用超时时间为30s，重试次数为2；显然已经超过这个超时时间导致了套餐订购服务重复调用；解决方法也很简单：注解中不允许重复调用即可

@Service(version = "${provider.service.version}", retries = 0)

到此，可以解决套餐重复订购问题。但是排查服务log时发现这条数据被消费了两次只是因为代码中处理了payLog的状态，而使得第二次消费时没有执行其他逻辑

kafka消费log

org.springframework.kafka.kafkaexception: seek to current after exception; n,kafka,分布式,dubbo

纳尼，本着对前同事代码极其相信的想法首先想到的是：是不是这款订单微信端回调时调用了两次导致往kafka中发了两条重复数据呢？Ok继续排查发送往kafka中发消息的相关日志：显然支付回调这块逻辑没有问题，只提交了一条消息。

kafka生产者log

org.springframework.kafka.kafkaexception: seek to current after exception; n,kafka,分布式,dubbo 现在可以确定问题为消费者重复消费，进一步查看订单处理逻辑发现流量订购失败时会有退款操作，但目前用户流量未到账且未退款，猜测应该是充值失败后服务某处出现异常导致消费者没有正确提交位移而引起重复消费；

org.springframework.kafka.kafkaexception: seek to current after exception; n,kafka,分布式,dubbo

查看项目错误日志发现以下信息：

2023-08-21 18:34:46,264 ERROR [org.springframework.kafka.KafkaListenerEndpointContainer#3-0-C-1] o.s.k.l.KafkaMessageListenerContainer [LogAccessor.java : 149] Error handler threw an exception
org.springframework.kafka.KafkaException: Seek to current after exception; nested exception is org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void com.coocaa.client.web.kafka.KafkaConsumer.orderNotify(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.String>) throws java.lang.Exception' threw exception; nested exception is java.net.SocketTimeoutException: connect timed out; nested exception is java.net.SocketTimeoutException: connect timed out
        at org.springframework.kafka.listener.SeekUtils.seekOrRecover(SeekUtils.java:208)
        at org.springframework.kafka.listener.DefaultErrorHandler.handleRemaining(DefaultErrorHandler.java:135)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeErrorHandler(KafkaMessageListenerContainer.java:2707)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:2588)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:2457)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:2335)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:2006)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1375)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1366)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1257)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.lang.Thread.run(Thread.java:834)

org.springframework.kafka.kafkaexception: seek to current after exception; n,kafka,分布式,dubbo

果不其然，项目中有超时信息具体在MQTT推送消息中，有两处推送服务是因为百度MQTT的新、旧两个版本（此时大致可以确定是之前新老版本同时推送，现在百度不再维护老版本的mqtt引起服务异常）

   mnoIoTPushService.pushSinglePayMessageAction(payLog.getVin(), findOneTimelyResult);
   mnoIoTPushServiceV2.pushSinglePayMessageAction(payLog.getVin(), findOneTimelyResult);

进一步查看推送服务代码，老版本是使用OkHttpClient（默认超时时间10_000）建立连接，log中是于2023-08-21 18:34:35,656 上报流量订购失败正常应立马向车机端推送MQTT消息，此处于2023-08-21 18:34:46,264上报连接超时，与OkHttp的10s超时时间基本一致。

org.springframework.kafka.kafkaexception: seek to current after exception; n,kafka,分布式,dubbo