安装Kafka集群

本文将介绍如何安装消息队列系统，Kafka集群：

1 安装Java

yum install -y java-1.8.0-openjdk-devel 

2 安装Zookeeper

下载、安装、启动Zookeeper

wget https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.4.11/zookeeper-3.4.11.tar.gz
tar vxf zookeeper-3.4.11.tar.gz
mv zookeeper-3.4.11 zookeeper
cd zookeeper/conf
mv zoo_sample.cfg zoo.cfg
cd ../bin
./zkServer.sh start

查看Zookeeper的端口2181已经打开：

3 安装Kafka

下载Kafka

wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/1.1.0/kafka_2.12-1.1.0.tgz
tar vxf kafka_2.12-1.1.0.tgz
mv kafka_2.12-1.1.0 kafka

增加hosts文件内容：

echo  "172.17.0.4 hwkafka01" >> /etc/hosts
echo  "172.17.0.5 hwkafka02" >> /etc/hosts

配置Kafka：

cd kafka/config
vim server.properties

broker.id=2
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://x.x.x.x:9020 #这里必须写公网IP
zookeeper.connect=hwkafka01:2181,hwkafka02:2181

启动Kafka：

bin/kafka-server-start.sh config/server.properties &

4 创建topic

./kafka-topics.sh --create --zookeeper hwkafka01:2181 --replication-factor 1 --partitions 1 --topic test

验证：

./kafka-topics.sh --list --zookeeper hwkafka01:2181
OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N

Test

5 用python作为producer测试

在server端开启consumer客户端：

./kafka-console-consumer.sh --bootstrap-server hwkafka01:9020 --topic test

安装pykafka

pip install pykafka

代码exmple：

import pykafka
from pykafka import KafkaClient
client = KafkaClient(hosts="139.219.108.52:9092",ssl_config=None)
print(client.topics)
topic = client.topics[b'test']
producer = topic.get_producer()
producer.produce(b'new message')

可以看到在consumer的窗口出现信息：