rom 1.5 pos listen add eight ati 接收 必須

1、zookeeper搭建

　　Kafka集群依賴zookeeper，需要提前搭建好zookeeper

　　單機模式（7步）(集群模式進階請移步：http://blog.51cto.com/nileader/795230)

Step1：

cd /usr/local/software 
wget http://download.oracle.com/otn-pub/java/jdk/8u161-b12/2f38c3b165be4555a1fa6e98c45e0808/jdk-8u161-linux-x64.tar.gz

配置JAVA環境。檢驗方法：執行java –version和javac –version命令。

Step2：下載並解壓zookeeper。

鏈接：http://mirror.bjtu.edu.cn/apache/zookeeper/zookeeper-3.4.3/，（更多版本：http://dwz.cn/37HGI

）最終生成目錄類似結構：/home/admin/taokeeper/zookeeper-3.4.3/bin

Step3：重命名 zoo_sample.cfg文件

 mv /home/admin/taokeeper/zookeeper-3.4.3/conf/zoo_sample.cfg  zoo.cfg

Step4：vi zoo.cfg，修改

dataDir=/home/admin/taokeeper/zookeeper-3.4.3/data

Step5：創建數據目錄：mkdir /home/admin/taokeeper/zookeeper-3.4.3/data

1. mkdir /home/admin/taokeeper/zookeeper-3.4.3/data

Step6：啟動zookeeper：執行

/home/admin/taokeeper/zookeeper-3.4.3/bin/zkServer.sh start

Step7：檢測是否成功啟動：執行

/home/admin/taokeeper/zookeeper-3.4.3/bin/zkCli.sh 或 echo stat|nc localhost 2181

2、下載Kafka

下載地址：http://kafka.apache.org/downloads.html

# mkdir -p /usr/local/software
# cd /usr/local/software
# wget http://mirror.bit.edu.cn/apache/kafka/1.0.0/kafka-1.0.0-src.tgz
# wget http://mirror.bit.edu.cn/apache/kafka/1.0.0/kafka_2.12-1.0.0.tgz

kafka_2.12-1.0.0.tgz //其中2.12-1.0.0為Scala的版本，kafka-1.0.0-src.tgz為kafka版本 3、解壓

# tar zxf kafka_2.12-1.0.0.tgz -C /usr/local/
# cd /usr/local/
# mv kafka_2.12-1.0.0/ kafka/

4、配置

# vi /usr/local/kafka/config/server.properties

# broker的ID，集群中每個broker ID不可相同
broker.id=0
# 監聽器，端口號和port一致即可
listeners=PLAINTEXT://:9092
# Broker的監聽端口
port=9092
# 必須填寫當前服務器IP地址
host.name=10.10.6.225
# 必須填寫當前服務器IP地址
advertised.host.name=10.10.6.225
# 暫未配置集群
#zookeeper.connect=zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181

5、配置Kafka的環境變量

# vi /etc/profile
　　export KAFKA_HOME=/usr/local/kafka
　　export PATH=$PATH:$KAFKA_HOME/bin
# source /etc/profile

6、啟動Kafka

# kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

　　官方推薦啟動方式：

# /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &

但這種方式退出shell後會自動斷開

7、驗證

# jps
    2256 Jps
    2237 Kafka

看到Kafka的進程，說明Kafka已經啟動 8、創建topic 創建名為test，partitions為3，replication為3的topic

# kafka-topics.sh --create --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --partitions 3 --replication-factor 3 --topic test

查看topic狀態

# kafka-topics.sh --describe --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --topic test
　　Topic:test PartitionCount:3 ReplicationFactor:3 Configs:
　　Topic: test Partition: 0 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2
　　Topic: test Partition: 1 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
　　Topic: test Partition: 2 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1

　　狀態說明：test有三個分區分別為1、2、3，分區0的leader是3（broker.id），分區0有三個副本，並且狀態都為lsr（ln-sync，表示可以參加選舉成為leader）。

刪除topic 在config/server.properties中加入delete.topic.enable=true並重啟服務，在執行如下命令

# kafka-topics.sh --delete --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --topic test

9、測試使用Kafka 發送消息

# kafka-console-producer.sh --broker-list zookeeper-01:9092,zookeeper-02:9092,zookeeper-03:9092 --topic test
輸入以下信息：
　　This is a message
　　This is another message

接收消息

# kafka-console-consumer.sh --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --topic test --from-beginning    //--from-beginning 表示從開始第一個消息開始接收

若看到上輸入的信息說明已經搭建成功。

Kafka單機搭建