分布式发布订阅消息系统Kafka

Kafka概述

KafKa是一个高吞吐量、分布式的发布——订阅消息系统。据KafKa官网介绍，当前的KafKa已经定位为一个分布式流式处理平台（a distributed streaming platform），它以可水平扩展和具有高吞吐量等特性而著称。越来越多的开源分布式处理系统（Flume、Apache Storm 、Spark、Flink等）支持与KafKa集成。
Kafka能够很好的满足以下三个特性：

• 能够允许发布和订阅流数据。从这个角度来讲，平台更像是一个消息队列或者企业级的消息系统；
• 存储流数据的时候提供相应的容错机制；
• 当流数据达到的时候能够即时的被处理；

Kafka架构和核心概念

image.png

• Topic：特指kafka处理的消息源
• Partition（分区）：Topic物理上的分组。一个Topic可以有多个Patition，每个Partition是一个有序的队列
• Message：消息，通信的基本单位
• Producer：生产者。向kafka的一个topic发布消息的过程叫做生产
• Consumer：消费者，订阅Topic并处理其发布的消息的过程叫做消费
• Broker：缓存代理，kafka集群中的一台或者多台服务器

Kafka部署和使用

Kafka需要用到zookeeper。所以在安装之前需要先安装zookeeper。

//下载并解压zookeeper的安装包到自定义目录
tar -zxvf zookeeper-3.4.5-cdh5.7.0.tar.gz -C ../apps/
// 配置环境变量并生效
vi ~/.bash_profile
              export ZK_HOME=/root/apps/zookeeper-3.4.5-cdh5.7.0
              export PATH=$ZK_HOME/bin:$PATH
source ~/.bash_profile
//进入conf目录，配置zoo.cfg文件
cp zoo_sample.cfg zoo.cfg
mkdir -p  ~/data/tmp/zk  // 创建数据目录
vi zoo.cfg
       dataDir=/root/data/tmp/zk/   //默认在tmp目录下，在关机重启后数据会丢失，这里做持久化保存配置
//进入bin目录，启动service
./zkServer.sh start
//验证
jps
      QuorumPeerMain //有zookeeper进程表示启动成功

单节点单Broker部署和使用

• 首先安装Kafka，我版本选用的是2.2.0

1)解压，配置系统环境变量 
tar -zxvf kafka_2.11-2.2.0.tgz -C ~/apps/
vi ~/.bash_profile
          export KAFKA_HOME=/root/apps/kafka_2.11-2.2.0
          export PATH=$KAFKA_HOME/bin:$PATH
source ~/. bash_profile
2)$KAFKA_HOME/config中配置文件server.properties中的参数说明： 
①broker.id : 每个broker编号，篮子编号不能冲突，只有一台机器，默认从0开始 
②监听端口默认在9092上面 
③hostname:当前机器
④log.dirs:用来存储kafka日志的目录   //默认为tmp目录，重启后会丢失，需要改为自定义目录
⑤num.partition：分区的数量，默认为1 
⑥zookeeper.connection: zookeeper地址

• 启动kafka(启动之前先要启动zookeeper)

kafka-server-start.sh $KAFKA_HOME/config/server.properties
注：如果只输入kafka-server-start.sh 
会出现提示信息：USAGE: /home/hadoop/app/kafka_2.11-0.9.0.0/bin/kafka-server-start.sh [-daemon] server.properties [–override property=value]* 
-daemon：表示是否以后台的形式运行 

• 启动后用jps可以看到kafka，用jps -m可以看到kafka对应哪个server.properties文件
• 创建topic：名称test_hi,执行命令需要到bin目录。

./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test_hi

创建topic需要指定zookeeper地址，副本系数为1，一个分区一个副本系数

• 查看目前有多少个topic，所有topic

./kafka-topics.sh --list --zookeeper localhost:2181

• 发送消息

./kafka-console-producer.sh --broker-list localhost:9092 --topic test_hi

bin目录下自带了一个控制台的生产者，要用broker-list,9092是前面配置的监听端口

• 消费消息

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_hi --from-beginning

