kafka restful api功能介绍与使用 - 分析设计

TOP

kafka restful api功能介绍与使用(二)

2019-09-17 17:47:11 【大中小】浏览:74次

rmat v1. Possible Error Codes * OFFSET_OUT_OF_RANGE (1) * UNKNOWN_TOPIC_OR_PARTITION (3) * NOT_LEADER_FOR_PARTITION (6) * REPLICA_NOT_AVAILABLE (9) * UNKNOWN (-1)

url_p="http://192.168.160.101:8082/topics/test_kfk/partitions/0/messages"
rst=requests.get(url_p,headers=header,params={"offset":3,"count":2})#,"count":2})
print(rst)
print(len(rst.json()))
if(rst.status_code!=500):
For itr in rst.json():
    print(base64.b64decode(itr['value']).decode())
print(rst.url)#http://192.168.160.101:8082/topics/test_kfk/partitions/0/messages?offset=3&count=2

View Code

获取当前订阅的topic列表.(api V2)

POST /consumers/(string:group_name)/instances/(string:instance)/subscription

获取手工指定的消费者的分区(api V2)

GET /consumers/(string:group_name)/instances/(string:instance)/assignments

GET /consumers/testgroup/instances/my_consumer/assignments HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Accept: application/vnd.kafka.v2+json
HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json
{
  "partitions": [
    {
      "topic": "test",
      "partition": 0
    },
    {
      "topic": "test",
      "partition": 1
    }
]
}

覆盖消费者即将消费的消息的偏移量(api V2)

POST /consumers/(string:group_name)/instances/(string:instance)/positions

POST /consumers/testgroup/instances/my_consumer/positions HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json
{
  "offsets": [
    {
      "topic": "test",
      "partition": 0,
      "offset": 20
    },
    {
      "topic": "test",
      "partition": 1,
      "offset": 30
    }
  ]
}

获取给定topic的分区的最后偏移

POST /consumers/(string:group_name)/instances/(string:instance)/positions/end

POST /consumers/testgroup/instances/my_consumer/positions/end HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json
{
  "partitions": [
    {
      "topic": "test",
      "partition": 0
    },
    {
      "topic": "test",
      "partition": 1
    }
]
}

使用分配和订阅api消费topic或者分区数据

GET /consumers/(string:group_name)/instances/(string:instance)/records

GET /consumers/testgroup/instances/my_consumer/records?timeout=3000&max_bytes=300000 HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Accept: application/vnd.kafka.binary.v2+json
Example binary response:
HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.binary.v2+json
[
  {
    "topic": "test",
    "key": "a2V5",
    "value": "Y29uZmx1ZW50",
    "partition": 1,
    "offset": 100,
  },
  {
    "topic": "test",
    "key": "a2V5",
    "value": "a2Fma2E=",
    "partition": 2,
    "offset": 101,
  }
]

首页上一页 1 2 下一页尾页 2/2/2
【大中小】【打印】【繁体】【投稿】【收藏】【推荐】【举报】【评论】【关闭】【返回顶部】

上一篇：「造个轮子」——cicada 设计一个..	下一篇：Java SSM 客户管理商户管理系统..