CKafka系列学习文章 - 用java调用Ckafka实例相关接口（十二）

原创

发哥说消息队列

修改于 2019-09-16 11:13:08

1.1K0

修改于 2019-09-16 11:13:08

文章被收录于专栏：发哥说消息队列发哥说消息队列

导语：创建后的实例信息，后期想进行调整，可以调用API进行管理。

一、获取实例列表

1、接口描述

接口请求域名：ckafka.api.qcloud.com 本接口（ListInstance）用于在用户账户下获取消息队列 CKafka 实例列表。

2、输入参数

以下请求参数列表仅列出了接口请求参数，其它参数见公共请求参数页面。

参数名称	是否必选	类型	描述
instanceId	否	String	（过滤条件）按照实例 ID 过滤。
searchWord	否	String	（过滤条件）按照实例名称过滤，支持模糊查询。
status.n	否	Int	（过滤条件）实例的状态。0：创建中，1：运行中，2：删除中，不填默认返回全部。
offset	否	Int	偏移量，不填默认为0。
limit	否	Int	返回数量，不填则默认为10，最大值为20。

3、java实现

 /* 获取实例列表 */
 public static String getListIinstance(String nonce,String timestamp) {
  Map<String, Object> param = new HashMap<>();
 param.put("Action", "ListInstance");
 param.put("Nonce", nonce);
 param.put("Timestamp",timestamp);
 param.put("SignatureMethod", signatureMethod);
 param.put("Region", "ap-guangzhou");
 param.put("SecretId", secretId);
  return getResult(param,url);
 }

4、调用

 //获取实例列表
  String str3=getListIinstance(nonce,timestamp);

5、结果

{"code":0,"message":"","codeDesc":"Success","data":{"totalCount":2,"instanceList":[{"instanceId":"ckafka-vekvsdp","instanceName":"bowenqiu_ckafka","status":1,"ifCommunity":true},{"instanceId":"ckafka-qwkshhnx","instanceName":"bearsi_cdn_log_test","status":1,"ifCommunity":true}]}}

二、获取实例属性

1、接口描述

接口请求域名：ckafka.api.qcloud.com 本接口（GetInstanceAttributes）用于在用户账户下获取消息队列 CKafka 实例属性。

2、输入参数

以下请求参数列表仅列出了接口请求参数，其它参数见公共请求参数页面。

参数名称	是否必选	类型		描述
instanceId	是	String		实例 ID

3、java实现

 /* 获取实例属性 */
 public static String GetInstanceAttributes(String nonce,String timestamp) {
  Map<String, Object> param = new HashMap<>();
 param.put("Action", " GetInstanceAttributes ");
 param.put("Nonce", nonce);
 param.put("instanceId",instanceId);
 param.put("Timestamp",timestamp);
 param.put("SignatureMethod", signatureMethod);
 param.put("Region", "ap-guangzhou");
 param.put("SecretId", secretId);
  return getResult(param,url);
 }

4、调用

//获取实例属性
  String str4=GetInstanceAttributes(nonce,timestamp);

5、结果

{"code":0,"message":"","codeDesc":"Success","data":{"tags":[],"instanceId":"ckafka-3vekvsdp","instanceName":"Jensen_ckafka","vipList":[{"vip":"10.66.249.112","vport":"9092"}],"vip":"10.66.249.112","vport":"9092","status":1,"bandwidth":320,"diskSize":300,"zoneId":100003,"vpcId":"","subnetId":"","healthy":1,"healthyMessage":"","createTime":1566388895,"expireTime":1569067295,"msgRetentionTime":2880,"config":{"auto.create.topics.enable":"true","num.partitions":2,"default.replication.factor":2},"remainderPartitions":43,"remainderTopics":22,"createdPartitions":17,"createdTopics":3}}

三、设置实例属性

1、接口描述

接口请求域名：ckafka.api.qcloud.com 本接口（SetInstanceAttributes）用于设置消息队列 CKafka 实例属性。

2、输入参数

以下请求参数列表仅列出了接口请求参数，其它参数见公共请求参数文档。

参数名称	是否必选	类型	描述
instanceId	是	String	实例 ID。
instanceName	否	String	待修改的实例名称，实例名称长度不超过64字节。
msgRetentionTime	否	Int	实例日志的最长保留时间，单位：分钟，最小为1min，最大为30天。
autoCreateTopicEnable	否	Int	是否开启自动创建 topic，1：开启，0：不开启。
defaultNumPartitions	否	Int	自动创建主题分区个数，如果 autoCreateTopicEnable 设置为 true，没有设置该值时，默认设置为3。
defaultReplicationFactor	否	Int	自动创建主题副本数，如果 autoCreateTopicEnable 设置为 true，没有设置该值时，默认设置为2。

