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是我其他生产者消息的日志目录，所以才抛出这个错误