简介

Kafka是一个实时数据处理系统，可以横向扩展、高可靠，而且还变态快，已经被很多公司使用。
那么什么是实时数据处理系统呢？顾名思义，实时数据处理系统就是数据一旦产生，就要能快速进行处理的系统。

对于实时数据处理，我们最常见的，就是消息中间件了，也叫MQ（Message Queue，消息队列），也有叫Message Broker的。

为什么需要消息中间件

消息中间件的作用主要有两点：
1. 解耦消息的生产和消费。
2. 缓冲。
想象一个场景，你的一个创建订单的操作，在订单创建完成之后，需要触发一系列其他的操作，比如进行用户订单数据的统计、给用户发送短信、给用户发送邮件等等，就像这样：

1
2
3
4
5
6
createOrder(...){
 ...
 statOrderData(...);
 sendSMS();
 sendEmail();
}

代码这样写似乎没什么问题，可是过了一段时间，你给系统引进了一个用户行为分析服务，它也需要在订单创建完成之后，进行一个分析用户行为的操作，而且随着系统的逐渐壮大，创建订单之后要触发的操作也就越来越多，代码也渐渐膨胀成这样：

1
2
3
4
5
6
7
8
9
10
11
12
createOrder(...){
 ...
 statOrderData(...);
 sendSMS();
 sendEmail();
 // new operation
 statUserBehavior(...);
 doXXX(...);
 doYYY(...);
 // more and more operations
 ...
}

导致代码越来越膨胀的症结在于，消息的生产和消费耦合在一起了。createOrder方法不仅仅要负责生产“订单已创建”这条消息，还要负责处理这条消息。

这就好比BBC的记者，在知道皇马拿到欧冠冠军之后，拿起手机，翻开皇马球迷通讯录，给球迷一个一个打电话，告诉他们，皇马夺冠了。

事实上，BBC的记者只需要在他们官网发布这条消息，然后球迷自行访问BBC，去上面获取这条新闻；又或者球迷订阅了BBC，那么订阅系统会主动把发布在官网的消息推送给球迷。

同样，createOrder也需要一个像BBC官网那样的载体，也就是消息中间件，在订单创建完成之后，把一条主题为“orderCreated”的消息，放到消息中间件去就ok了，不必关心需要把这条消息发给谁。这就完成了消息的生产。

至于需要在订单创建完成之后触发操作的服务，则只需要订阅主题为“orderCreated”的消息，在消息中间件出现新的“orderCreated”消息时，就会收到这条消息，然后进行相应的处理。

因此，通过使用消息中间件，上面的代码也就简化成了：

1
2
3
4
createOrder(...){
 ...
 sendOrderCreatedMessage(...);
}

以后如果在订单创建之后有新的操作需要执行，这串代码也不需要修改，只需要给对消息进行订阅即可。

另外，通过这样的解耦，消费者在消费数据时更加的灵活，不必每次消息一产生就要马上去处理（虽然通常消费者侧也会有线程池等缓冲机制），可以等自己有空了的时候，再过来消息中间件这里取数据进行处理。这就是消息中间件带来的缓冲作用。

Kafka一代 - 消息队列

从上面的描述，我们可以看出，消息中间件之所以可以解耦消息的生产和消费，主要是它提供了一个存放消息的地方——生产者把消息放进来，消费者在从中取出消息进行处理。

那么这个存放消息的地方，应该采用什么数据结构呢？

在绝大多数情况下，我们都希望先发送进来的消息，可以先被处理（FIFO），这符合大多数的业务逻辑，少数情况下我们会给消息设置优先级。不管怎样，对于消息中间件来说，一个先进先出的队列，是非常合适的数据结构：
在这里插入图片描述

那么要怎样保证消息可以被顺序消费呢？

消费者过来获取消息时，每次都把index=0的数据返回过去，然后再删除index=0的那条数据？

很明显不行，因为订阅了这条消息的消费者数量，可能是0，也可能是1，还可能大于1。如果每次消费完就删除了，那么其他订阅了这条消息的消费者就获取不到这条消息了。

事实上，Kafka会对数据进行持久化存储（至于存放多长时间，这是可以配置的），消费者端会记录一个offset，表明该消费者当前消费到哪条数据，所以下次消费者想继续消费，只需从offset+1的位置继续消费就好了。

消费者甚至可以通过调整offset的值，重新消费以前的数据。

那么这就是Kafka了吗？不，这只是一条非常普通的消息队列，我们姑且叫它为Kafka一代吧。

这个Kafka一代用一条消息队列实现了消息中间件，这样的简单实现存在不少问题：

· Topic鱼龙混杂。想象一下，一个只订阅了topic为“A”的消费者，却要在一条有ABCDEFG…等各种各样topic的队列里头去寻找topic为A的消息，这样性能岂不是很慢？
· 吞吐量低。我们把全部消息都放在一条队列了，请求一多，它肯定应付不过来。

