Kafka 消费消息回滚以及linux查询GroupId消费情况

秒速五厘米 2022-05-21 10:46 503阅读 0赞

**用相同的GroupId消费已经消费过的信息，可以进行回滚。**

import time
    import redis
    import json
    import math
    import Queue
    import os
    
    def cont(t):
        time_local = time.localtime(t)
        dt = time.strftime("%Y-%m-%d %H:%M:%S",time_local)
        return dt
    
    
    def seek2ts(ts,topic,consumer):
    
        x= [0,0,0,0,0,0 ]
        ts*=1000
        m=[]
        for y in range(6):
           i=TopicPartition(topic, y)
           x[y] =  consumer.offsets_for_times({i:ts})[i][0]
           m.append(i)
        consumer.assign(m)
        for y in range(6):
                i=TopicPartition(topic, y)
                consumer.seek(i,x[y])
        return x
    
    
    def main():
            consumer = KafkaConsumer(bootstrap_servers=[#集群IP] , group_id ="")
            seek2ts(1530748800#回滚到的时间戳,"#TOPIC",consumer)
            consumer.poll()
            consumer.commit()
    
            print "done"
    
    main()

在linux查询kafka中GroupId的消费情况：

/root/kafka\_2.11-0.11.0.2/bin/kafka-run-class.sh kafka.admin.ConsumerGroupCommand --bootstrap-server k3:9092  --describe --group online\_dist2ali

k3:9092   是kafka地址

online\_dist2ali  是GroupId

发表评论取消回复

表情：

评论列表（有 0 条评论，503人围观）

还没有评论，来说两句吧...

相关阅读

相关 Kafka：消费者消费消息流程

消费者组初始化过程 ![4aee20e5335044cbaff1dd63e17c4b6e.png][] 步骤： 1.每一个Broker都有coordinator（

╰+哭是因爲堅強的太久メ/ 2024年03月24日 17:06/ 0 赞/ 12 阅读

相关 java生产消费kafka消息

maven依赖 <dependency> <groupId>org.apache.kafka</groupId> <artifactI

我会带着你远行/ 2023年09月27日 09:08/ 0 赞/ 85 阅读

相关 Kafka设定时段消费消息

Consumer类 @Component @Slf4j public class KafkaConsumer { @Auto

逃离我推掉我的手/ 2022年11月29日 05:45/ 0 赞/ 412 阅读

相关【Kafka】——顺序消费、重复消费、消息丢失

前言 Kafka可以说是为分布式而生的一个消息中间件，功能很强大，提到这个，我们可能就会想到消息中间件常提到的几个问题，消费的顺序性、重复消费、消息丢失等问题，接下来我们

逃离我推掉我的手/ 2022年10月05日 11:00/ 1 赞/ 391 阅读

相关 kafka如何消费消息

转载自：http://generalcode.cn/archives/255 消费者与消费组假设这么个场景：我们从Kafka中读取消息，并且进行检查，最后产生结果

分手后的思念是犯贱/ 2022年05月24日 10:14/ 0 赞/ 225 阅读

相关 kafka_Kafka的消息消费机制

测试：启动生产者与消费者。使用zooinspector工具查看当前zook情况：找到当前的消费者：（我也是一个一个找的） ![70][] 可以看到当前的off

小灰灰/ 2022年05月24日 03:29/ 0 赞/ 223 阅读

相关 Kafka 消费消息回滚以及linux查询GroupId消费情况

用相同的GroupId消费已经消费过的信息，可以进行回滚。 import time import redis import json impo

秒速五厘米/ 2022年05月21日 10:46/ 0 赞/ 504 阅读

相关 spark消费kafka消息

教程： [http://spark.apache.org/docs/2.2.0/streaming-kafka-0-10-integration.html][http_spa

待我称王封你为后i/ 2022年05月14日 05:45/ 0 赞/ 253 阅读

相关 Flink消费kafka消息实战

![640?wx\_fmt=png][640_wx_fmt_png] 本次实战的内容是开发Flink应用，消费来自kafka的消息，进行实时计算；环境情况本

柔光的暖阳◎/ 2022年04月24日 05:18/ 0 赞/ 317 阅读

相关 springboot消费kafka设置topics 以及 groupId

关于kafka的这些概念和理论请见这篇博客 [https://www.jianshu.com/p/d3e963ff8b70][https_www.jianshu.com_p_d

叁歲伎倆/ 2021年11月11日 12:16/ 0 赞/ 1041 阅读