3、java实现

/* 设置实例属性 */
    public static String SetInstanceAttributes(String nonce,String timestamp) {

    	Map<String, Object> param = new HashMap<>();
        param.put("Action", " SetInstanceAttributes");
        param.put("Nonce", nonce);
        param.put("instanceId",instanceId);
        param.put("Timestamp",timestamp);
        param.put("SignatureMethod", signatureMethod);
        param.put("Region", "ap-guangzhou");
        param.put("SecretId", secretId);
        //将实例的名称修改为Jensen_ckafka
        param.put("instanceName", "Jensen_ckafka");
        //将实例日志的最长保留时间（单位：分钟），2天：60*24*2
        param.put("msgRetentionTime", 2880);
        //是否开启自动创建 topic，1：开启 ，0：不开启
        param.put("autoCreateTopicEnable ", 1); //这个参数没设置时，下面的两个参数是不生效的。
        //自动创建主题分区个数，如果 autoCreateTopicEnable 设置为 true， 没有设置该值时，默认设置为3。
        param.put("defaultNumPartitions", 2);
        //自动创建主题副本数，如果 autoCreateTopicEnable 设置为 true， 没有设置该值时，默认设置为2。
        param.put("defaultReplicationFactor",2);
    	return getResult(param,url);
    }

4、调用

    	//设置实例属性
    	String str5=SetInstanceAttributes(nonce,timestamp);

5、结果

{"code":0,"message":"","codeDesc":"Success"}

四、查询消费分组信息

1、接口描述

本接口（ListConsumerGroup）用于在用户账户下获取 CKafka 消费分组信息。

接口请求域名：ckafka.api.qcloud.com

2、输入参数

以下请求参数列表仅列出了接口请求参数，其它参数见公共请求参数页面。

参数名称	是否必选	类型	描述
instanceId	是	String	（过滤条件）按照实例 ID 过滤
groupName	否	String	消费分组名称（精确匹配）
topicName	否	String	主题名称，若不填 groupName，则该参数会被忽略
offset	否	Int	偏移量，不填默认为0
limit	否	Int	返回数量，不填则默认50，最大值50

3、java实现

/* 
	 *  查询消费分组信息 
	 * 查询所有的消费分组信息
	 */
    public static String GetListConsumerGroup(String nonce,String timestamp) {

    	Map<String, Object> param = new HashMap<>();
        param.put("Action", "ListConsumerGroup");
        param.put("Nonce", nonce);
        param.put("instanceId",instanceId);
        param.put("Timestamp",timestamp);
        param.put("SignatureMethod", signatureMethod);
        param.put("Region", "ap-guangzhou");
        param.put("SecretId", secretId);
    	return getResult(param,url);
    }

4、调用

    	//查询消费分组信息 
    	String str6=GetListConsumerGroup(nonce,timestamp);

5、结果

{"code":0,"message":"","codeDesc":"Success","data":{"totalCount":11,"topicList":
{"topic_test1":"topic-7p2pzlo4","bowenqiu_test1":"inter-topic-kfjiqvgo","bowenqiu_topic1":"topic-3yueh1ge"},"groupList":{
"console-consumer-17873":{"topic_test1":[349333]},
"console-consumer-18492":{"topic_test1":[349358]},
"console-consumer-2529":{"bowenqiu_test1":[1,0,0],"topic_test1":[349344]},
"console-consumer-31005":{"topic_test1":[349333]},
"console-consumer-39207":{"topic_test1":[349335]},
"console-consumer-44775":{"topic_test1":[349390]},
"console-consumer-54312":{"topic_test1":[349353]},
"console-consumer-56107":{"topic_test1":[349341]},
"console-consumer-56480":{"topic_test1":[349362]},
"console-consumer-92575":{"topic_test1":[349368]},
"console-consumer-92728":{"topic_test1":[349374]}},
"totalPartition":0,"partitionListForMonitor":[],"totalTopic":2,"topicListForMonitor":[{"topicName":"topic_test1","topicId":"topic-7p2pzlo4"},{"topicName":"bowenqiu_test1","topicId":"inter-topic-kfjiqvgo"}],"groupListForMonitor":[{"groupName":"console-consumer-17873"},{"groupName":"console-consumer-18492"},{"groupName":"console-consumer-2529"},{"groupName":"console-consumer-31005"},{"groupName":"console-consumer-39207"},{"groupName":"console-consumer-44775"},{"groupName":"console-consumer-54312"},{"groupName":"console-consumer-56107"},{"groupName":"console-consumer-56480"},{"groupName":"console-consumer-92575"},{"groupName":"console-consumer-92728"}]}}