由此就引申出了Kafka二代。

Kafka二代 - Partition

要解决Kafka一代的那两个问题，很简单——分布存储。

二代Kafka引入了Partition的概念，也就是采用多条队列， 每条队列里面的消息都是相同的topic：
在这里插入图片描述
Partition的设计解决了上面提到的两个问题：

纯Topic队列。一个队列只有一种topic，消费者再也不用担心会碰到不是自己想要的topic的消息了。
提高吞吐量。不同topic的消息交给不同队列去存储，再也不用以一敌十了。

一个队列只有一种topic，但是一种topic的消息却可以根据自定义的key值，分散到多条队列中。也就是说，上图的p1和p2，可以都是同一种topic的队列。不过这是属于比较高级的应用了，以后有机会再和大家讨论。

Kafka二代足够完美了吗？当然不是，我们虽然通过Partition提升了性能，但是我们忽略了一个很重要的问题——高可用。

万一机器挂掉了怎么办？单点系统总是不可靠的。我们必须考虑备用节点和数据备份的问题。

Kafka三代 - Broker集群

很明显，为了解决高可用问题，我们需要集群。

Kafka对集群的支持也是非常友好的。在Kafka中，集群里的每个实例叫做Broker，就像这样：
在这里插入图片描述
每个partition不再只有一个，而是有一个leader(红色)和多个replica(蓝色)，生产者根据消息的topic和key值，确定了消息要发往哪个partition之后（假设是p1），会找到partition对应的leader(也就是broker2里的p1)，然后将消息发给leader，leader负责消息的写入，并与其余的replica进行同步。

一旦某一个partition的leader挂掉了，那么只需提拔一个replica出来，让它成为leader就ok了，系统依旧可以正常运行。

通过Broker集群的设计，我们不仅解决了系统高可用的问题，还进一步提升了系统的吞吐量，因为replica同样可以为消费者提供数据查找的功能。

Kafka没那么简单

1. kafka的消息结构
我们只知道Kafka内部是一个消息队列，但是队列里的元素长什么样，包含了哪些消息呢？

参考：Kafka - messageformat

2. zookeeper和kafka的关系
如果玩过Kafka的Quick Start教程，就会发现，我们在使用Kafka时，需要先启动一个ZK，那么这个ZK的作用到底是什么呢？

参考：What-is-the-actual-role-of-Zookeeper-in-Kafka

3. 数据可靠性和重复消费
生产者把消息发给Kafka，发送过程中挂掉、或者Kafka保存消息时发送异常怎么办？

同理，消费者获取消费时发生异常怎么办？

甚至，如果消费者已经消费了数据，但是修改offset时失败了，导致重复消费怎么办？

等等这些异常场景，都是Kafka需要考虑的。

参考：Kafka - Message Delivery Semantics

4. pull or push
消费者侧在获取消息时，是通过主动去pull消息呢？还是由Kafka给消费者push消息？

这两种方式各自有什么优劣？

参考：Kafka - push vs pull

5. 如何提高消费者处理性能
还是之前的订单创建的例子，订单创建后，你要给用户发送短信，现在你发现由于你只有一个消费者在发送短信，忙不过来，怎么办？这就有了Kafka里头的消费者组（Consumer Group）的设计。

参考：Understanding-kafka-consumer-groups-and-consumer

6.终极问题：一条消息从生产，到被消费，完整流程是怎样的？
如果能详尽透彻地回答这个问题，那你对Kafka的理解也就非常深入了。

7. 参考文献&学习资源
官网：

Apache Kafka
Kafka简介
Kafka官网文档
一些不错的博客：

Kafka-in-a-nutshell（入门绝佳读物）

What every software engineer should know about real-time data’s unifying abstraction（从这篇文章可以知道LinkedIn为何要开发Kafka）

How to choose the number of topics/partitions in a Kafka cluster?（对Kafka Partition的深入讲解和性能优化指导）
书籍（没看过，但是感觉不错的书）：

Kafka权威指南
Apache Kafka源码剖析（可以自己先看看源码，再看看这本书）

安装JAVA（jdk）

下载安装包

（官网）链接：
在这里插入图片描述
还可以直接通过wget命令直接把JDK安装包下载下来，具体执行命令请度娘。

解压安装包

可自定义一个目录存（/home/temp）放并进行解压,执行命令行如下：

1
2
3
# mkdir /home/temp
# cd  /home/temp
# tar  -zxvf  jdk-8u181-linux-x64.tar.gz

将解压后的【jdk1.8.0_181】里面的所有数据移动到指定的文件夹下（如/usr/local/java）

1
# mkdir /usr/local/java

将【jdk1.8.0_181】里的数据拷贝至java目录下