• 测试
  可以开两个命令行终端，分别执行生产和消费消息命令，输入即可实时显示，在生产者终端中键入消息并看到它们出现在消费者终端中。默认情况下，每行作为单独一个消息发送。可以发现，消费终端关闭再开启，消息不丢失，这是因为用了--from-beginning，说明从头开始消费。如果不加这个参数，只从当前开始消费。
• 补充：
  1)./kafka-topics.sh --describe --zookeeper localhost:2181
  可以看到所有topic的详细信息
  2)./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test_hi
  可以查看指定topic的信息

单节点多Broker部署和使用

• 先为每个broker配置一个server.properties，主要修改其中的broker.id、监听端口号和日志输出目录，因为我们是在一台机器上运行。
• 启动zk
• 启动kafka，现在以后台的方式启动，这样终端关了也没有关系

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

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

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

4）启动后，用jps可以看到三个kafka，jps -m可以看到分别对应的配置文件
5）创建topic

kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-topic

3个broker对应3个副本

• 查看topic及详细信息同上
  其中，Leader指的是主broker的id是3，replicas表示3个broker列表，Isr表示当前存活的brokerId
• 发送消息

kafka-console-producer.sh --broker-list localhost:9093,localhost:9094,localhost:9095 --topic my-topic

3个broker端口号与配置文件中一致

• 消费信息

./kafka-console-consumer.sh --bootstrap-server localhost:9093,localhost:9094,localhost:9095 --topic my-topic

现在可以启动三个终端输入命令，实时接受信息

多节点多Broker部署和使用

和单节点多Broker部署类似。这里由于机器原因不做演示。

Kafka容错性测试

在多Broker案例中，我们随便杀掉一个broker进程，在生成者中发送消息，发现消费者还是可以正常接收，查看topic信息，发现活得节点还有两个，所以说在多Broker的环境中具有容错性。

image.png

Kafka和Flume整合完成实时数据采集

在https://www.jianshu.com/p/cc85af7c6cec
中已经简述了Flume的安装和使用。现在整合Flume完成实时的数据采集，采集到的数据通过Kafka的消费者消费，后续会学习spark streaming对收集到的数据进行实时处理。

image.png

• 配置agent
  agent1的配置

agentA.sources = r1
agentA.sinks = k1
agentA.channels = c1

# Describe/configure the source
agentA.sources.r1.type = exec
agentA.sources.r1.command= tail -F /root/data/data.log
agentA.sources.r1.shell = /bin/sh -c

# Describe the sink
agentA.sinks.k1.type = avro
agentA.sinks.k1.hostname  = localhost
agentA.sinks.k1.port = 44444

# Use a channel which buffers events in memory
agentA.channels.c1.type = memory

# Bind the source and sink to the channel
agentA.sources.r1.channels = c1
agentA.sinks.k1.channel = c1

agent2的配置

# Name the components on this agent
kafka-agent.sources = avro-source
kafka-agent.sinks = kafka-sink
kafka-agent.channels = memory-channel

# Describe/configure the source
kafka-agent.sources.avro-source.type = avro
kafka-agent.sources.avro-source.bind = localhost
kafka-agent.sources.avro-source.port = 44444

# Describe the sink
kafka-agent.sinks.kafka-sink.type = org.apache.flume.sink.kafka.KafkaSink
kafka-agent.sinks.kafka-sink.brokerList = 192.168.30.131:9092
kafka-agent.sinks.kafka-sink.topic = test_hi
kafka-agent.sinks.kafka-sink.batchSize = 3
kafka-agent.sinks.kafka-sink.requiredAcks = 1

# Use a channel which buffers events in memory
kafka-agent.channels.memory-channel.type = memory

# Bind the source and sink to the channel
kafka-agent.sources.avro-source.channels = memory-channel
kafka-agent.sinks.kafka-sink.channel = memory-channel

• 启动两个agent

flume-ng agent --name agentA --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/clientA.conf -Dflume.root.logger=DEBUG,console

flume-ng agent --name kafka-agent --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/kafka-agent.conf -Dflume.root.logger=DEBUG,console

• 启动Kafka的消费者

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_hi

• 在日志文件中输入字符，模拟日志的生成。

echo kafka-test >> data.log

• 在Kafka的消费者界面查看是否消费到了消息

  
  image.png
  

  如图可见消费者可以消费到，接下来会学习spark streaming来替换消费者。