clj-kafka - 消費者空

Question

我正在嘗試usgin庫clj-kafka。

這裏我的代碼

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

我開始zookepper服務器和卡夫卡本身

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

配置/ zookepper.properties：

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

和配置/ 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 

'問題' 是，當我執行：這裏

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

任何想法？

在此先感謝

Answer 1

看到這個github上issue。似乎文檔不是很好。您必須強制使用doall實現序列（這是惰性的）。試試這個：

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