1
# mv /home/temp/jdk1.8.0_181 /usr/local/java

修改环境变量

修改环境变量，通过命令

1
# vim /etc/profile

在文件末尾添加一下内容：

1
2
3
4
5
export JAVA_HOME=/usr/local/java
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib:$CLASSPATH
export JAVA_PATH=${JAVA_HOME}/bin:${JRE_HOME}/bin
export PATH=$PATH:${JAVA_PATH}

然后，保存并退出(按：wq!)

保存完之后，通过命令source /etc/profile让profile文件立即生效

第四步、测试是否安装成功

①、使用javac命令，不会出现command not found错误

②、使用java -version，出现版本为java version “1.8.0_181”

③、echo $PATH，看看自己刚刚设置的的环境变量配置是否都正确

测试没问题即安装成功

安装Zookeeper

下载

到官网去下载你想要的版本 官网下载地址
在这里插入图片描述

解压

1
#tar -zxvf zookeeper-3.4.10.tar.gz

将解压后的文件复制到/usr/local目录下，并重命名为zookeeper：

1
2
[root@localhost tmp]# cp zookeeper-3.4.10 /usr/local/zookeeper -r  //复制所有文件到zookeeper 文件夹下
[root@localhost tmp]# cd /usr/local/zookeeper   //切换到/usr/local/zookeeper目录下

在这里插入图片描述

配置

but 这里需要更改一下 .cfg 文件名 zookeeper 启动脚本默认是寻找 zoo.cfg 文件。。。。之所以 得修改文件名
切换到zookeeper目录下的conf目录下，重新复制一份zoo_sample.cfg文件并命名为zoo.cfg：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
[root@localhost zookeeper]# cd conf  //切换到目录下
[root@localhost conf]# ll//显示目录下的信息
总用量 12
-rw-r--r--. 1 root root  535 5月   8 18:17 configuration.xsl
-rw-r--r--. 1 root root 2161 5月   8 18:17 log4j.properties
-rw-r--r--. 1 root root  922 5月   8 18:17 zoo_sample.cfg
[root@localhost conf]# cp zoo_sample.cfg zoo.cfg//copy一份到当前目录下，并命名为zoo.cfg
[root@localhost conf]# ll
总用量 16
-rw-r--r--. 1 root root  535 5月   8 18:17 configuration.xsl
-rw-r--r--. 1 root root 2161 5月   8 18:17 log4j.properties
-rw-r--r--. 1 root root  922 5月   8 18:29 zoo.cfg
-rw-r--r--. 1 root root  922 5月   8 18:17 zoo_sample.cfg
[root@localhost conf]#

修改zoo.cfg文件如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
[root@localhost conf]# vi zoo.cfg


# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataDir=D:\\zookeeper-3.5.4-beta\\data
dataLogDir=D:\\zookeeper-3.5.4-beta\\log
# the port at which the clients will connect
admin.serverPort=8082
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1
server.0=192.168.2.3:9092:113

修改内容：

增加了：admin.serverPort=8082 #不然会出现端口被占用的情况，因为默认是和Apache.Tomcat使用的8080端口
修改了：dataDir=D:\zookeeper-3.5.4-beta\data #保存数据的目录
dataLogDir=D:\zookeeper-3.5.4-beta\log #保存日志的目录

tickTime：这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳。
dataDir：顾名思义就是 Zookeeper 保存数据的目录，默认情况下，Zookeeper 将写数据的日志文件也保存在这个目录里。
dataLogDir：顾名思义就是 Zookeeper 保存日志文件的目录
clientPort：这个端口就是客户端连接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户端的访问请求。
server.A = B:C:D
A表示这个是第几号服务器
B 是这个服务器的 ip 地址
C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口
D 表示的是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的 Leader

最后一步配置环境变量：

打开/etc/profile

1
[root@localhost zookeeper]# vi /etc/profile//编辑文件

添加如下内容：

1
2
export ZOOKEEPER=/usr/local/zookeeper
export PATH=$PATH:$ZOOKEEPER/bin

重启配置文件

1
[root@localhost zookeeper]# source /etc/profile    //使生效

启动Zookeeper

因为配置了环境变量，所以在任意目录下都可以运行以下启动命令启动Zookeeper。

1
2
[root@localhost ~]# zkServer.sh start //启动
[root@localhost ~]# zkServer.sh status  //查看运行状态

在这里插入图片描述

启动客户端：

1
[root@localhost ~]# zkCli.sh   //启动客户端

在这里插入图片描述

配置开机启动zookeeper

  1. 在/etc/init.d目录下新建zookeeper文件

    vi /etc/init.d/zookeeper//vi 编辑zookeeper文件，不存在时就创建该文件

输入以下内容：

