我想去图书馆 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服务器和kafka本身
bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties
config/zookeper.properties:
dataDir=/tmp/zookeeper
clientPort=2181
maxClientCnxns=0
和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
“问题”是当我执行:
(with-resource [c (consumer config)]
shutdown
(take 2 (messages c "test"))) ;; return empty ()
你知道吗?
提前谢谢
1条答案
按热度按时间gmol16391#
查看此github问题。似乎文件不是很好。你必须强迫序列的实现(这是懒惰的)
doall
. 试试这个: