kafka安装使用

使用kafka的前提是安装好了 jdk 和 zookeeper

下载解压kafka：

[root@localhost ~]# wget http://mirror.bit.edu.cn/apache/kafka/2.3.0/kafka_2.12-2.3.0.tgz
[root@localhost ~]# tar -xzf kafka_2.12-2.3.0.tgz
[root@localhost ~]# mv kafka_2.12-2.3.0 kafka
[root@localhost ~]# cd kafka

启动kafka之前，需要先启动zookeeper。

单机部署模式：
单机模式下可以不动配置，直接启动kafka服务

[root@localhost kafka]# nohup bin/kafka-server-start.sh config/server.properties &
[1] 36113
[root@localhost kafka]# nohup: 忽略输入并把输出追加到"nohup.out"

测试：

===========================创建topic=========================
[root@localhost kafka]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic HmcfTest
Created topic HmcfTest.

--topic 定义 topic 名
--replication-factor  定义副本数
--partitions  定义分区数


===========================删除topic=========================
bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic topicName 
需要 server.properties 中设置 delete.topic.enable=true 否则只是标记删除或者直接重启（最好不要设置）


=========================查看topic列表========================
[root@localhost kafka]# bin/kafka-topics.sh --list --zookeeper localhost:2181
HmcfTest
[root@localhost kafka]#


=========================向topic发送消息========================
[root@localhost kafka]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic HmcfTest
>hello kafka
>there is a pythoner or goer


========================查看topic对象信息========================
[root@localhost kafka]# bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic HmcfTest
Topic:HmcfTest  PartitionCount:1        ReplicationFactor:1     Configs:
        Topic: HmcfTest Partition: 0    Leader: 0       Replicas: 0     Isr: 0


=============================消费消息============================
[root@localhost kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic HmcfTest --from-beginning
hello kafka
there is a pythoner or goer
souga

===========================停止kafka服务=========================
[root@localhost kafka]# bin/kafka-server-stop.sh

集群部署模式
对于Kafka来说，一个单独的broker就是一个大小为1的集群，所以集群模式无非多启动几个broker实例。

在多机器下需要修改/etc/hosts文件，将用于kafka集群的机器配置上去。
如：

[root@localhost kafka]# cat /etc/hosts
127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6

192.168.64.11 hmcf-01
192.168.64.12 hmcf-02
192.168.64.13 hmcf-03

然后修改对应的server.properties配置文件
其中一些配置参数可参考网络说明。
注意踩坑
配置文件中的advertised.listeners参数，需要写成IP，写主机名的话会导致外网无法连接，如python脚本无法连接使用。

# server-0
broker.id=0
listeners=PLAINTEXT://192.168.64.11:9092
advertised.listeners=PLAINTEXT://192.168.64.11:9092
log.dirs=/tmp/kafka-logs
num.partitions=5
zookeeper.connect=localhost:2181,192.168.64.12:2182,192.168.64.13:2181

# server-1
broker.id=1
listeners=PLAINTEXT://192.168.64.12:9092
advertised.listeners=PLAINTEXT://192.168.64.12:9092
log.dirs=/tmp/kafka-logs
num.partitions=5
zookeeper.connect=localhost:2181,192.168.64.12:2182,192.168.64.13:2181

# server-2
broker.id=2
listeners=PLAINTEXT://192.168.64.13:9092
advertised.listeners=PLAINTEXT://192.168.64.13:9092
log.dirs=/tmp/kafka-logs
num.partitions=5
zookeeper.connect=localhost:2181,192.168.64.12:2182,192.168.64.13:2181

最后启动三个broker

[root@localhost kafka]# nohup bin/kafka-server-start.sh config/server.properties &

使用jps命令可以查看到kafka进程

[root@localhost kafka]# jps
1428 QuorumPeerMain
5333 Jps
4959 Kafka

====================鸡儿分割==============================

创建集群topic

 bin/kafka-topics.sh --create --zookeeper hmcf-01:2181, hmcf-02:2181, hmcf-03:2181 --replication-factor 3 --partitions 3 --topic hmcf_test

查看集群topic对象信息

[root@localhost kafka]# bin/kafka-topics.sh --describe --zookeeper hmcf-01:2181, hmcf-02:2181, hmcf-03:2181 --topic hmcf_test
Topic:hmcf_test PartitionCount:3        ReplicationFactor:3     Configs:
        Topic: hmcf_test        Partition: 0    Leader: 2       Replicas: 2,1,0 Isr: 2,1,0
        Topic: hmcf_test        Partition: 1    Leader: 0       Replicas: 0,2,1 Isr: 0,2,1
        Topic: hmcf_test        Partition: 2    Leader: 1       Replicas: 1,0,2 Isr: 1,0,2

生产与消费

[root@localhost kafka]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic hmcf_test 
>new girl
>boduoyejiyi
>over

[root@localhost kafka]# bin/kafka-console-consumer.sh --bootstrap-server hmcf-01:9092, hmcf-02:9092, hmcf-03:9092 --topic hmcf_test --from-beginning
new girl
boduoyejiyi
over
^CProcessed a total of 3 messages

指定partition 和 指针位置
[root@localhost kafka]# bin/kafka-console-consumer.sh --bootstrap-server hmcf-01:9092, hmcf-02:9092, hmcf-03:9092 --topic hmcf_test --partition 0  --offset 2
over

python kafka测试

生产者

import json
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092')

msg_dict = {
    "sleep_time": 10,
    "db_config": {
        "database": "test_1",
        "host": "xxxx",
        "user": "root",
        "password": "root"
    },
    "table": "msg",
    "msg": "Hello World"
}
msg = json.dumps(msg_dict)
producer.send('HmcfTest', msg, partition=0)
producer.close()

==============================================================
消费者

from kafka import KafkaConsumer

consumer = KafkaConsumer('HmcfTest', bootstrap_servers=['192.168.64.11:9092'])
for msg in consumer:
    recv = "%s:%d:%d: key=%s value=%s" %
           (msg.topic, msg.partition, msg.offset, msg.key, msg.value)
    print(recv)