1
2
3
4
5
6
7
8
9
10
#!/bin/bash
ZK_PATH=/usr/local/zookeeper
export JAVA_HOME=/usr/local/java/jdk1.8.0_171
case $1 in
         start) sh  $ZK_PATH/bin/zkServer.sh start;;
         stop)  sh  $ZK_PATH/bin/zkServer.sh stop;;
         status) sh  $ZK_PATH/bin/zkServer.sh status;;
         restart) sh $ZK_PATH/bin/zkServer.sh restart;;
         *)  echo "require start|stop|status|restart"  ;;
esac
  1. 保存并退出后，执行以下chkconfig --add 指令把脚本注册为Service：
1
chkconfig --add zookeeper

你可以使用chkconfig --list查看你的注册操作时否成功
在这里插入图片描述
注意：
1. Ubuntu 16.04 下安装 Nginx 服务器，在添加 nginx 服务时出现如下信息

1
2
# chkconfig --add nginx
chkconfig: command not found

问题原因
Ubuntu 中 chkconfig 已经被 sysv-rc-conf 所替代，chkconfig 命令如下：

1
2
# chkconfig --add nginx  
# chkconfig nginx on

问题解决

1
2
3
# apt-get update
# apt-get install sysv-rc-conf
# sysv-rc-conf nginx on

2.Ubuntu下安装sysv-rc-conf报错：ubuntuE: Unable to locate package sysv-rc-conf

当我安装sysv-rc-conf时，报了如下的错：
E: Unable to locate package sysv-rc-conf（无法定位sysv-rc-conf包）
在这里插入图片描述
提供一个解决办法，如下：
在软件源列表sources.list（该文本的位置在/etc/apt/sources.list）文件中的末尾添加如下内容：
deb http://archive.ubuntu.com/ubuntu/ trusty main universe restricted multiverse

第一步：终端输入：sudo gedit /etc/apt/sources.list（这里我使用的是gedit，你可以使用自己喜欢的工具，如vi、vim等），打开该文件

1
sudo gedit /etc/apt/sources.list

第二步：在软件源sources.list文件中添加如下一列文本：
deb http://archive.ubuntu.com/ubuntu/ trusty main universe restricted multiverse
在这里插入图片描述
第三步：更新apt-get，在终端输入sudo apt-get update

1
sudo apt-get update

注意，此处更新可能会出错，重复执行更新就好了

第四步：完成更新后，重新安装sysv-rc-conf，在终端输入sudo apt-get install sysv-rc-conf，即可成功安装。
在这里插入图片描述

安装 kafka

下载代码

下载地址：https://kafka.apache.org/downloads，ubuntu下可以用wget直接下载，我是下载到了/home/kafka目录

1
wget http://mirrors.shuosc.org/apache/kafka/1.0.0/kafka_2.11-1.0.0.tgz

解压

1
tar -zxvf kafka_2.11-1.0.0.tgz

在kafka解压目录下创建日志存储目录
在这里插入图片描述
修改kafka-server 的配置文件

1
vim config/server.properties

修改配置文件中21、31、36和60行

1
2
3
4
**broker.id=1** //--使用单机模式可以不修改
**listeners=PLAINTEXT://:9092**  //--使用单机模式可以不修改
**advertised.listeners=PLAINTEXT://host_ip:9092** //--使用单机模式可以不修改
**log.dirs=/home/yzy/kafka/kafka_2.12-2.6.0/kafka_logs**

启动服务器

Kafka 使用 ZooKeeper 如果你还没有ZooKeeper服务器，你需要先启动一个ZooKeeper服务器。 您可以通过与kafka打包在一起的便捷脚本来快速简单地创建一个单节点ZooKeeper实例。

1
bin/zookeeper-server-start.sh config/zookeeper.properties //--该启动脚本在kafka文件夹bin下面，不是在zookeeper里面

现在启动Kafka服务器：

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

创建一个topic

让我们创建一个名为“test”的topic，它有一个分区和一个副本：

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

现在我们可以运行list（列表）命令来查看这个topic：

1
2
bin/kafka-topics.sh --list --zookeeper localhost:2181
test

或者，您也可将代理配置为：在发布的topic不存在时，自动创建topic，而不是手动创建。

发送消息

Kafka自带一个命令行客户端，它从文件或标准输入中获取输入，并将其作为message（消息）发送到Kafka集群。默认情况下，每行将作为单独的message发送。

运行 producer，然后在控制台输入一些消息以发送到服务器。

1
2
3
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>This is a message
>This is another message

启动一个consumer

Kafka 还有一个命令行consumer（消费者），将消息转储到标准输出。

1
2
3
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
This is a message
This is another message

如果您将上述命令在不同的终端中运行，那么现在就可以将消息输入到生产者终端中，并将它们在消费终端中显示出来。

所有的命令行工具都有其他选项；运行不带任何参数的命令将显示更加详细的使用信息。