五、查询消费分组信息（精简版）

1、接口描述

本接口 (ListGroup) 用于在用户账户下获取 CKafka 消费分组信息，该接口为精简版接口，返回消费分组信息较少，可以调用 GetGroupInfo 查询消费分组详细信息。

接口请求域名：ckafka.api.qcloud.com

2、输入参数

以下请求参数列表仅列出了接口请求参数，其它参数见公共请求参数页面。

参数名称	是否必选	类型	描述
instanceId	是	String	（过滤条件）按照实例 ID 过滤。
group	否	String	消费分组（精确匹配）。
searchWord	否	String	模糊匹配 group。
offset	否	Int	偏移量，不填默认为 0。
limit	否	Int	返回数量，不填则默认 20，最大值 50。

3、java实现

	/* 查询消费分组信息（精简版） */
    public static String GetListGroup(String nonce,String timestamp) {

    	Map<String, Object> param = new HashMap<>();
        param.put("Action", "ListGroup");
        param.put("Nonce", nonce);
        param.put("instanceId",instanceId);//不加，会报4000错误
        param.put("Timestamp",timestamp);
        param.put("SignatureMethod", signatureMethod);
        param.put("Region", "ap-guangzhou");
        param.put("SecretId", secretId);
    	return getResult(param,url);
    }

4、调用

    	//查询消费分组信息（精简版）
    	String str7=GetListGroup(nonce,timestamp);

5、结果

{"code":0,"message":"","codeDesc":"Success","data":{"totalCount":11,"groupList":[{"group":"console-consumer-92728","protocol":"consumer"},{"group":"console-consumer-92575","protocol":"consumer"},{"group":"console-consumer-2529","protocol":"consumer"},{"group":"console-consumer-56107","protocol":"consumer"},{"group":"console-consumer-56480","protocol":"consumer"},{"group":"console-consumer-17873","protocol":"consumer"},{"group":"console-consumer-18492","protocol":"consumer"},{"group":"console-consumer-44775","protocol":"consumer"},{"group":"console-consumer-54312","protocol":"consumer"},{"group":"console-consumer-31005","protocol":"consumer"},{"group":"console-consumer-39207","protocol":"consumer"}]}}

六、获取消费分组信息

1、接口描述

本接口（GetGroupInfo）用于在用户账户下获取 CKafka 消费分组详细信息。接口请求域名：ckafka.api.qcloud.com

2、输入参数

以下请求参数列表仅列出了接口请求参数，其它参数见公共请求参数页面。

参数名称	是否必选	类型	描述
instanceId	是	String	（过滤条件）按照实例 ID 过滤。
group.N	是	String Array	Kafka 消费分组，Consumer-group，这里是数组形式，格式：group.0=xxx&group.1=yyy。

3、java实现

	/* 获取消费分组信息 */
    public static String GetGroupInfo(String nonce,String timestamp) {

    	Map<String, Object> param = new HashMap<>();
        param.put("Action", "GetGroupInfo");
        param.put("Nonce", nonce);
        param.put("Timestamp",timestamp);
        param.put("SignatureMethod", signatureMethod);
        param.put("Region", "ap-guangzhou");
        param.put("SecretId", secretId);
        param.put("instanceId",instanceId);
        param.put("group.0","console-consumer-2529");
        param.put("group.1","console-consumer-56107");
    	return getResult(param,url);
    }

4、调用

    	//获取消费分组信息
    	String str8=GetGroupInfo(nonce,timestamp);

5、结果

{"code":0,"message":"","codeDesc":"Success","data":[
{"errCode":0,"state":"Stable","protocolType":"consumer","protocol":"range","group":"console-consumer-2529","members":[{"member_id":"consumer-1-\/10.1.1.250-2019-09-09 20:18:16:801-6fdd1719-ff19-4bc1-804b-b833a0925523","client_id":"consumer-1","client_host":"\/10.1.1.250","assignment":{"version":0,"topics":[{"topic":"topic_test1","partitions":[0]}]}}]},
{"errCode":0,"state":"Empty","protocolType":"consumer","protocol":"","group":"console-consumer-56107","members":[]}]}

七、获取消费分组offset

1、接口描述

接口请求域名：ckafka.api.qcloud.com 本接口（GetGroupOffsets）用于在用户账户下获取 CKafka 消息分组 offset。

2、输入参数

以下请求参数列表仅列出了接口请求参数，其它参数见公共请求参数页面。

参数名称	是否必选	类型	描述
instanceId	是	String	（过滤条件）按照实例 ID 过滤
group	是	String	Kafka 消费分组
topics	否	String Array	group 订阅的主题名称数组，如果没有该数组，则表示指定的 group 下所有 topic 信息
searchWord	否	String	模糊匹配 topicName
offset	否	Int	本次查询的偏移位置，默认为0
limit	否	Int	本次返回结果的最大个数，默认为50，最大值为50

