kafka的使用

zookeeper相关

​ zookeeper的安装，查看文档：zookeeper的使用

​ 在kafka中，zookeeper主要存kafka节点的数据。

​ 查看znode信息，可以看到kafka的 broker, topic等信息

​ 可以使用zookeeper命令，用来模拟创建topic，在kafka使用 kafka-topics.sh 也是能查到数据的。

​ 所以证明，kafka的许多信息就是使用zookeeper来存取的。

安装kafka

​ 直接下载，解压就行。

kafka的一些特性

1. 高吞吐量、低延迟；
2. 可扩展性；
3. 持久性、可靠性；
4. 容错性；
5. 高并发；
6. 支持实时在线处理和离线处理。

kafka的使用场景

官方文档显示如下几种场景可以使用：

1. 网站活动追踪

   用户的活动追踪，网站的活动（网页游览，搜索或其他用户的操作信息）发布到不同的话题中心，这些消息可实时处理，实时监测，也可加载到Hadoop或离线处理数据仓库

2. 指标

   分布式应用程序生成的统计数据集中聚合

3. 日志聚合

   使用kafka代替一个日志聚合的解决方案。

4. 流处理

   kafka消息处理包含多个阶段。其中原始输入数据是从kafka主题消费的，然后汇总，丰富，或者以其他的方式处理转化为新主题

5. 事件采集

   事件采集是一种应用程序的设计风格，其中状态的变化根据时间的顺序记录下来，kafka支持这种非常大的存储日志数据的场景。

6. 提交日志

   kafka可以作为一种分布式的外部提交日志，日志帮助节点之间复制数据，并作为失败的节点来恢复数据重新同步。

kafka的一些基本概念

Topic（主题）

​ 每一类的消息，称之为一个主题。

​ 一个主题能分为多个partition，每个partition对应一个文件夹，每一个消息发送到Broker时，会根据partition规则选择存储到哪一个partition。

Producer（生产者）

​ 发布消息的对象。

producer将会与topic所有的partition leader保持socker连接；

消息由producer直接通过socket发送到broker，中间不会经过任何路由层；

事实上，消息被路由到哪个partition由producer决定

Consumer（消费者）

​ 订阅消息并处理发布消息。

​ kafka稳定状态下，每一个consumer实例只会消费某一个或多个特定的partition的数据。consumer与partition的数目有以下三种对应关系：

1. consumer < partition ， 至少有一个consumer会消费多个partition的数据；

2. consumer = partition ， 正好有一个consumer消费一个partition数据；

3. consumer > parition, 会有部分consumer无法消费该topic下任何一条消息。

   kafka对于消费消息：

4. 不删除已消费的消息；

   1. 基于时间
   2. 基于partition文件大小

5. 保证同一个consumer group只有一个consumer会消费一条消息；

6. 允许不同consumer group同时消费一条消息。

consumer端向broker发送"fetch"请求，并告知其获取消息的offset。正常情况下，会在消费完一条消息后线性增加这个offset。

Broker （代理）

​ 已发布的消息保存在一组服务器中，称之为kafka集群。集群中的每一个服务器都是一个代理(broker)。

kafka的目录相关

• /bin 操作kafka的可执行脚本， 还包含windows下脚本；
• /config 配置文件所在目录
• /libs 依赖库目录
• /logs 日志数据目录，目录kafka把server端日志分为：server, request, state, log-cleaner, controller.

kafka命令行下的使用

我们可以先用kafka提供的命令行工具，来熟悉kafka的基本使用。

0、最基本的命令

​ 开启kafka服务。

​ 在开启kafka服务之前，先开启zookeeper服务。

/path/to/zookeeper/bin/zkServer.sh start #默认访问的配置文件../conf/zoo.cfg
/path/to/kafka/bin/kafka-server-start.sh -daemon ../config/server.properties
jps #查看是否开启服务
18723 Jps
1972 ZooKeeperMain #zookeeper
4549 Kafka  #kafka

1、创建一个主题

(learn) [root@localhost bin]# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test2
Created topic "test2".

2、获取最近创建的主题列表

