日志服务
日志服务
日志服务
文档指南
请输入
用户指南
TLS Agent
资源管理
数据采集
LogCollector 采集
LogCollector 简介
安装与升级
采集容器日志
LogCollector 插件
管理 LogCollector
云产品日志采集
TOS 日志采集
CDN 日志采集
DBW 日志采集
VPN 日志采集
VKE 日志采集
CLB 日志采集
CloudTrail 日志采集
CFW 日志采集
NTA 日志采集
ALB 日志采集
APMPlus 日志采集
veFaaS 日志采集
VPC 日志采集
veImageX 日志采集
TrafficRoute DNS 日志采集
DDos 日志采集
WAF 日志采集
DCDN 日志采集
多云安全平台日志采集
APIG 日志采集
机器学习任务日志采集
采集火山方舟应用实验室日志
CR 日志采集
FileNAS 日志采集
大模型应用防火墙日志采集
vePFS 日志采集
云安全中心日志采集
EFS 日志采集
检索分析
使用第三方工具检索分析日志
关联外部数据源检索分析日志
SQL 函数
可视化
日志应用
云产品观测日志应用
大模型观测日志应用
Coding Agent 日志应用
ModelProvider 日志应用
No-coding Agent 日志应用
Framework 日志应用
基础服务监控日志应用
数据处理
数据加工
消费与投递
告警
Trace 存储
API 参考
日志管理
机器组管理
采集配置管理
数据处理器
定时 SQL 分析
告警管理
消费组管理
Trace 存储
SDK 参考
Go SDK
Java SDK
Android SDK
Node.js SDK
CLI 参考
最佳实践
消费与投递
可视化
常见问题
数据采集常见问题
索引配置常见问题
- 文档首页
日志服务SDK 参考Python SDK通过 Python SDK 消费组消费日志
复制全文
下载 pdf通过 Python SDK 消费组消费日志
文档反馈
问问助手日志服务通过 SDK 提供了消费组（ConsumerGroup）功能，支持通过消费组消费日志数据。本文档介绍如何使用 Python SDK 消费组消费日志。
前提条件
- 已创建并获取火山引擎密钥 AccessKey。
- 火山引擎账号的访问密钥 AccessKey 拥有所有 API 的全部权限。建议您通过 IAM 用户进行 API 相关操作和日常运维。使用 IAM 用户前，主账号应为 IAM 用户授予消费组相关的权限。授权示例请参考IAM 概述。
- 已安装日志服务 SDK。安装步骤请参考安装 Python SDK。
配置说明
日志服务通过 SDK 提供了消费组（ConsumerGroup）功能，支持通过消费组消费日志数据，通过消费组消费时，日志服务会自动均衡各个消费者的消费能力与进度，自动分配 Shard，您无需关注消费组的内部调度细节及消费者之间的负载均衡、故障转移等，只需要专注于业务逻辑。
说明
日志服务 SDK 消费组实现了请求失败自动重试、消费进度检查点自动上报等机制。因此，您仅需要关注于如何处理每次消费得到的 LogGroupList 的业务逻辑实现即可。
Python SDK 中，ConsumerConfig 类的构造函数返回了 Python SDK 消费组配置，其中应配置 endpoint、region、accessKeyID、accessKeySecret 等基本信息、日志项目 ID 和日志主题 ID 列表、消费组名称和消费者名称。
除此之外，您还可通过 ConsumerConfig 的其他字段进行额外的自定义配置。ConsumerConfig 支持的参数如下：
参数 | 类型 | 示例值 | 描述 |
max_fetch_log_group_count | int | 100 | 消费者单次消费日志时，最大获取 LogGroup 数量，默认为 100，最大为 1000。 |
heartbeat_interval_in_second | int | 20 | Consumer 心跳上报时间间隔，单位为秒。 |
data_fetch_interval_in_millisecond | int | 200 | Consumer 消费日志时间间隔，单位为毫秒。 |
flush_checkpoint_interval_in_second | int | 5 | Consumer 上传消费进度的时间间隔，单位为秒。 |
consume_from | str | begin | 开始消费时的默认消费位点，与 DescribeCursor 的 From 参数一致。仅在该消费者从未上传过消费位点时有效。 |
ordered_consume | bool | False | 是否开启顺序消费。开启顺序消费后，消费者会根据 Shard 分裂的父子关系进行消费。 例如 Shard0 分裂为 Shard1 与 Shard2，而 Shard1 又分裂为 Shard3 与 Shard4。在开启顺序消费之后，会根据 (Shard0) -> (Shard1, Shard2) -> (Shard2, Shard3, Shard4) 的顺序进行消费。 |
示例代码
以下代码以 Python SDK 为例，演示通过 SDK 创建消费组和消费者，并消费日志的整体流程。
# coding=utf-8
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import os
import time
from volcengine.tls.TLSService import TLSService
from volcengine.tls.consumer.consumer import TLSConsumer, LogProcessor
from volcengine.tls.consumer.consumer_model import ConsumerConfig
from volcengine.tls.log_pb2 import LogGroupList
# 您需要实现一个继承LogProcessor的类，并按照业务需要自行实现process函数，用于处理消费到的每个LogGroupList
# 下面展示了逐个打印每条日志的基本代码实现
class MyLogProcessor(LogProcessor):
def process(self, topic_id: str, shard_id: int, log_group_list: LogGroupList):
print(topic_id + " --- " + str(shard_id))
count = 0
for log_group in log_group_list.log_groups:
for log in log_group.logs:
count += 1
print("*** Count = {} ***".format(count))
for content in log.contents:
print("{}: {}".format(content.key, content.value))
print()
if __name__ == "__main__":
# 初始化客户端，推荐通过环境变量动态获取火山引擎密钥等身份认证信息，以免 AccessKey 硬编码引发数据安全风险。详细说明请参考https://www.volcengine.com/docs/6470/1166455
# 使用 STS 时，ak 和 sk 均使用临时密钥，且设置 VOLCENGINE_TOKEN；不使用 STS 时，VOLCENGINE_TOKEN 部分传空
endpoint = os.environ["VOLCENGINE_ENDPOINT"]
region = os.environ["VOLCENGINE_REGION"]
access_key_id = os.environ["VOLCENGINE_ACCESS_KEY_ID"]
access_key_secret = os.environ["VOLCENGINE_ACCESS_KEY_SECRET"]
# 实例化TLS客户端
tls_service = TLSService(endpoint, access_key_id, access_key_secret, region)
# 配置消费组的必填参数，ConsumerConfig构造函数设定了一些默认参数，也可根据需要自定义配置
consumer_config = ConsumerConfig(project_id="your-project-id",
consumer_group_name="python-consumer-group",
consumer_name="python-consumer",
topic_id_list=["your-topic-id"])
tls_consumer = TLSConsumer(consumer_config, tls_service, MyLogProcessor())
# 调用start方法开始持续消费
tls_consumer.start()
# 可通过调用tls_consumer.stop()来结束消费组消费
time.sleep(10)
tls_consumer.stop()
最近更新时间：2025.09.26 19:05:49
这个页面对您有帮助吗？
有用
有用
无用
无用