使用原生方法從kafka消費消息


    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(); } } }

 kafka-http.properties

#for read from kafka
zookeeper.connect=192.20.34.144:2181
group.id=group
auto.offset.reset=smallest
consumer.timeout.ms=500

 

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


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM