kafka 小案例【二】 --kafka 設置多個消費着集群

這個配是我在http://www.cnblogs.com/zhangXingSheng/p/6646972.html 的基礎上再添加的配置

設置多個消息集群

（1）復制兩份配置文件

> cp config/server.properties config/server-1.properties
> cp config/server.properties config/server-2.properties

（2）編輯配置文件

server1.properties

broker.id=1
listeners=PLAINTEXT://:9093
log.dir=/home/zhangxs/datainfo/developmentData/kafka/log/log1

server2.propreties

broker.id=2
listeners=PLAINTEXT://:9094
log.dir=/home/zhangxs/datainfo/developmentData/kafka/log/log2

(3)啟動這個兩個消息服務

 bin/kafka-server-start.sh config/server-1.properties
 bin/kafka-server-start.sh config/server-2.properties

（4）創建一個新的topic

> bin/kafka-topics.sh --create --zookeeper  192.168.177.120:2181 --replication-factor 3 --partitions 1 --topic test2

(5)啟動2個消費者進程，來訂閱這個topic test2

> bin/kafka-console-consumer.sh --bootstrap-server 192.168.177.120:9092 --from-beginning --topic
 test2

(6)輸入消息

linux

(7)顯示結果

 

遇到的問題：

（2）在我啟動生產者集群的時候，拋出的異常

kafka.common.InconsistentBrokerIdException: Configured broker.id 1 doesn't match
 stored broker.id 0 in meta.properties. If you moved your data, make sure your c
onfigured broker.id matches. If you intend to create a new broker, you should re
move all data in your data directories (log.dirs).
    at kafka.server.KafkaServer.getBrokerId(KafkaServer.scala:686)
    at kafka.server.KafkaServer.startup(KafkaServer.scala:194)
    at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:
39)
    at kafka.Kafka$.main(Kafka.scala:67)
    at kafka.Kafka.main(Kafka.scala)
[2017-03-30 21:31:32,267] INFO shutting down (kafka.server.KafkaServer)
[2017-03-30 21:31:32,269] INFO Terminate ZkClient event thread. (org.I0Itec.zkcl
ient.ZkEventThread)
[2017-03-30 21:31:32,389] INFO Session: 0x15b1f53fc120002 closed (org.apache.zoo
keeper.ZooKeeper)
[2017-03-30 21:31:32,392] INFO EventThread shut down for session: 0x15b1f53fc120
002 (org.apache.zookeeper.ClientCnxn)
[2017-03-30 21:31:32,393] INFO shut down completed (kafka.server.KafkaServer)
[2017-03-30 21:31:32,394] FATAL Fatal error during KafkaServerStartable startup.
 Prepare to shutdown (kafka.server.KafkaServerStartable)
kafka.common.InconsistentBrokerIdException: Configured broker.id 1 doesn't match stored broker.id 0 in meta.properties. If you moved your data, make s
ure your configured broker.id matches. If you intend to create a new broker, you should remove all data in your data directories (log.dirs).
    at kafka.server.KafkaServer.getBrokerId(KafkaServer.scala:686)
    at kafka.server.KafkaServer.startup(KafkaServer.scala:194)
    at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:39)
    at kafka.Kafka$.main(Kafka.scala:67)
    at kafka.Kafka.main(Kafka.scala)
因為server1.properties的broker.id=0，與server.properties配置的broker,id重復

這個是因為server1.properties配置文件的上面已經配置broker.id=0,我沒有把這段配置注釋掉，導致我的配置沒有生效，與server.properties配置的broker,id重復。

（2）在我啟動另一個生產者服務拋出異常

 

kafka.common.KafkaException: Found directory /home/zhangxs/datainfo/developmentData/kafka/log/log1, 'log1' is not in the form of topic-partition
If a directory does not contain Kafka topic data it should not exist in Kafka's log directory
    at kafka.log.Log$.exception$1(Log.scala:1131)
    at kafka.log.Log$.parseTopicPartitionName(Log.scala:1139)
    at kafka.log.LogManager$$anonfun$loadLogs$2$$anonfun$3$$anonfun$apply$10$$anonfun$apply$1.apply$mcV$sp(LogManager.scala:153)
    at kafka.utils.CoreUtils$$anon$1.run(CoreUtils.scala:57)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
[2017-03-30 21:42:44,783] INFO [Kafka Server 2], shutting down (kafka.server.KafkaServer)
[2017-03-30 21:42:44,788] INFO Terminate ZkClient event thread. (org.I0Itec.zkclient.ZkEventThread)
[2017-03-30 21:42:44,820] INFO Session: 0x15b1f53fc120005 closed (org.apache.zookeeper.ZooKeeper)
[2017-03-30 21:42:44,821] INFO EventThread shut down for session: 0x15b1f53fc120005 (org.apache.zookeeper.ClientCnxn)
[2017-03-30 21:42:44,822] INFO [Kafka Server 2], shut down completed (kafka.server.KafkaServer)
[2017-03-30 21:42:44,823] FATAL Fatal error during KafkaServerStartable startup. Prepare to shutdown (kafka.server.KafkaServerStartable)
kafka.common.KafkaException: Found directory /home/zhangxs/datainfo/developmentData/kafka/log/log1, 'log1' is not in the form of topic-partition
If a directory does not contain Kafka topic data it should not exist in Kafka's log directory
    at kafka.log.Log$.exception$1(Log.scala:1131)

 

這個是我配置日志目錄有問題，我原目錄配置是[/home/zhangxs/datainfo/developmentData/kafka/log],他自動找到這個目錄的下一級目錄log1.   log1是我其他生產者消息的日志目錄，所以才拋出這個錯誤