详情页标题前

阿里云日志服务SLS使用Java SDK管理消费组-云淘科技

• 阿里云日志服务SLS
详情页1

通过消费组（ConsumerGroup）消费日志数据有显著优点，您无需关注日志服务的实现细节和消费者之间的负载均衡、Failover等，只需关注业务逻辑。本文通过代码示例介绍如何创建、修改、查询、删除消费组等。

前提条件

  • 已创建RAM用户并完成授权。具体操作，请参见创建RAM用户并完成授权。

  • 已配置环境变量ALIBABA_CLOUD_ACCESS_KEY_ID和ALIBABA_CLOUD_ACCESS_KEY_SECRET。具体操作，请参见配置环境变量。

    重要

    • 阿里云账号的AccessKey拥有所有API的访问权限，建议您使用RAM用户的AccessKey进行API访问或日常运维。

    • 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。

  • 已安装日志服务Java SDK。具体操作，请参见安装Java SDK。

  • 已写入日志到Logstore并开启索引。具体操作，请参见创建Logstore示例代码和创建索引。

注意事项

本示例以华东1（杭州）的公网Endpoint为例，其公网Endpoint为https://cn-hangzhou.log.aliyuncs.com。如果您通过与Project同地域的其他阿里云产品访问日志服务，请使用内网Endpointhttps://cn-hangzhou-intranet.log.aliyuncs.com。关于日志服务支持的地域与Endpoint的对应关系，请参见服务入口。

创建消费组示例代码

以下代码用于创建名为ali-test-consumergroup的消费组。

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;
        }
    }
}

预期结果如下：

ready to create consumergroup
create consumergroup ali-test-consumergroup success

修改消费组示例代码

以下代码用于修改名为ali-test-consumergroup的消费组信息。

import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.exception.LogException;

public class UpdateConsumerGroup {
    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-consumergroup";
            System.out.println("ready to update consumergroup");

            // 修改消费组超时时间为350秒。
            client.UpdateConsumerGroup(projectName, logstoreName, consumerGroupName, false, 350);

            System.out.println(String.format("update 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;
        }
    }
}

预期结果如下：

ready to update consumergroup
update consumergroup ali-test-consumergroup success

查询所有消费组示例代码

以下代码用于查询指定Logstore的所有消费组。

import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.common.ConsumerGroup;
import com.aliyun.openservices.log.exception.LogException;
import com.aliyun.openservices.log.response.ListConsumerGroupResponse;

public class ListConsumerGroup {
    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 {
            System.out.println("ready to list consumergroup");

            // 查询指定Logstore的所有消费组。
            ListConsumerGroupResponse response = client.ListConsumerGroup(projectName,logstoreName);

            for(ConsumerGroup consumerGroup : response.GetConsumerGroups()){
                System.out.println("ConsumerName is : " + consumerGroup.getConsumerGroupName());
            }

            System.out.println(String.format("list consumergroup from %s success",projectName));

        } 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;
        }
    }
}

预期结果如下：

ready to list consumergroup
ConsumerName is : ali-test-consumergroup2
ConsumerName is : ali-test-consumergroup
list consumergroup from ali-test-project success

删除消费组示例代码

以下代码用于删除目标Project下的消费组。

import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.exception.LogException;

public class DeleteConsumerGroup {
    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-consumergroup";
            System.out.println("ready to delete consumergroup");

            // 删除一个指定的消费组。
            client.DeleteConsumerGroup(projectName,logstoreName,consumerGroupName);

            System.out.println(String.format("delete 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;
        }
    }
}

预期结果如下：

ready to delete consumergroup
delete consumergroup ali-test-consumergroup success

获取消费组Checkpoint示例代码

以下代码用于获取指定消费组的Checkpoint。

import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.exception.LogException;
import com.aliyun.openservices.log.response.GetCheckPointResponse;

public class GetCheckPoint {
    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 = "consumerGroupX";
            System.out.println("ready to get consumergroup checkpoint");

            // 获取指定消费组中Shard的CheckPoint。
            GetCheckPointResponse response1 = client.getCheckpoint(projectName,logstoreName,consumerGroupName,0);
            GetCheckPointResponse response2 = client.getCheckpoint(projectName,logstoreName,consumerGroupName,1);
            System.out.println("The checkpoint of Shard 0 is : " + response1.getCheckpoint());
            System.out.println("The checkpoint of Shard 1 is : " + response2.getCheckpoint());

            System.out.println(String.format("get consumergroup %s checkpoint 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;
        }
    }
}

预期结果如下：

ready to get consumergroup checkpoint
The checkpoint of Shard 0 is : ConsumerGroupShardCheckPoint [shard=0, checkPoint=MTY2NzgxMDc0Nzk5MDk5MzAyMg==, updateTime=1668750821709044, consumer=consumer_1]
The checkpoint of Shard 1 is : ConsumerGroupShardCheckPoint [shard=1, checkPoint=MTY2NzgxMDc0Nzk5MTk0NTU0NQ==, updateTime=1668750828790425, consumer=consumer_1]
get consumergroup consumerGroupX checkpoint success

更新指定消费组Checkpoint示例代码

以下代码用于更新指定消费组的Checkpoint。

import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.common.ConsumerGroup;
import com.aliyun.openservices.log.exception.LogException;
import com.aliyun.openservices.log.response.ListConsumerGroupResponse;

public class ConsumerGroupUpdateCheckpoint {
    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 = "consumerGroupX";
            System.out.println("ready to update checkpoint");

            // 查询指定Logstore的所有消费组。
            ListConsumerGroupResponse response = client.ListConsumerGroup(projectName, logstoreName);

