日志服务
日志服务
日志服务
文档指南
请输入
用户指南
TLS Agent
资源管理
数据采集
LogCollector 采集
LogCollector 简介
安装与升级（Linux）
安装与升级（Windows）
采集容器日志
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 参考Go SDK通过 Go SDK 消费组消费日志
复制全文
下载 pdf通过 Go SDK 消费组消费日志
文档反馈
问问助手日志服务通过 SDK 提供了消费组（ConsumerGroup）功能，支持通过消费组消费日志数据。本文档介绍如何使用 Go SDK 消费组消费日志。
前提条件
- 已创建并获取火山引擎密钥 AccessKey。
- 火山引擎账号的访问密钥 AccessKey 拥有所有 API 的全部权限。建议您通过 IAM 用户进行 API 相关操作和日常运维。使用 IAM 用户前，主账号应为 IAM 用户授予消费组相关的权限。授权示例请参考IAM 概述。
- 已安装日志服务 SDK。安装步骤请参考安装 Go SDK。
配置说明
日志服务通过 SDK 提供了消费组（ConsumerGroup）功能，支持通过消费组消费日志数据，通过消费组消费时，日志服务会自动均衡各个消费者的消费能力与进度，自动分配 Shard，您无需关注消费组的内部调度细节及消费者之间的负载均衡、故障转移等，只需要专注于业务逻辑。
说明
日志服务 SDK 消费组实现了请求失败自动重试、消费进度检查点自动上报等机制。因此，您仅需要关注于如何处理每次消费得到的 LogGroupList 的业务逻辑实现即可。
Go SDK 中，log_consumer.GetDefaultConsumerConfig() 函数返回消费组的默认配置 consumerCfg，并支持配置 Endpoint、Region、AccessKeyID、AccessKeySecret 等基本信息、日志项目 ID 和日志主题 ID 列表、消费组名称和消费者名称。除此之外，您还可以通过 consumerCfg 的其他字段进行其他自定义配置。
consumerCfg 支持的字段如下所示。
参数 | 类型 | 示例值 | 描述 |
参数 | 类型 | 示例值 | 描述 |
MaxFetchLogGroupCount | Integer | 100 | 消费者单次消费日志时，最大获取 LogGroup 数量，默认为 100，最大为 1000。 |
HeartbeatIntervalInSecond | Integer | 20 | Consumer 心跳上报时间间隔，单位为秒。 |
DataFetchIntervalInMillisecond | Integer | 200 | Consumer 消费日志时间间隔，单位为毫秒。 |
FlushCheckpointIntervalSecond | Integer | 5 | Consumer 上传消费进度的时间间隔，单位为秒。 |
ConsumeFrom | String | begin | 开始消费时的默认消费位点，与 DescribeCursor 的 From 参数一致。仅在该消费者从未上传过消费位点时有效。 |
OrderedConsume | Boolean | false | 是否开启顺序消费。开启顺序消费后，消费者会根据 Shard 分裂的父子关系进行消费。 例如 Shard0 分裂为 Shard1 与 Shard2，而 Shard1 又分裂为 Shard3 与 Shard4。在开启顺序消费之后，会根据 (Shard0) -> (Shard1, Shard2) -> (Shard2, Shard3, Shard4) 的顺序进行消费。 |
LoggerConfig | LoggerConfig | / | 日志相关配置，详细信息请参考下表。 |
LoggerConfig 配置说明：
参数 | 类型 | 示例值 | 描述 |
参数 | 类型 | 示例值 | 描述 |
LogLevel | String | info | 设置日志输出级别，默认值为 Info。consumer 中一共有 4 种日志输出级别，分别为 debug、info、warn 和 error。 |
LogFileName | String | /var/log/tls/consumer.log | 日志文件输出路径。若未设置，则默认输出到 stdout。 |
IsJsonType | Boolean | false | 是否格式化文件输出格式，默认为 false。 |
LogMaxSize | Integer | 10 | 单个日志文件的最大存储数量，默认为 10MiB。 |
LogCompress | Boolean | false | 日志是否开启压缩。 |
示例代码
以下代码以 Go SDK 为例，演示通过 SDK 创建消费组和消费者，并消费日志的整体流程。
package tls
import (
"context"
"fmt"
"os"
"time"
"github.com/pkg/errors"
log_consumer "github.com/volcengine/volc-sdk-golang/service/tls/consumer"
"github.com/volcengine/volc-sdk-golang/service/tls/pb"
)
func launchConsumer() error {
// 获取消费组的默认配置
consumerCfg := log_consumer.GetDefaultConsumerConfig()
// 推荐通过环境变量动态获取火山引擎密钥等身份认证信息，以免 AccessKey 硬编码引发数据安全风险。详细说明请参考https://www.volcengine.com/docs/6470/1166455
// 使用 STS 时，ak 和 sk 均使用临时密钥，且设置 VOLCENGINE_TOKEN；不使用 STS 时，VOLCENGINE_TOKEN 部分传空
consumerCfg.Endpoint = os.Getenv("VOLCENGINE_ENDPOINT")
consumerCfg.Region = os.Getenv("VOLCENGINE_REGION")
consumerCfg.AccessKeyID = os.Getenv("VOLCENGINE_ACCESS_KEY_ID")
consumerCfg.AccessKeySecret = os.Getenv("VOLCENGINE_ACCESS_KEY_SECRET")
// 请配置您的日志项目ID和日志主题ID列表
consumerCfg.ProjectID = "<YOUR-PROJECT-ID>"
consumerCfg.TopicIDList = []string{"<YOUR-TOPIC-ID>"}
// 请配置您的消费组名称（若您未创建过消费组，SDK将默认为您创建指定名称的消费组）
consumerCfg.ConsumerGroupName = "<CONSUMER-GROUP-NAME>"
// 请配置消费者名称（同一个消费组的不同消费者需要保证不同名）
consumerCfg.ConsumerName = "<CONSUMER_NAME>"
// 定义日志消费函数，您可根据业务需要，自行实现处理LogGroupList的日志消费函数
// 下面展示了逐个打印消费到的每条日志的每个键值对的代码实现示例
var handleLogs = func(topicID string, shardID int, l *pb.LogGroupList) {
fmt.Printf("received new logs from topic: %s, shard: %d\n", topicID, shardID)
for _, logGroup := range l.LogGroups {
for _, log := range logGroup.Logs {
for _, content := range log.Contents {
fmt.Printf("%s: %s\n", content.Key, content.Value)
}
}
}
}
// 创建消费者
consumer, err := log_consumer.NewConsumer(context.TODO(), consumerCfg, handleLogs)
if err != nil {
return errors.Wrap(err, "get new consumer failed: ")
}
// 启动消费者消费
if err := consumer.Start(); err != nil {
return errors.Wrap(err, "start consumer failed: ")
}
// 等待消费
<-time.After(time.Second * 60)
// 停止消费
consumer.Stop()
return nil
}
func main() {
if err := launchConsumer(); err != nil {
fmt.Println(err.Error())
}
}
最近更新时间：2026.10.19 17:52:08
这个页面对您有帮助吗？
有用
有用
无用
无用