卡夫卡休息的例子答案

【问题标题】：Examples of Kafka Rest卡夫卡休息的例子
【发布时间】：2016-12-23 08:39:34
【问题描述】：

有没有使用 Java 中的 Kafka rest api 的生产者和消费者组的好例子。我不是在寻找生产者和消费者的 simpleconsumer 或 kafka 客户端示例。任何帮助表示赞赏。

【问题讨论】：

标签： apache-kafka kafka-consumer-api kafka-producer-api confluent-platform

【解决方案1】：

最好实现生产者和消费者，然后为生产者和消费者集成rest API。

producer(){
//your implementation for producer
}

consumer(){
//your implementation for consumer
}

REST API：

@POST
restProducer(){
producer();
}

@GET
restConsumer(){
consumer();
}

--否则

尝试利用融合实现的 REST API

http://docs.confluent.io/1.0/kafka-rest/docs/intro.html

【讨论】：

【解决方案2】：

这是来自 Confluent 的示例 Rest API（Rest Proxy）代码。 不幸的是，不是在 Java 中，而是在 Python 中。 :(
我必须输入它，所以它可能包含一些拼写错误。我希望这对您有所帮助。

（使用 Python 编写的 REST API 的生产者）

import requests
import base64
import json

url = "http://restproxy:8082/topics/my_topic"
headers = {
    "Content-Type" : "application/vnd.kafka.binary.v1 + json",
}
# Create one or more messages
payload = {"records":
       [{
           "key":base64.b64encode('firstkey'),
           "value":base64.b64encode('firstvalue'),
       }],
}
# Send the message
r = requests.post(url, data=json.dumps(payload), headers=headers)
if r.status_code != 200:
   print("Status Code: " + str(r.status_code))
   print(r.text)

（消费者使用 Python 编写的 Rest API）

import requests
import base64
import json
import sys

#Base URL for interacting with REST server
baseurl = "http://restproxy:8082/consumers/group1"

#Create the Consumer instance
print("Creating consumer instance")
payload {
    "format": "binary",
}
headers = {
    "Content-Type" : "application/vnd.kafka.v1+json",
}
r = requests.post(baseurl, data=json.dumps(payload), headers=headers)

if r.status_code !=200:
    print("Status Code: " + str(r.status_code))
    print(r.text)
    sys.exit("Error thrown while creating consumer")

# Base URI is used to identify the consumer instance
base_uri = r.json()["base_uri"]

#Get the messages from the consumer
headers = {
    "Accept" : "application/vnd.kafka.binary.v1 + json",
}

# Request messages for the instance on the Topic
r = requests.get(base_uri + "/topics/my_topic", headers = headers, timeout =20)

if r.status_code != 200: 
    print("Status Code: " + str(r.status_code))
    print(r.text)
    sys.exit("Error thrown while getting message")

# Output all messages
for message in r.json():
    if message["key"] is not None:
        print("Message Key:" + base64.b64decode(message["key"]))
    print("Message Value:" + base64.b64decode(message["value"]))

# When we're done, delete the consumer
headers = {
    "Accept" : "application/vnd.kafka.v1+json",
}

r = requests.delete(base_uri, headers=headers)

if r.status_code != 204: 
    print("Status Code: " + str(r.status_code))
    print(r.text)

【讨论】：

如何从这里删除记录以发送数据，例如 {"Key1" : "value1", "key2": "value2"}