使用日志服务SDK创建消费组_日志服务(SLS)-阿里云帮助中心

使用CreateConsumerGroup接口创建消费组。当您使用第三方软件、多语言应用、云产品、流式计算框架等通过SDK实时消费日志服务的数据时，无法满足日志服务的实现细节及消费者之间的负载均衡、故障转移（Failover）等，您可以通过消费组消费日志，消费组提供秒级实时消费。

前提条件

您已完成以下操作：

参数说明

def create_consumer_group(self, project, logstore, consumer_group, timeout, in_order=False):

请求参数

参数	类型	是否必填	说明
project	String	是	Project的名称。
logstore	String	是	Logstore的名称。
consumer_group	String	是	消费组的名称，在Logstore下必须唯一。
timeout	int	是	超时时间。在超时时间段内没有收到心跳，消费者将被删除。单位：秒。
in_order	bool	否	同一Shard上消费数据的顺序与数据写入时的顺序严格一致，in_order选项只影响分裂或合并前后的不同Shard间消费的顺序，默认为False。 True：按顺序消费。 Shard分裂后，先消费原Shard数据，消费完原Shard数据后，才会开始消费新分裂的Shard的数据。 Shard合并后，先消费原Shard数据，消费完原Shard数据后，才会开始消费合并后的Shard的数据。 False：不按顺序消费。即所有Shard同时开始消费，Shard分裂或者合并后产生的新Shard，也会立即消费。

返回参数

请参见CreateConsumerGroup - 创建消费组。

示例代码

from aliyun.log import LogClient
import os


def main():
    # 本示例从环境变量中获取AccessKey ID和AccessKey Secret
    access_key_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
    access_key_secret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')
    # 日志服务的服务接入点
    endpoint = "cn-hangzhou.log.aliyuncs.com"

    client = LogClient(endpoint, access_key_id, access_key_secret)

    project = "ali-test-peoject"
    logstore = "test-logstore"
    consumer_group = "consumer_group_test"
    timeout = 30
    in_order = False

    try:
        response = client.create_consumer_group(project, logstore, consumer_group, timeout, in_order)
        response.log_print()
    except Exception as e:
        print(f"An error occurred while creating the consumer group: {e}")


if __name__ == '__main__':
    main()

前提条件

参数说明

请求参数

返回参数

示例代码

相关文档