[root@mini01 ~]# cat /etc/hosts

 127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4

 ::         localhost localhost.localdomain localhost6 localhost6.localdomain6

 172.16.1.11    mini01

 172.16.1.12    mini02

 172.16.1.13    mini03

 172.16.1.14    mini04

 172.16.1.15    mini05

 c:\windows\system32\drivers\etc

 ########################################## 追加信息如下：

 10.0.0.11    mini01

 10.0.0.12    mini02

 10.0.0.13    mini03

 10.0.0.14    mini04

 10.0.0.15    mini05

 [yun@mini01 software]$ pwd

 /app/software

 [yun@mini01 software]$ tar xf kafka_2.-2.0..tgz

 [yun@mini01 software]$ mv kafka_2.-2.0. /app/

 [yun@mini01 software]$ cd /app/

 [yun@mini01 ~]$ ln -s kafka_2.-2.0./ kafka

 [yun@mini01 ~]$ ll -d kafka*

 lrwxrwxrwx  yun yun  Sep  : kafka -> kafka_2.-2.0./

 drwxr-xr-x  yun yun  Jul  : kafka_2.-2.0.

 [root@mini01 ~]# tail /etc/profile

 done

 unset i

 unset -f pathmunge

 # kafka

 export KAFKA_HOME=/app/kafka

 export PATH=$KAFKA_HOME/bin:$PATH

 [root@mini01 profile.d]# logout

 [yun@mini01 hbase]$ source /etc/profile  # 使用yun用户，并重新加载环境变量

 [yun@mini01 config]$ pwd

 /app/kafka/config

 [yun@mini01 config]$ vim server.properties

 ############################# Server Basics #############################

 # 每一个broker在集群中的唯一标示★★★

 # 比如mini01 为0  mini02 为1  mini03 为2

 broker.id=

 ############################# Socket Server Settings #############################

 # The address the socket server listens on. It will get the value returned from

 # java.net.InetAddress.getCanonicalHostName() if not configured.

 #   FORMAT:

 #     listeners = listener_name://host_name:port

 #   EXAMPLE:

 #     listeners = PLAINTEXT://your.host.name:9092

 # 启动kafka服务监听的ip和端口，默认为java.net.InetAddress.getCanonicalHostName()获取的ip

 #listeners=PLAINTEXT://:9092

 # broker通知到producers和consumers的主机地址和端口号

 # 如果未设置，使用listeners的配置。否则，使用java.net.InetAddress.getCanonicalHostName()返回的值

 # 对于ipv4,基本就是localhost了 127.0.0.1   最后就是访问失败

 #advertised.listeners=PLAINTEXT://your.host.name:9092

 # 在不同的机器，名称不同  如mini01、mini02、mini03★★★

 advertised.listeners=PLAINTEXT://mini01:9092

 # broker 处理消息的最大线程数，一般情况下不需要去修改

 num.network.threads=

 # broker处理磁盘IO 的线程数 ，数值应该大于你的硬盘数

 num.io.threads=

 # 发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能

 socket.send.buffer.bytes=

 # kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘

 socket.receive.buffer.bytes=

 # 这个参数是向kafka请求消息或者向kafka发送消息的请求的最大数，这个值不能超过java的堆栈大小

 socket.request.max.bytes=

 ############################# Log Basics #############################

 # kafka数据的存放地址，多个地址的话用逗号分割 /tmp/kafka-logs-，/tmp/kafka-logs-

 log.dirs=/app/kafka/logs

 # 默认的分区数，一个topic默认1个分区数

 num.partitions=

 # 每个数据目录在启动时用于日志恢复和关闭时用于刷新的线程数。

 num.recovery.threads.per.data.dir=

 ############################# Internal Topic Settings  #############################

 # 组元数据内部主题“__consumer_offset”和“__transaction_state”的复制因子

 # 对于开发测试之外的任何内容，建议使用大于1的值来确保可用性，比如3。

 offsets.topic.replication.factor=

 transaction.state.log.replication.factor=

 transaction.state.log.min.isr=

 ############################# Log Flush Policy #############################

 # 在持久化到磁盘前message最大接收条数

 #log.flush.interval.messages=

 log.flush.interval.messages=

 # 持久化的最大时间间隔

 #log.flush.interval.ms=

 log.flush.interval.ms=

 ############################# Log Retention Policy #############################

 # 默认消息的最大持久化时间，168小时，7天

 # segment 文件保留的最长时间，超时则被删除

 log.retention.hours=

 # 当分片的大小超过该值时，就会被删除。该功能不依赖于log.retention.hours。为 -1没有大小限制

 #log.retention.bytes=

 log.retention.bytes=-

 # 滚动生成新的segment文件的最大时间

 log.roll.hours=

 # 单个分片的上限，达到该大小后会生成新的日志分片 1G

 log.segment.bytes=

 # 日志分片的检测时间间隔，每隔该时间会根据log保留策略决定是否删除log分片

 log.retention.check.interval.ms=

 # 默认为true  启用日志清理器进程在服务器上运行

 log.cleaner.enable=true

 # 默认为true 【当前版本】  启用删除主题。 如果此配置已关闭，则通过管理工具删除主题将不起作用

 delete.topic.enable=true

 ############################# Zookeeper #############################

 # Zookeeper connection string (see zookeeper docs for details).

 # This is a comma separated host:port pairs, each corresponding to a zk

 # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".

 # You can also append an optional chroot string to the urls to specify the

 # root directory for all kafka znodes.

 zookeeper.connect=mini01:,mini02:,mini03:,mini04:,mini05:

 # Timeout in ms for connecting to zookeeper

 zookeeper.connection.timeout.ms=

 ############################# Group Coordinator Settings #############################

 # 在开发测试环境下该值设置为0，保证启动后马上可以使用。但在生产环境下，默认值3秒更适合

 group.initial.rebalance.delay.ms=

 [yun@mini01 config]$ pwd

 /app/kafka/config

 [yun@mini01 config]$ vim producer.properties

 ………………

 # 用于建立到Kafka集群的初始连接的主机/端口对列表。客户机将使用所有服务器，而不管这里为引导绑定指定了哪些服务器——此列表只影响用于发现完整服务器集的初始主机。

 # 由于这些服务器仅用于初始连接，以发现完整的集群成员关系(可能会动态更改)，因此这个列表不需要包含完整的服务器集(但是，如果服务器宕机，您可能需要多个服务器)。

 bootstrap.servers=mini01:,mini02:,mini03:

 ………………

 [yun@mini01 config]$ vim consumer.properties

 ………………

 bootstrap.servers=mini01:,mini02:,mini03:

 ………………

 [yun@mini01 ~]$ kafka-server-start.sh -daemon /app/kafka/config/server.properties  # -daemon 可选参数，后台运行

 # 如果是后台运行，则控制台没有下面日志信息

 [-- ::,] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$)

 [-- ::,] INFO starting (kafka.server.KafkaServer)

 ……………………

 [yun@mini01 ~]$ kafka-server-stop.sh  # 停止kafka

 [zk: localhost:(CONNECTED) ] ls /

 [cluster, controller, brokers, zookeeper, admin, isr_change_notification, log_dir_event_notification, controller_epoch, zhang01, consumers, latest_producer_id_block, config, hbase]

 [zk: localhost:(CONNECTED) ] ls /brokers

 [ids, topics, seqid]

 [zk: localhost:(CONNECTED) ] ls /brokers/ids

 [, , ]