2016-04-22 7 views
2

Ich versuche usgin Bibliothek clj-kafka.clj-kafka - Verbraucher leer

Hier mein Code

(use [clj-kafka.producer] 
     [clj-kafka.zk] 
     [clj-kafka.consumer.zk] 
     [clj-kafka.core])) 

(brokers {"zookeeper.connect" "localhost:2181"}) 

(def p (producer {"metadata.broker.list" "localhost:9092" 
        "serializer.class" "kafka.serializer.DefaultEncoder" 
        "partitioner.class" "kafka.producer.DefaultPartitioner"})) 

(send-message p (message "test" (.getBytes "this is my message"))) 

(def config {"zookeeper.connect" "localhost:2181" 
      "group.id" "clj-kafka.consumer" 
      "auto.offset.reset" "smallest" 
      "auto.commit.enable" "false"}) 

(with-resource [c (consumer config)] 
    shutdown 
    (take 2 (messages c "test"))) ;; return() 

I zookepper-Server und kafka sich mit

bin/zookeeper-server-start.sh config/zookeeper.properties 
bin/kafka-server-start.sh config/server.properties 

Die config/zookepper.properties starten:

dataDir=/tmp/zookeeper 
clientPort=2181 
maxClientCnxns=0 

und config/server.properties :

broker.id=0 
listeners=PLAINTEXT://:9092 
num.network.threads=3 
num.io.threads=8 
socket.send.buffer.bytes=102400 
zocket.receive.buffer.bytes=102400 
socket.request.max.bytes=104857600 
log.dirs=/tmp/kafka-logs 
num.partitions=1 
num.recovery.threads.per.data.dir=1 
log.retention.hours=168 
log.segment.bytes=1073741824 
log.retention.check.interval.ms=300000 
zookeeper.connect=localhost:2181 
zookeeper.connection.timeout.ms=6000 

Das 'Problem' ist, wenn ich ausführen:

(with-resource [c (consumer config)] 
    shutdown 
    (take 2 (messages c "test"))) ;; return empty() 

Jede Idee hier?

Vielen Dank im Voraus

Antwort

1

Sehen Sie diese Github issue. Scheint, die Dokumentation ist nicht großartig. Sie müssen die Realisierung der Sequenz (die faul ist) mit erzwingen. versuchen Sie dies:

(with-resource [c (consumer config)] 
    shutdown 
    (doall (take 2 (messages c "test")))) 
+0

Hallo @leeor Sie sind richtig. Ich habe dieses Detail hier vermisst. Und ja, die Dokumentation versagt in diesem Punkt. Danke – elf

+0

Hallo nochmal, Und wenn ich alle Nachrichten zum Thema nehmen möchte? Onde ist es eine faule Sequenz, ist es möglich? Ich versuche so etwas: '' (doseq [msg (doall (Nachrichten (Consumer Config) "Protokolle))] (lassen Sie [log (entschlüsseln (get msg: Wert))] (println log))) '' – elf