3、java实现

    /* 获取消费分组 offset接口 */
    public static String GetGroupOffsets(String nonce,String timestamp) {
    	
    	Map<String, Object> param = new HashMap<>();
        param.put("Action", "GetGroupOffsets");
        param.put("Nonce", nonce);
        param.put("instanceId",instanceId);
        param.put("Timestamp",timestamp);
        param.put("SignatureMethod", signatureMethod);
        param.put("Region", "ap-guangzhou");
        param.put("SecretId", secretId);
        param.put("group", "console-consumer-2529");      
    	return getResult(param,url);
    }

4、调用

    	//获取消费分组 offset接口 
    	String str1=GetGroupOffsets(nonce,timestamp);

5、结果

{"code":0,"message":"","codeDesc":"Success","data":{"totalCount":2,"topicList":[
{"topic":"bowenqiu_test1","partitions":
[{"partition":0,"offset":1,"meta":null,"errCode":0,"logEndOffset":1,"lag":0},
{"partition":1,"offset":0,"meta":null,"errCode":0,"logEndOffset":0,"lag":0},
{"partition":2,"offset":0,"meta":null,"errCode":0,"logEndOffset":0,"lag":0}]},
{"topic":"topic_test1","partitions":
[{"partition":0,"offset":349416,"meta":null,"errCode":0,"logEndOffset":349416,"lag":0}]}]}}

八、设置消费分组offset

1、接口描述

本接口 (SetGroupOffsets) 用于在用户账户下设置 CKafka 实例某个消费分组 offset。

接口请求域名：ckafka.api.qcloud.com

2、输入参数

以下请求参数列表仅列出了接口请求参数，其它参数见公共请求参数页面。

参数名称	是否必选	类型	描述
instanceId	是	String	（过滤条件）按照实例 ID 过滤。
group	是	String	kafka 消费分组。
topics	否	String Array	表示需要重置 offset 的 topic 数组，不填表示全部 topic。
strategy	是	Int	重置 offset 的策略，入参含义： 0：对齐 shift-by 参数，代表把 offset 向前或向后移动 shift 条。 1：对齐参考(by-duration,to-datetime,to-earliest,to-latest),代表把 offset 移动到指定timestamp的位置。 2：对齐参考(to-offset)，代表把 offset 移动到指定的 offset 位置。
shift	否	Int	当 strategy 为 0 时，必须包含该字段，可以大于零代表会把 offset 向后移动 shift 条，小于零则将 offset 向前回溯 shift 条数。正确重置后新的 offset 应该是(old_offset + shift)，如果新的 offset 小于 partition 的 earliest 则会设置为 earliest，如果大于partition 的 latest 则会设置为 latest。
timestamp	否	Int	单位：ms。当 strategy 为 1 时，必须包含该字段，其中 -2 表示重置 offset 到最开始的位置，-1 表示重置到最新的位置(相当于清空)，其它值则代表指定的时间，会获取 topic 中指定时间的 offset 然后进行重置，如果指定的时间不存在消息，则获取最末尾的 offset。
offset	否	Int	需要重新设置的 offset 位置。当 strategy 为 2 时，必须包含该字段。

3、java实现

    /* 设置消费分组 offset */
    public static String SetGroupOffsets(String nonce,String timestamp) {
    	Map<String, Object> param = new HashMap<>();
        param.put("Action", "SetGroupOffsets");
        param.put("Nonce", nonce);
        param.put("Timestamp",timestamp);
        param.put("SignatureMethod", signatureMethod);
        param.put("Region", "ap-guangzhou");
        param.put("SecretId", secretId);
        param.put("instanceId",instanceId);
        param.put("group", "console-consumer-2529");  
        param.put("topics.0","topic_test1");//String Array的表示方式
        param.put("topics.1","bowenqiu_test1");
        param.put("strategy",0);
        param.put("shift",2);
    	return getResult(param,url);
    }

4、调用

    	//设置消费分组 offset
    	String str9=SetGroupOffsets(nonce,timestamp);

5、结果

{"code":0,"message":"","codeDesc":"Success","data":{"succ":[{"topic":"topic_test1","error_code":0,"partitions":[{"partition":0,"offset":349416,"error_code":0}]},{"topic":"bowenqiu_test1","error_code":0,"partitions":[{"partition":2,"offset":0,"error_code":0},{"partition":1,"offset":0,"error_code":0},{"partition":0,"offset":1,"error_code":0}]}],"failed":[]}}

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

kafka