Kafka Rest Proxy基础（java版）

import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.net.URISyntaxException;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.json.simple.parser.JSONParser;
/**
 * 
 *  compile group: 'org.apache.httpcomponents', name: 'httpclient', version: '4.5.12'
    compile "com.googlecode.json-simple:json-simple:1.1.1"
    compile "org.apache.kafka:kafka-clients:0.10.2.0"
	compile "org.apache.kafka:kafka_2.11:0.10.2.0"
 */
public class KafkaRestProxyDemo {
	private static String kafka_rest_baseUrl = "http://127.0.0.1:8082/";
	private static CloseableHttpClient client = HttpClients.createDefault();

	public static void main(String[] args) {

		try {
			//get();
			post();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	public static void get() throws Exception {
		HttpGet get = new HttpGet(kafka_rest_baseUrl + "topics");
		HttpResponse response = client.execute(get);
		if (response.getStatusLine().getStatusCode() == 200) {
			HttpEntity entity = response.getEntity();
			InputStream content = entity.getContent();
			json(content);
		}
	}	 
	public static void post() throws URISyntaxException {
		try {
			HttpPost post = new HttpPost(kafka_rest_baseUrl + "topics/" + "xxx444");
			post.setHeader("Content-Type", "application/vnd.kafka.binary.v1+json");
			HttpEntity reqEntity = new StringEntity("{\"records\":[{\"value\":\"5Lit5paHIGhlbGxvIHdvcmQ=\"}]}");
			post.setEntity(reqEntity);
			HttpResponse response = client.execute(post); 
			if (response.getStatusLine().getStatusCode()==200) {
				 json(response.getEntity().getContent());
			}else {
				System.out.println("error:"+response.getStatusLine());
			}
		} catch ( Exception e) {
			e.printStackTrace();
		}
	}

	public static void json(InputStream inputStream) {
		try {
			Reader reader = new InputStreamReader(inputStream);
			Object parse = new JSONParser().parse(reader);
			System.out.println(parse.toString());
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

}

 

本文地址：https://blog.csdn.net/qq_25337221/article/details/110231855