前述 採用confluent kafka-rest proxy實現kafka restful service時候（具體參考上一篇筆記），通過http協議數據傳輸，需要註意的是採用了base64編碼（或者稱之為加密），如果消息再post之前不採用base64處理將會出現：服務端消息亂碼、程式報錯等，因 ...
- 前述
採用confluent kafka-rest proxy實現kafka restful service時候（具體參考上一篇筆記），通過http協議數據傳輸，需要註意的是採用了base64編碼（或者稱之為加密），如果消息再post之前不採用base64處理將會出現：服務端消息亂碼、程式報錯等，因此正常的處理流程是：
1.先對待post的消息做UTF-8統一處理
2.採用base64編碼包處理消息
s='Kafka,hi' ad="hi,kafka,i'm xnchall" aa=ad.encode()#UTF-8統一處理 print(aa) b64=base64.b64encode(ad.encode())#base64編碼包統一處理
- 利用kafka-rest生產消息
POST /topics/(string:topic_name)
data={"records":[
{
"key":"a2V5",
"value":"Y29uZmx1ZW50"
},
{
"value":"a2Fma2E=",
"partition":1
},
{
"value":"bG9ncw=="
}
]}
data1={"records":[{"value":"5bCK5pWs55qE5a6i5oi35oKo5aW977yMaGkga2Fma2EsIGknbSB4bmNoYWxs"}]}
header={"Content-Type":"application/vnd.kafka.v1+json"}
r=requests.post(url=url,json=data,headers=header)
r=requests.post(url=url,json=data1,headers=header)
View Code
- 向指定分區生產消息：Produce messages to one partition of the topic
POST /topics/(string:topic_name)/partitions/(int:partition_id)
ad="hi kafka,i'm xnchall" url11="http://192.168.160.101:8082/topics/test_kfk_lk/partitions/1" data2={"records":[{"value":(base64.b64encode(ad.encode())).decode()}]} print(data2) r2=requests.post(url=url11,json=data2,headers=header) print(r2) print(r2.content)View Code
- 創建或者註冊消費實例：Create a new consumer instance in the consumer group
POST /consumers/(string:group_name)
url3="http://192.168.160.101:8082/consumers/my_group" data3={ "id":"my_consumer1", "format":"binary", "auto.offset.reset":"smallest", "auto.commit.enable":"false" } r3=requests.post(url=url3,json=data3,headers=header)View Code
- 提交偏移 Commit offsets for the consumer
POST /consumers/(string:group_name)/instances/(string:instance)/offsets
url4="http://192.168.160.101:8082/consumers/my_group/instances/my_consumer1/offsets" r4=requests.post(url=url4,headers=header)View Code
- 消費消息
GET /consumers/(string:group_name)/instances/(string:instance)/topics/(string:topic_name)
url_get2="http://192.168.160.101:8082/consumers/my_group/instances/my_consumer1/topics/test_kfk_lk" rr2=requests.get(url=url_get2,headers=header)#,params={"timeout":3000000} print(rr2) print(rr2.content) print(rr2.text)View Code
- 刪除消費者實例 Destroy the consumer instance
DELETE /consumers/(string:group_name)/instances/(string:instance)
#url_del="http://192.168.160.101:8082/consumers/test_kfk_lk/instances/my_consumer" #d1=requests.delete(url_del)#刪除消費者實例 #print(d1)View Code
- 獲取指定分區、偏移消息： Consume messages from one partition of the topic.(api V2)
GET /topics/(string:topic_name)/partitions/(int:partition_id)/messages?offset=(int)[&count=(int)]
Fetch Response v1 only contains message format v0. Fetch Response v2 might either contain message format v0 or message format 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=2View 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,
}
]
