Kafka基本使用

使用docker部署kafka

基礎環境

操作系統	硬件配置
CentOS 7 Server	磁盤:40GB 內存:8GB 網卡:ens3(外網)

網絡配置

# vim ifcfg-ens3
TYPE=Ethernet
BOOTPROTO=static
DEFROUTE=yes
PEERDNS=yes
PEERROUTES=yes
NAME=ens3
DEVICE=ens3
ONBOOT=yes
IPADDR=192.168.200.13
NETMASK=255.255.255.0
GATEWAY=192.168.200.1
DNS1=114.114.114.114

基礎環境配置

安裝PIP

# yum install epel-release
# yum install python-pip
# pip install -U pip

安裝PIP編譯環境

# yum install python-devel libffi-devel gcc openssl-devel

安裝docker

# curl -sSL http://acs-public-mirror.oss-cn-hangzhou.aliyuncs.com/docker-engine/internet | sh -
# docker --version

配置docker環境

# mkdir -p /etc/systemd/system/docker.service.d
# tee /etc/systemd/system/docker.service.d/kolla.conf <<-'EOF'
[Service]
MountFlags=shared
EOF

### 配置鏡像加速
# mkdir -p /etc/docker
# tee /etc/docker/daemon.json <<-'EOF'
{
"registry-mirrors":  ["https://ar4vj99q.mirror.aliyuncs.com"]
}
EOF
# systemctl daemon-reload
# systemctl restart docker
# systemctl enable docker

安裝docker-compose

# pip install docker-compose

部署Kafka

下載docker-compose文件

# git clone https://github.com/wurstmeister/kafka-docker.git
# cd kafka-docker
# vim docker-compose.yml
      KAFKA_ADVERTISED_HOST_NAME: 192.168.200.13
### 啟動kafka
# docker-compose up
### 列舉broker
# ./broker-list.sh
:32768

### 如果希望創建多個broker，在執行docker-compose up后執行
# docker-compose scale kafka=3

測試安裝

下載kafka

### 測試主機IP： 192.168.200.11
# wget http://mirror.bit.edu.cn/apache/kafka/0.10.2.1/kafka_2.12-0.10.2.1.tgz -O kafka_2.12-0.10.2.1.tgz
# tar zxf kafka_2.12-0.10.2.1.tgz
# yum install java -y
# cd kafka_2.12-0.10.2.1

創建主題

# bin/kafka-topics.sh --create --zookeeper 192.168.200.13:2181 --replication-factor 1 --partitions 1 --topic fwd-test
# bin/kafka-topics.sh --list --zookeeper 192.168.200.13:2181

### 或者在192.168.200.13執行
# yum install net-tools
# cd kafka-docker
# ./create-topics.sh -n test

測試生產者

### 執行命令后隨便輸入寫字符串並回車結束
# bin/kafka-console-producer.sh --broker-list 192.168.200.13:32768 --topic fwd-test

測試消費者

# bin/kafka-console-consumer.sh --bootstrap-server 192.168.200.13:32768 --topic fwd-test --from-beginning

手工部署kafka

安裝Kafka

### 在Sched和Nova節點上執行
# wget http://mirror.bit.edu.cn/apache/kafka/0.10.2.1/kafka_2.12-0.10.2.1.tgz -O kafka_2.12-0.10.2.1.tgz
# tar zxf kafka_2.12-0.10.2.1.tgz
# yum install java -y
# systemctl disable firewalld
# systemctl stop firewalld

啟動Zookeeper

### 在Sched節點上執行
# bin/zookeeper-server-start.sh config/zookeeper.properties &

啟動kafka broker

### 在Sched節點上執行
# vim config/server.properties
advertised.listeners=PLAINTEXT://192.168.200.11:9092
# bin/kafka-server-start.sh config/server.properties &

創建topic

### 在Sched節點上執行
# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic rsyslog-prod

查看創建的topic

### 在Sched節點上執行
# bin/kafka-topics.sh --list --zookeeper localhost:2181

### 在Nova節點上執行
# bin/kafka-topics.sh --list --zookeeper 192.168.200.11:2181

測試producer

### 在Sched節點上執行，並輸入些字符串
# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic rsyslog-prod
this server

### 在Nova節點上執行，並輸入些字符串
# bin/kafka-console-producer.sh --broker-list 192.168.200.11:9092 --topic rsyslog-prod
this client

測試consumer

### ### 在Sched節點上執行，如果配置無異常則正常顯示如下內容
# bin/kafka-console-consumer.sh --bootstrap-server 192.168.200.11:9092 --topic rsyslog-prod --from-beginning
this server
this client

問題處理

kafka-server啟動報錯

ERROR Processor got uncaught exception. (kafka.network.Processor)
java.lang.NoClassDefFoundError: Could not initialize class kafka.network.RequestChannel$
    at kafka.network.RequestChannel$Request.<init>(RequestChannel.scala:93)
    at kafka.network.Processor$$anonfun$run$11.apply(SocketServer.scala:426)
    at kafka.network.Processor$$anonfun$run$11.apply(SocketServer.scala:421)
    at scala.collection.Iterator$class.foreach(Iterator.scala:742)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1194)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
    at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
    at kafka.network.Processor.run(SocketServer.scala:421)
    at java.lang.Thread.run(Thread.java:745)

### 解決辦法，將hostname設置進/etc/hosts
# ip addr show
# hostname
# 

參考文檔

Landoop GitHub

Confluent官方文檔