通过消费组消费日志

方案概览

一个 Logstore 中包含多个 Shard，通过消费组消费数据就是将 Shard 分配给一个消费组中的消费者，分配方式遵循以下原则。

• 在一个消费组中，一个 Shard 只会分配到一个消费者。

• 在一个消费组中，一个消费者可以被分配多个 Shard。

新的消费者加入消费组后，这个消费组下面的 Shard 从属关系会调整，以实现消费的负载均衡，但是仍遵循上述分配原则。

基本概念

前提条件 ​

• 已创建 Project、标准型 Logstore 并完成日志采集配置。具体操作， 请参见 创建项目 Project 、 创建 Logstore 和 数据采集概述 。

• 已 配置访问凭证 。

• 已 初始化日志服务 Java SDK 。

步骤一：创建消费组

下面分别介绍通过 SDK、API 和 CLI 方式创建消费组。

import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.common.ConsumerGroup;
import com.aliyun.openservices.log.exception.LogException;
public class CreateConsumerGroup {
    public static void main(String[] args) throws LogException {
         // 本示例从环境变量中获取AccessKey ID和AccessKey Secret。
        String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        // 输入Project名称。
        String projectName = "ali-test-project";
        // 输入Logstore名称。
        String logstoreName = "ali-test-logstore";
        // 设置日志服务的服务接入点。此处以杭州为例，其它地域请根据实际情况填写。
        String host = "https://cn-hangzhou.log.aliyuncs.com";
        // 创建日志服务Client。
        Client client = new Client(host, accessId, accessKey);
        try {
            // 设置消费组名称。
            String consumerGroupName = "ali-test-consumergroup2";
            System.out.println("ready to create consumergroup");
            ConsumerGroup consumerGroup = new ConsumerGroup(consumerGroupName, 300, true);
            client.CreateConsumerGroup(projectName, logstoreName, consumerGroup);
            System.out.println(String.format("create consumergroup %s success", consumerGroupName));
        } catch (LogException e) {
            System.out.println("LogException e :" + e.toString());
            System.out.println("error code :" + e.GetErrorCode());
            System.out.println("error message :" + e.GetErrorMessage());
            throw e;
}

消费原理

消费组 SDK 的消费者在首次启动时，当消费组不存在时会创建消费组。起始消费位点是指创建消费组时的数据起始消费位点，该消费位点仅在第一次创建时有效。后续重启消费者时，消费者会从上次服务端保存的消费位点处继续消费。以本示例为例：

• LogHubConfig.ConsumePosition.BEGIN_CURSOR ：消费组从头开始消费日志，起始消费位点为 Logstore 中的第一条日志。

• LogHubConfig.ConsumePosition.END_CURSOR ：此消费位点记录 Logstore 日志的最后一条日志之后。

消费示例

步骤三：查看消费组状态

下面介绍两种查看消费组状态方式：

相关操作

• RAM 用户授权

  使用 RAM 用户操作时，需授予 RAM 用户操作消费组的相关权限。具体操作，请参见 创建 RAM 用户及授权 。

  授权的动作（Action）如下表所示。

  动作（Action）	说明	授权策略中的资源描述方式（Resource）
  log:GetCursorOrData（ GetCursor - 通过时间查询 Cursor ）	根据时间获取游标（cursor）。	acs:log: ${regionName} : ${projectOwnerAliUid} :project/ ${projectName} /logstore/ ${logstoreName}
  log:CreateConsumerGroup（ CreateConsumerGroup - 创建消费组 ）	在指定的 Logstore 上创建一个消费组。	acs:log: ${regionName} : ${projectOwnerAliUid} :project/ ${projectName} /logstore/ ${logstoreName} /consumergroup/ ${consumerGroupName}
  log:ListConsumerGroup（ ListConsumerGroup - 查询消费组 ）	查询指定 Logstore 的所有消费组。	acs:log: ${regionName} : ${projectOwnerAliUid} :project/ ${projectName} /logstore/ ${logstoreName} /consumergroup/*
  log:ConsumerGroupUpdateCheckPoint（ ConsumerGroupUpdateCheckPoint - 更新消费进度 ）	更新指定消费组的某个 Shard 的 Checkpoint。	acs:log: ${regionName} : ${projectOwnerAliUid} :project/ ${projectName} /logstore/ ${logstoreName} /consumergroup/ ${consumerGroupName}
  log:ConsumerGroupHeartBeat（ ConsumerGroupHeartBeat - 消费者发送心跳到服务端 ）	为指定消费者发送心跳到服务端。	acs:log: ${regionName} : ${projectOwnerAliUid} :project/ ${projectName} /logstore/ ${logstoreName} /consumergroup/ ${consumerGroupName}
  log:UpdateConsumerGroup（ UpdateConsumerGroup - 更新消费者组 ）	修改指定消费组属性。	acs:log: ${regionName} : ${projectOwnerAliUid} :project/ ${projectName} /logstore/ ${logstoreName} /consumergroup/ ${consumerGroupName}
  log:GetConsumerGroupCheckPoint（ GetCheckPoint - 获取指定消费组的消费点 ）	获取指定消费组消费的某个或者所有 Shard 的 Checkpoint。	acs:log: ${regionName} : ${projectOwnerAliUid} :project/ ${projectName} /logstore/ ${logstoreName} /consumergroup/ ${consumerGroupName}

  例如，消费组的相关资源信息如下所示，您要通过 RAM 用户操作该消费组，则需为 RAM 用户授予以下权限。

  • Project 所属的阿里云账号：174649****602745。

  • Project 所在地域 ID：cn-hangzhou。

  • Project 名称：project-test。

  • Logstore 名称：logstore-test。

  • 消费组名称：consumergroup-test。

  授权代码如下所示：

  {
    "Version": "1",
    "Statement": [
        "Effect": "Allow",
        "Action": [
          "log:GetCursorOrData"
        "Resource": "acs:log:cn-hangzhou:174649****602745:project/project-test/logstore/logstore-test"
        "Effect": "Allow",
        "Action": [
          "log:CreateConsumerGroup",
          "log:ListConsumerGroup"
        "Resource": "acs:log:cn-hangzhou:174649****602745:project/project-test/logstore/logstore-test/consumergroup/*"
        "Effect": "Allow",
        "Action": [
          "log:ConsumerGroupUpdateCheckPoint",
          "log:ConsumerGroupHeartBeat",
          "log:UpdateConsumerGroup",
          "log:GetConsumerGroupCheckPoint"
        "Resource": "acs:log:cn-hangzhou:174649****602745:project/project-test/logstore/logstore-test/consumergroup/consumergroup-test"
  }

• 异常诊断

  建议您为消费者程序配置 Log4j，将消费组内部遇到的异常信息打印出来，便于定位。log4j.properties 典型配置：

  log4j.rootLogger = info,stdout
  log4j.appender.stdout = org.apache.log4j.ConsoleAppender
  log4j.appender.stdout.Target = System.out
  log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
  log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n

  配置 Log4j 后，执行消费者程序可以看到类似如下异常信息：

  [WARN ] 2018-03-14 12:01:52,747 method:com.aliyun.openservices.loghub.client.LogHubConsumer.sampleLogError(LogHubConsumer.java:159)
  com.aliyun.openservices.log.exception.LogException: Invalid loggroup count, (0,1000]

• 通过消费组消费从某个时间开始的数据

  // consumerStartTimeInSeconds表示消费这个时间点之后的数据。
  public LogHubConfig(String consumerGroupName, 
                        String consumerName, 
                        String loghubEndPoint,
                        String project, String logStore,
                        String accessId, String accessKey,
                        int consumerStartTimeInSeconds);
  // position是个枚举变量，LogHubConfig.ConsumePosition.BEGIN_CURSOR表示从最老的数据开始消费，LogHubConfig.ConsumePosition.END_CURSOR表示从最新的数据开始消费。
  public LogHubConfig(String consumerGroupName, 
                        String consumerName, 
                        String loghubEndPoint,
                        String project, String logStore,
                        String accessId, String accessKey,
                        ConsumePosition position);

  说明

  • 按照消费需求，请您使用不同的构造方法。

  • 当服务端已保存 Checkpoint，则开始消费位置以服务端保存的 Checkpoint 为准。

  • 日志服务消费数据时，默认优先使用 Checkpoint 作为消费点。当您指定从固定时间点开始消费数据时，必须保证 consumerStartTimeInSeconds 时间点落到 TTL 周期内，否则会造成消费不生效。

• 重置 Checkpoint

  public static void updateCheckpoint() throws Exception {
          Client client = new Client(host, accessId, accessKey);
          // 这里 timestamp 需要是以秒为单位的 unix timestamp，如果您的时间戳以毫秒为单位，需要如下所示除以1000
          long timestamp = Timestamp.valueOf("2017-11-15 00:00:00").getTime() / 1000;
          ListShardResponse response = client.ListShard(new ListShardRequest(project, logStore));
          for (Shard shard : response.GetShards()) {
              int shardId = shard.GetShardId();
              String cursor = client.GetCursor(project, logStore, shardId, timestamp).GetCursor();
              client.UpdateCheckPoint(project, logStore, consumerGroup, shardId, cursor);
      }

相关文档

• API

  操作	API 接口
  创建消费组	CreateConsumerGroup - 创建消费组
  查询消费组	ListConsumerGroup - 查询消费组
  删除消费组	DeleteConsumerGroup - 删除消费组
  更新消费组	UpdateConsumerGroup - 更新消费者组
  发送消费者心跳	ConsumerGroupHeartBeat - 消费者发送心跳到服务端
  查询消费组的 Checkpoint	GetCheckPoint - 获取指定消费组的消费点
  更新消费组的 Checkpoint	ConsumerGroupUpdateCheckPoint - 更新消费进度

• SDK

  语言	文档链接
  Java	使用 Java SDK 管理消费组 通过消费组消费数据
  Python	使用 Python SDK 管理消费组 通过消费组消费数据

• CLI

  操作	命令行接口
  创建消费组	create_consumer_group
  查询消费组	list_consumer_group
  更新消费组	update_consumer_group
  删除消费组	delete_consumer_group
  查询消费组的 Checkpoint	get_check_point
  更新消费组的 Checkpoint	update_check_point