通过 Go SDK 消费组消费日志

日志服务通过 SDK 提供了消费组（ConsumerGroup）功能，支持通过消费组消费日志数据。本文档介绍如何使用 Go SDK 消费组消费日志。

前提条件

• 已创建并获取火山引擎密钥 AccessKey。
• 火山引擎账号的访问密钥 AccessKey 拥有所有 API 的全部权限。建议您通过 IAM 用户进行 API 相关操作和日常运维。使用 IAM 用户前，主账号应为 IAM 用户授予消费组相关的权限。授权示例请参考基于 IAM 管理权限。
• 已安装日志服务 SDK。安装步骤请参考安装 Go SDK。

配置说明

日志服务通过 SDK 提供了消费组（ConsumerGroup）功能，支持通过消费组消费日志数据，通过消费组消费时，日志服务会自动均衡各个消费者的消费能力与进度，自动分配 Shard，您无需关注消费组的内部调度细节及消费者之间的负载均衡、故障转移等，只需要专注于业务逻辑。
关于消费组消费日志数据的基本概念等背景信息，请参考通过消费组消费数据。

Go SDK 中，log_consumer.GetDefaultConsumerConfig() 函数返回消费组的默认配置 consumerCfg，并支持配置 Endpoint、Region、AccessKeyID、AccessKeySecret 等基本信息、日志项目 ID 和日志主题 ID 列表、消费组名称和消费者名称。除此之外，您还可以通过 consumerCfg 的其他字段进行其他自定义配置。
consumerCfg 支持的字段如下所示。

LoggerConfig 配置说明：

示例代码

以下代码以 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())
    }
}