伪分布式：

1、先将zk启动，如果是在伪分布式下，kafka已经集成了zk

nohup /kafka_2.11-0.10.0.1/bin/zookeeper-server-start.sh /kafka_2.11-0.10.0.1/config/zookeeper.properties &

 

2、启动broker

nohup /kafka_2.11-0.10.0.1/bin/kafka-server-start.sh /kafka_2.11-0.10.0.1/config/server.properties &

3、测试

   模拟一个生产者，向另一个消费者发送数据

   但是消费是必须属于某一个主题

   3.1先得创建一个主题

  /kafka_2.11-0.10.0.1/bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic mytopic --partitions 1 --replication-factor 1

 

   --topic:主题的名称

   --partitions：主题下有几个分区

   --replication-factor：复本数

   3.2生产者的创建

   /kafka_2.11-0.10.0.1/bin/kafka-console-producer.sh --topic mytopic --broker-list localhost:9092

   3.3消费者的创建

  /kafka_2.11-0.10.0.1/bin/kafka-console-consumer.sh  --topic mytopic --zookeeper localhost:2181

 

kafka集群配置：

1、事先准备好zookeeper

2、导入kafka的包

3、配置环境变量

4、进入kafka的config下修改server.properties

    master的是broker.id=0

       log.dirs修改地址，之前是在/tmp目录下，我的修改为/yang/kafka-logs

    zookeeper.connect=自己的zk集群地址   为：matser,slave01,slave02

5、将/etc/profile分别发送到slave01,slave02并source

    将kafka_2.11-0.10.0.1  发送到slave01,slave02

    并将slave01的kafka的config下修改server.properties    broker.id=1

    并将slave02的kafka的config下修改server.properties    broker.id=2

 

6、启动

    1、zkServer.sh start三台

    2、nohup kafka-server-start.sh /kafka_2.11-0.10.0.1/config/server.properties &

7、创建一个名为mytopic的主题  分区数为3  副本因子为3

kafka-topics.sh --create --zookeeper master,slave01,slave02 --topic mytest --partitions 3 --replication-factor 3

查看kafka-topics.sh --describe --zookeeper master,slave01,slave02 --topic mytopic

生产者创建：kafka-console-producer.sh --topic mytopic --broker-list localhost:9092

消费者创建：kafka-console-consumer.sh --zookeeper master,slave01,slave02 --topic mytopic