自定义消费

# -*- coding: utf-8 -*-

import os
from threading import RLock
from tencentcloud.log.consumer import *
from tencentcloud.log.logclient import YunApiLogClient

root = logging.getLogger()

handler = logging.StreamHandler()

handler.setFormatter(logging.Formatter(

    fmt='[%(asctime)s] - [%(threadName)s] - {%(module)s:%(funcName)s:%(lineno)d} %(levelname)s - %(message)s',

    datefmt='%Y-%m-%d %H:%M:%S'))

root.setLevel(logging.INFO)

root.addHandler(handler)

logger = logging.getLogger(__name__)

class SampleConsumer(ConsumerProcessorBase):
    #记录上次消费offset的时间。
    last_check_time = 0
    #记录消费的日志数据
    log_results = []

    lock = RLock()

    def initialize(self, topic_id):

        self.topic_id = topic_id

    def process(self, log_groups, offset_tracker):
        #获取消费的日志数据
        for log_group in log_groups:

            for log in log_group.logs:

                item = dict()

                item['time'] = log.time

                item['filename'] = log_group.filename

                item['source'] = log_group.source

                for content in log.contents:

                    item[content.key] = content.value

                with SampleConsumer.lock:

                    SampleConsumer.log_results.append(item)
                    
        #提交offset
        current_time = time.time()

        if current_time - self.last_check_time > 3:

            try:

                self.last_check_time = current_time

                offset_tracker.save_offset(True)

            except Exception:

                import traceback

                traceback.print_exc()

        else:

            try:

                offset_tracker.save_offset(False)

            except Exception:

                import traceback

                traceback.print_exc()

        return None

 def shutdown(self, offset_tracker):

        try:

            offset_tracker.save_offset(True)

        except Exception:

            import traceback

            traceback.print_exc()


def sleep_until(seconds, exit_condition=None, expect_error=False):

    if not exit_condition:

        time.sleep(seconds)

        return

    s = time.time()

    while time.time() - s < seconds:

        try:

            if exit_condition():

                break

        except Exception:

            if expect_error:

                continue

        time.sleep(1)
# 创建消费组
def sample_consumer_group():

    # load options from envs
    #请求域名
    endpoint = os.environ.get('TENCENTCLOUD_LOG_SAMPLE_ENDPOINT', '')
    #用户的Secret_id
    access_key_id = os.environ.get('TENCENTCLOUD_LOG_SAMPLE_ACCESSID', '')
    #用户的Secret_key
    access_key = os.environ.get('TENCENTCLOUD_LOG_SAMPLE_ACCESSKEY', '')
    #日志集id
    logset_id = os.environ.get('TENCENTCLOUD_LOG_SAMPLE_LOGSET_ID', '')
    #日志主题id列表，支持多个
    topic_ids = ['topic_id_1','topic_id_2']
    #消费组，同一个日志集下的消费组名称唯一
    consumer_group = 'consumer-group-1'
    #消费者1
    consumer_name1 = "consumer-group-1-A"
    #消费者2
    consumer_name2 = "consumer-group-1-B"
    #地域
    region = "ap-guangzhou"

    assert endpoint and access_key_id and access_key and logset_id, ValueError("endpoint/access_id/access_key and logset_id cannot be empty")
    #创建访问云API接口的Client
    client = YunApiLogClient(access_key_id, access_key, region=region)
    #初始化消费结果列表
    SampleConsumer.log_results = []

    try:

        # 创建两个消费者配置
      
        option1 = LogHubConfig(endpoint, access_key_id, access_key, region, logset_id, topic_ids, consumer_group,

                               consumer_name1, heartbeat_interval=3, data_fetch_interval=1,

                               offset_start_time="end", max_fetch_log_group_size=10485760)

        option2 = LogHubConfig(endpoint, access_key_id, access_key, region, logset_id, topic_ids, consumer_group,

                               consumer_name2, heartbeat_interval=3, data_fetch_interval=1,

                               offset_start_time="end", max_fetch_log_group_size=10485760)

        print("*** start to consume data...")
        #消费者1
        client_worker1 = ConsumerWorker(SampleConsumer, consumer_option=option1)
        #启动消费者1
        client_worker1.start()

        client_worker2 = ConsumerWorker(SampleConsumer, consumer_option=option2)

        client_worker2.start()
        
        sleep_until(120, lambda: len(SampleConsumer.log_results) > 0)

        print("*** consumer group status ***")
        #打印消费组信息：消费组的名称、消费的日志主题、消费者心跳超时时间
        ret = client.list_consumer_group(logset_id, topic_ids)
        
        ret.log_print()

        print("*** stopping workers")
        #关闭消费者
        client_worker1.shutdown()

        client_worker2.shutdown()

        print("*** delete consumer group")
        #删除消费者组
        client.delete_consumer_group(logset_id, consumer_group)

    except Exception as e:

        raise e
        
    #打印消费的日志数据
    ret = str(SampleConsumer.log_results)

    print("*** get content:")

    print(ret)


if __name__ == '__main__':

    sample_consumer_group()