标签：dict1 value kafka topic 偏移量 key print message 上次

#!/usr/bin/env python
# coding=utf-8
from kafka import *
from  kafka import KafkaConsumer
import datetime,time
import json
def get_kafka_reviews(bootstrap_servers,topics):
    # print type(self.bootstrap_servers)
    consumer = KafkaConsumer(bootstrap_servers=[bootstrap_servers], group_id='yyjk01', auto_offset_reset='latest', enable_auto_commit=True)
    consumer.subscribe(topics=(topics))  #订阅要消费的主题
    # print consumer.topics()
    # print "+++++++",consumer.position(TopicPartition(topic=u'ctripapi_duplicateddata_review', partition=1)) #获取当前主题的最新偏移量

    review_list =[]
    for message in consumer:
        print message
        # str_time = datetime.datetime.fromtimestamp(message.timestamp / 1000)
        # print message.timestamp
        # print type(message.timestamp)
        # #print message.topic ,message.timestamp,message.value
        #
        #
        # print message.topic, str_time, message.value
        # print type(message.value)
        # dict1 = json.loads(message.value)
        # print dict1
        # print type(dict1)
        # print '-------------------'
        # for key in dict1:
        #
        #     print str_time,key,dict1[key]
        # print '-------------------'
        # #print '====%s:%d:%d:key-%s value=%s=='%(message.topic,message.partition,message.offset,message.key,message.value)
        # #review_list.append(message.value)
    #return  review_list

print get_kafka_reviews('192.168.137.2:9092','test_topic')

标签：dict1,value,kafka,topic,偏移量,key,print,message,上次
来源： https://blog.csdn.net/zhaoyangjian724/article/details/120565522

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。