kafka重复数据问题排查记录

旧城等待， 2022-05-22 05:58 132阅读 0赞

## 问题 ##

向kafka写数据，然后读kafka数据，生产的数据量和消费的数据量对不上。

开始怀疑人生，以前奠定的基础受到挑战。。。

原来的测试为什么没有覆盖生产量和消费量的对比？

消费者写的有问题？反复检，无果。。。

再反过来梳理生产者的代码，检查了几遍，也并无发现疏漏。

开始怀疑Kafka  
以前看过文章，说kafka数据有可能重复和丢失数据，难道kafka的这么不精确，丢了这么多数据？

## 原因 ##

最后排查到生产者，kafka生产者使用confluent-kafka-go。  
向kafka生产数据的伪代码如下：

type Message struct { Content []byte }
    
    message = &Message{}
    
    for { content = getFromQ() message.Content =content Send(message) }

这段代码是有问题的。

Send()操作不是直接发送出去，而是放入一个队列，这样，如果队列数据发送前，下一次循环再次调用

message.Content =content

会将原来的content覆盖掉，就会出现写入kafka的数据存在重复。并且如果前一次数据长度大于下一次数据长度，后者覆盖前者，消费量就会小于生产量。

## 解决办法 ##

需要将代码修改为：

type Message struct {
    Content  []byte
    
    }
    
    for {
        content = getFromQ()
    
        message = &Message{}
        message.Content =content
    
        Send(message)
    
    }

每次发送消息时，重新定义一个变量保存数据，这样就不会出现覆盖问题。

关于confluent-kafka-go如何生产数据可参考  
[https://blog.csdn.net/lanyang123456/article/details/78377152][https_blog.csdn.net_lanyang123456_article_details_78377152]

## 参考 ##

kafka丢数据和数据重复  
[https://blog.csdn.net/xinlangtianxia/article/details/51387689][https_blog.csdn.net_xinlangtianxia_article_details_51387689]

kafka重复消费解决方案  
[https://blog.csdn.net/qq\_16583617/article/details/79648221][https_blog.csdn.net_qq_16583617_article_details_79648221]

[https_blog.csdn.net_lanyang123456_article_details_78377152]: https://blog.csdn.net/lanyang123456/article/details/78377152
[https_blog.csdn.net_xinlangtianxia_article_details_51387689]: https://blog.csdn.net/xinlangtianxia/article/details/51387689
[https_blog.csdn.net_qq_16583617_article_details_79648221]: https://blog.csdn.net/qq_16583617/article/details/79648221