[root@localhost bin]# ./kafka-topics.sh --list --zookeeper localhost:2181 
__consumer_offsets
test
test2   #刚才创建的主题

3、查看主题详细信息

[root@localhost bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test2
Topic:test2 PartitionCount:1    ReplicationFactor:1 Configs:
    Topic: test2    Partition: 0    Leader: none    Replicas: 1001  Isr: 

这里发现Leader竟然显示为None,这是一个错误！！！ 以下开始排错

排错的博文

也即是说，我们kafka存到zookeeper的信息出错； 问题的原因，是因为我之前将zookeeper的datadir目录下的所有文件都删除，导致错误的出现。

[zk: localhost:2181(CONNECTED) 0] ls /    
[cluster, controller_epoch, controller, brokers, zookeeper, admin, isr_change_notification, consumers, log_dir_event_notification, latest_producer_id_block, config]
[zk: localhost:2181(CONNECTED) 2] ls /brokers
[ids, topics, seqid]
[zk: localhost:2181(CONNECTED) 3] ls /brokers/topics
[test2, test, __consumer_offsets]
# 这里可以看出，我们test2下没有任何信息
[zk: localhost:2181(CONNECTED) 4] ls /brokers/topics/test2
[]
# test2节点存的信息如下
[zk: localhost:2181(CONNECTED) 1] get /brokers/topics/test2
{"version":1,"partitions":{"0":[1001]}}
cZxid = 0x104
ctime = Sat Dec 29 14:28:38 CST 2018
mZxid = 0x104
mtime = Sat Dec 29 14:28:38 CST 2018
pZxid = 0x10d
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 39
numChildren = 1
#然后我们开始创建test2的分区父节点
[zk: localhost:2181(CONNECTED) 5] create /brokers/topics/test2/partitions null 
Created /brokers/topics/test2/partitions
#然后创建分区0的节点
[zk: localhost:2181(CONNECTED) 6] create /brokers/topics/test2/partitions/0 null
Created /brokers/topics/test2/partitions/0
#创建0分区的状态节点
[zk: localhost:2181(CONNECTED) 8] create /brokers/topics/test2/partitions/0/state
#设置节点信息
[zk: localhost:2181(CONNECTED) 6] set /brokers/topics/test2/partitions/0/state {"controller_epoch":2,"leader":0,"version":1,"leader_epoch":0,"isr":[1001]}

以上操作完成之后。我们再一次查看 topic信息

[root@localhost bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test2
Topic:test2 PartitionCount:1    ReplicationFactor:1 Configs:
    Topic: test2    Partition: 0    Leader: 0   Replicas: 1001  Isr: 1001

这样我们就能正确的获得到topic的详细信息了。

4、发送消息

[root@localhost bin]# ./kafka-console-producer.sh --broker-list localhost:9092,localhost:9093 --topic test2
>aaa
>bbb
>ccc

5、接收消息

[root@localhost bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092,localhost:9093 --topic test2
aaa
bbb
ccc

6、设置多个broker集群

​ 上面我就是开启了两个伪Broker集群，也就是一台机器上开了2个代理。

​ server-0.properties ，其他的配置不做修改

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0
############################# 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
listeners=PLAINTEXT://:9093

# Hostname and port the broker will advertise to producers and consumers. If not set, 
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners=PLAINTEXT://192.168.11.120:9093

​ server.properties

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1

############################# 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
listeners=PLAINTEXT://:9092

# Hostname and port the broker will advertise to producers and consumers. If not set, 
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners=PLAINTEXT://192.168.11.120:9092

7、使用kafka Connect 来导入/导出数据

> bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

​ 主要就是修改connect-file-source.properties文件中file的路径名（全路径）

​ 还有修改config/connect-file-sink.properties 导出的目标路径名

使用一下命令查看主题:

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic connect-test --from-beginning

8、使用Kafka Stream来处理数据

步骤如下：

• 运行中间处理数据的java库

./bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo

• 创建topic

bin/kafka-topics.sh --create \
            --zookeeper localhost:2181 \
            --replication-factor 1 \
            --partitions 1 \
            --topic streams-file-input

（暂时没试验）

上面的命令有一些看上去没有什么规律，有的地方用zookeeper， 用的地方用broker-list， 有的地方用bootstrap-server， 我们只需要记住，如果要查询信息，那么使用的就是zookeeper，因为kafka的信息都在zookeeper上，发送命令用broker-list， 接收用bootstrap-server。 在代码中使用大多都是使用bootstrap-server的方式。

9、如何清空zookeeper与kafka测试数据

1、步骤先关闭zookeeper,kafka;

2、查看zookeeper配置文件中 datadir的具体位置，然后删除version-2/ ；

3、删除zookeeper的log文件

4、 删除kafka的log文件。

Confluent-kafka-python的使用

​ 常用的三个库有kafka-python， pykafka, confluent-kafka-python，我选择测试的库是confluent-kafka-python，kafka推荐使用该库。

​ confluent-kafka-python是基于librdkafka的高性能python客户端，具有完整的协议支持。

主要测试的api分为以下三个：

Producer

​ 具体测试代码如下：

from collections import defaultdict
from confluent_kafka import Producer
# 初始化Producer
p = Producer({'bootstrap.servers': '192.168.11.120:9092,192.168.11.120:9093'})

parition_count = defaultdict(int)
def delivery_report(err, msg):
    if err is not None:
        print("Message delivery failed: {}".format(err))
    else:
        parition_count[msg.partition()] += 1
        # print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition()))
count = 100
while count > 0:
    for data in range(1000):
        p.poll(0)
        p.produce('test', str(data).encode('utf-8'), callback=delivery_report)
    count -= 1

p.flush()

编写过程如下：

• 初始化Producer，在初始化中，必须提供一个字典，其中必须包括一个'bootstrap-server'的键值。(在测试的过程中，我想知道producer的其他参数如何设置，在其给出的example中没有给出具体用例，所以这里提供一个网站用来查询配置：配置项)
• 创建一个回调函数，用来处理消息发布后的流程
• poll函数用来轮询生产者的事件，参数为time，单位seconds， 表示事件最大等待时间。
• produce函数，就是用来给topic来生产消息的主要函数，函数完整形式如下：

produce(topic, [value], [key], [partition], [on_delivery], [timestamp], [headers])

• flush函数，强刷所有的缓冲发送到brokers

其他的函数还有：

list_topics（）这个函数可以用来获取broker的很多详细信息。比如列举topic,查询分区信息等。

Consumer

from collections import defaultdict
from confluent_kafka import Consumer, KafkaError,TopicPartition

c = Consumer({
    'bootstrap.servers': '192.168.11.120:9092, 192.168.11.120:9093',
    'group.id': 'test-consumer-group',
    'auto.offset.reset': 'latest',
    'fetch.wait.max.ms': 5000
})
partition_count = defaultdict(int)
c.subscribe(["test"])
while True:
    msg = c.poll(0.1)
    if msg is None:
        continue
    if msg.error():
        if msg.error().code() == -191:
            print('EOF')
            partition_count = defaultdict(int)
            continue
        print("Consumer error: {}".format(msg.error()))
        continue
    partition_count[msg.partition()] += 1
    print('Received message: {} partition: {}'.format(msg.value().decode('utf-8'), msg.partition()))
    print(partition_count)

c.close()

编写过程如下：

• 新建Consumer客户端，配置与上面Producer相似；
• 使用subscribe函数，来订阅topic消息，如果开启多个分区，那么就会自动负载。 这里还可以使用另一个函数assign()，这个函数可以用来订阅某一topic的固定的一个分区。
• 以下就是循环获取数据。

Admin

​ Kafka AdminClient为代理支持的Kafka代理，主题，组和其他资源类型提供管理操作。这里使用这个函数的目的，就是用来管理broker。

​ 在生产环境下，我建议关闭自动创建topic这有个配置选项，理由，因为如果我们producer代码与consumer代码topic不一致，那么我们排查起来很麻烦。 所以，建议还是使用特定的工具去干特定的事情。

​ 具体的方法如下：

alter_config:

create_partitions:

create_topics:

delete_topics:

describe_configs：

（暂时没有使用，打算在封装库之中使用）