            for(ConsumerGroup consumerGroup : response.GetConsumerGroups()){
                System.out.println("ConsumerName is : " + consumerGroup.getConsumerGroupName());
                System.out.println("Consumer order is : " + consumerGroup.isInOrder());
            }

            // 更新Shard 0的Checkpoint。您可以通过GetCursor接口获取对应时间的Cursor。
            client.UpdateCheckPoint(projectName, logstoreName, consumerGroupName, 0, "MTY2NzgxMDc0Nzk5MTAwNjQ3Mg==");
            System.out.println(String.format("update checkpoint of %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;
        }
    }
}

预期结果如下：

ready to update checkpoint
ConsumerName is : consumerGroupX
Consumer order is : false
ConsumerName is : ali-test-consumergroup2
Consumer order is : true
ConsumerName is : ali-test-consumergroup
Consumer order is : false
update consumergroup checkpoint is:consumerGroupX
update checkpoint of consumerGroupX success

相关文档

  • 在调用API接口过程中，若服务端返回结果中包含错误信息，则表示调用API接口失败。您可以参考API错误码对照表查找对应的解决方法。更多信息，请参见API错误处理对照表。
  • 阿里云OpenAPI开发者门户提供调试、SDK、示例和配套文档。通过OpenAPI，您无需手动封装请求和签名操作，就可以快速对日志服务API进行调试。更多信息，请参见OpenAPI开发者门户。
  • 为满足越来越多的自动化日志服务配置需求，日志服务提供命令行工具CLI（Command Line Interface）。更多信息，请参见日志服务命令行工具CLI。

  • 关于消费组API接口说明，请参见如下：

    • CreateConsumerGroup

    • DeleteConsumerGroup

    • UpdateConsumerGroup

    • ListConsumerGroup

    • GetCheckPoint

    • UpdateCheckPoint

  • 更多示例代码，请参见Aliyun Log Java SDK on GitHub。

内容没看懂？ 不太想学习？想快速解决？ 有偿解决： 联系专家

阿里云企业补贴进行中： 马上申请

腾讯云限时活动1折起，即将结束： 马上收藏

同尘科技为腾讯云授权服务中心。

购买腾讯云产品享受折上折，更有现金返利:同意关联，立享优惠

转转请注明出处：http://www.yunxiaoer.com/162405.html

赞 (0)
0
上一篇 2023年12月10日
下一篇 2023年12月10日
详情页2

相关推荐

  • 阿里云大数据开发治理平台 DataWorksListQualityResultsByEntity-云淘科技

    调用ListQualityResultsByEntity，根据分区表达式查询历史校验结果。 调试 您可以在OpenAPI Explorer中直接运行该接口，免去您计算签名的困扰。运行成功后，OpenAPI Explorer可以自动生成SDK代码示例。 请求参数 名称 类型 是否必选 示例值 描述 Action String 是 ListQualityResu…

    阿里云大数据开发治理平台 DataWorks 2023年12月10日

  • 阿里云负载均衡DescribeServerCertificates – 查询指定地域的服务器证书列表-云淘科技

    查询指定地域的服务器证书列表。 接口说明 说明 为了保证安全性，只返回证书的指纹和名称，不返回证书和私钥的内容。调试您可以在OpenAPI Explorer中直接运行该接口，免去您计算签名的困扰。运行成功后，OpenAPI Explorer可以自动生成SDK代码示例。 调试调试授权信息下表是API对应的授权信息，可以在RAM权限策略语句的Action元素中使…

    阿里云负载均衡 2023年12月10日
  • 信息流广告，信息流部分建议宽度830px，只针对默认列表样式，顺序随机

  • 阿里云负载均衡UploadServerCertificate – 上传服务器证书-云淘科技

    上传服务器证书。 接口说明 一次只能上传一份服务器证书和对应的私钥。 上传成功后，返回该用户的所有服务器证书列表的指纹信息。 调试 您可以在OpenAPI Explorer中直接运行该接口，免去您计算签名的困扰。运行成功后，OpenAPI Explorer可以自动生成SDK代码示例。 调试调试授权信息下表是API对应的授权信息，可以在RAM权限策略语句的Ac…

    阿里云负载均衡 2023年12月10日

  • 阿里云RDS数据库DeleteDBInstanceEndpointAddress – 释放Endpoint外网连接地址-云淘科技

    该接口用于释放RDS MySQL集群系列实例的Endpoint的外网连接地址。 接口说明 适用引擎 RDS MySQL 注意事项 删除Endpoint中的地址，目前仅支持删除Endpoint的外网地址。如需删除内网地址，可以直接删除Endpoint。 调试 您可以在OpenAPI Explorer中直接运行该接口，免去您计算签名的困扰。运行成功后，OpenA…

    阿里云数据库 2023年12月9日
  • 阿里云RDS数据库创建数据库和账号-云淘科技 阿里云数据库

    阿里云RDS数据库创建数据库和账号-云淘科技

    本文介绍如何为RDS MySQL实例创建数据库和账号。 创建数据库 访问RDS实例列表，在上方选择地域，然后单击目标实例ID。 在左侧导航栏中单击数据库管理。 单击创建数据库。 设置如下参数。 参数 说明 数据库（DB）名称 长度为2~64个字符。 以小写字母开头，以小写字母或数字结尾。 支持小写字母、数字、下划线和中划线。 数据库名称在实例内必须唯一。 说…

    2023年12月9日

联系我们

400-800-8888

在线咨询： QQ交谈

邮件：admin@example.com

工作时间：周一至周五，9:30-18:30，节假日休息

关注微信
本站为广大会员提供阿里云、腾讯云、华为云、百度云等一线大厂的购买，续费优惠，保证底价，买贵退差。