使用原生方法从kafka消费消息

Posted 2020-06-13

    kafka最早是linkedin开发的一套高性能类队列结构，具有发布—订阅功能。现在是apache的项目之一。支持很多种客户端从其中进行consume，网上也有许多第三方的客户端（注1），但下面我们只使用其自己的包中的方法，来进行consume。我们的这个例子是从一个servlet中调用kafka的Consumer相关类，来读取远端kafka中的message。

代码如下：

    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        String topic = "test";
        List<Message> list = new ArrayList<Message>();
        StringBuilder builder = new StringBuilder();
        KafkaHttpConsumer consumer = new KafkaHttpConsumer();
        list = consumer.consume(topic);
        builder.append("[");
        for(int i=0; i<list.size(); i++){
            builder.append(list.get(i).message);
            builder.append(",");
        }
        builder.deleteCharAt(builder.length()-1);
        builder.append("]");
        response.getWriter().append(builder.toString());
    }

import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import com.fasterxml.jackson.annotation.JsonInclude;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerTimeoutException;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;

public class KafkaHttpConsumer {

    public List<Message> consume(String topic) {
        Properties prop = new Properties();
        try {
            prop.load(this.getClass().getResourceAsStream("/kafka-http.properties"));
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
        ConsumerConfig config = new ConsumerConfig(prop);
        ConsumerConnector connector = Consumer.createJavaConsumerConnector(config);

        Map<String, Integer> streamCounts = Collections.singletonMap(topic, 1);
        Map<String, List<KafkaStream<byte[], byte[]>>> streams = connector.createMessageStreams(streamCounts);
        KafkaStream<byte[], byte[]> stream = streams.get(topic).get(0);

        List<Message> messages = new ArrayList<>();
        try {
            for (MessageAndMetadata<byte[], byte[]> messageAndMetadata : stream)
                messages.add(new Message(messageAndMetadata));
        } catch (ConsumerTimeoutException ignore) {
        } finally {
            connector.commitOffsets();
            connector.shutdown();
        }
        return messages;
    }

    /* for test */
    public static void main(String[] args) {
        Properties prop = new Properties();
        try {
            prop.load(KafkaHttpConsumer.class.getResourceAsStream("/kafka-http.properties"));
            Iterator<Object> ite = prop.keySet().iterator();
            while(ite.hasNext()){
                String key = (String)ite.next();
                System.out.println("value:" + prop.getProperty(key));
            }
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } 
    }

    public static class Message {
        public String topic;

        @JsonInclude(JsonInclude.Include.NON_NULL)
        public String key;
        public String message;

        public int partition;
        public long offset;

        public Message(MessageAndMetadata<byte[], byte[]> message) {
            this.topic = message.topic();

            this.key = message.key() != null ? new String(message.key(), Charset.forName("utf-8")) : null;
            this.message = new String(message.message(), Charset.forName("utf-8"));

            this.partition = message.partition();
            this.offset = message.offset();
        }
    }
}

 

注1：https://cwiki.apache.org/confluence/display/KAFKA/Clients