今天给公司小伙伴培训的 Kafka 入门，了解一下？

消息中间件/系统

中间件

中间件是一类链接软件组件和应用的计算机软件，它包括一组服务。以便于运行在一台或多台机器上的多个软件经过网络进行交互。

该技术所提供的互操做性，推进了一致分布式体系架构的演进，该架构一般用于支持并简化那些复杂的分布式应用程序，它包括 web服务器、事务监控器和消息队列软件。

消息队列

在计算机科学中，消息队列（英语：Message queue）是一种进程间通讯或同一进程的不一样线程间的通讯方式，软件的贮列用来处理一系列的输入，一般是来自用户。

消息队列提供了异步的通讯协议，每个贮列中的纪录包含详细说明的数据，包含发生的时间，输入设备的种类，以及特定的输入参数，也就是说：消息的发送者和接收者不须要同时与消息队列交互。消息会保存在队列中，直到接收者取回它。

消息队列经常保存在链表结构中。拥有权限的进程能够向消息队列中写入或读取消息。

--以上来自 维基百科

让咱们用快递员送快递的例子来理解。

最开始送上门，你有时候不在家，又只有等次日，如今多了个菜鸟驿站，快递小哥只须要往里面放，你有快递本身去驿站拿就好了。

快递小哥也不用直接对接这么多的客户，有包裹往驿站丢就行，而后咱们去驿站拿，就算有时候咱们不空，也可让包裹在驿站多放几天，这也就是它的堆积能力。

为何咱们要用消息队列

异步处理

假设你有一个系统调用的链路，系统A 调用系统B 耗时 50ms,系统B 调用系统C 又须要200ms ,系统C 调用系统 D ，须要作比较超时的操做，须要 2s,以下图所示：

如今上面最大的问题在于：一个用户请求过来，整个链路的调用时间是 50ms + 200ms + 2000ms = 2250ms,也就是2秒多。

而事实上，调用链路中，系统A 调用系统 B，系统B 调用系统 C 总共加起来也才 250ms，可是系统C调用系统D 却用了 2S。

正是加入系统C调用系统D 这个链路，致使系统响应时间 从 250ms 增长到了 2250 ms，足足慢了 10倍。

若是说，咱们把系统D 从链路中抽离出去，让 C 系统异步调用D，那么在 B系统调用 C，C 处理完成本身的逻辑，发送一个异步的请求去调用D系统，不用阻赛等到 D系统响应了再返回。这是否是好不少了呢？

举一个例子，就以咱们日常点外卖为例：

咱们日常点完餐，付完款，系统而后平给帐户扣款、建立订单、通知商家准备菜品。

接着，是否是须要找个骑手给你送餐？那这个找骑手的过程，是须要一套复杂算法来实现调度的，比较耗时。

那么咱们是否是就能够把找骑手给你送餐的这个步骤从链路中抽离出去，作成异步化的，哪怕延迟个几十秒，可是只要在必定时间范围内给你找到一个骑手去送餐就能够了。

这样是否是就可让你下订单点外卖的速度变得超快？支付成功以后，直接建立好订单、帐户扣款、通知商家立马给你准备作菜就ok了，这个过程可能就几百毫秒。而后后台异步化的耗费可能几十秒经过调度算法给你找到一个骑手去送餐，可是这个步骤不影响咱们快速下订单。

因此上面的链路也是同理，若是业务流程支持异步化的话，是否是就能够考虑把系统C对系统D的调用抽离出去作成异步化的，不要放在链路中同步依次调用。

整个过程以下：

消息解耦

假如你如今系统A，这个系统会产出一个核心数据，下游系统 B和 C 都须要这个数据。

那么咱们日常作的就是直接调用系统 B 和系统 C，发送数据过去。

过程以下：

过几天，其余的业务系统D、E 也须要这个数据，而后成了这样

若是后续还有系统要呢？你不应代码改死。。。

这种状况系统耦合很是的严重，若是你发送一个系统的调用失败了怎么整？

针对上面的问题，咱们可使用消息队列来实现系统解藕。

系统A 把数据发送到消息队列中，其余的系统，谁须要，本身去 消息队列 取就完了。

流量消峰&日志收集

假设你有一个系统，平时正常的时候每秒可能就几百个请求，系统部署在8核16G的机器的上，正常处理都是ok的，每秒几百请求是能够轻松抗住的。

可是以下图所示，在高峰期一会儿来了每秒钟几千请求，瞬时出现了流量高峰，此时你的选择是要搞10台机器，抗住每秒几千请求的瞬时高峰吗？

那若是瞬时高峰天天就那么半个小时，接着直接就下降为了每秒就几百请求，若是你线上部署了不少台机器，那么每台机器就处理每秒几十个请求就能够了，这不是有点浪费机器资源吗？

大部分时候，每秒几百请求，一台机器就足够了，可是为了抗那天天瞬时的高峰，硬是部署了10台机器，天天就那半个小时有用，别的时候都是浪费资源的。

此时咱们就可使用消息队列来帮忙了，进行消峰。全部机器前面部署一层MQ，平时每秒几百请求你们均可以轻松接收消息。

一旦到了瞬时高峰期，一下涌入每秒几千的请求，就能够积压在MQ里面，而后那一台机器慢慢的处理和消费。

等高峰期过了，再消费一段时间，MQ里积压的数据就消费完毕了。

以下图：

以上内容来自于【石杉的架构笔记】，图是自个画的，算是本身的一些理解吧。

要了解他的更多内容，移步 juejin.im/user/5be058…

消息队列有哪些

四种经常使用的消息队列 ActiveMQ、RabbitMQ、RocketMQ、Kafka

juejin.im/post/5b41fe…

Kafka 是什么

Kafka是一种高吞吐量的分布式发布订阅消息系统，使用Scala编写。

对于熟悉JMS（Java Message Service）规范的同窗来讲，消息系统已经不是什么新概念了(例如ActiveMQ，RabbitMQ等)。

Kafka拥有做为一个消息系统应该具有的功能，可是确有着独特的设计。能够这样来讲，Kafka借鉴了JMS规范的思想，可是确并无彻底遵循JMS规范。

kafka是一个分布式的，分区的消息(官方称之为commit log)服务。它提供一个消息系统应该具有的功能，可是确有着独特的设计。首先，让咱们来看一下基础的消息(Message)相关术语：

• Topic: Kafka按照Topic分类来维护消息
• Producer： 咱们将发布(publish)消息到Topic的进程称之为生产者(producer)
• Consumer： 咱们将订阅(subscribe)Topic而且处理Topic中消息的进程称之为消费者(consumer)
• Broker： Kafka以集群的方式运行，集群中的每一台服务器称之为一个代理(broker)。

所以，从一个较高的层面上来看，producers 经过网络发送消息到Kafka集群，而后consumers来进行消费，以下图：

服务端(brokers)和客户端(producer、consumer)之间通讯经过TCP协议来完成。咱们为Kafka提供了一个Java客户端，可是也可使用其余语言编写的客户端。

Kafka的架构图

Kafka的核心概念

Topic 和 Log

让咱们首先深刻理解Kafka提出一个高层次的抽象概念 --Topic。

能够理解Topic是一个类别的名称，全部的message发送到Topic下面。对于每个Topic，kafka集群按照以下方式维护一个分区(Partition,能够就理解为一个队列Queue)日志文件:

partition 是一个有序的 message 序列，这些 message 按顺序添加到一个叫作 commit log 的文件中。每一个 partition 中的消息都有一个惟一的编号，称之为 offset，用来惟一标示某个分区中的message。

partition 支持消息位移读取，消息位移有消费者自身管理，好比下图：

由上图能够看出，不一样消费者对同一分区的消息读取互不干扰，消费者能够经过设置消息位移（offset）来控制本身想要获取的数据，好比：能够从头读取，最新数据读取，重读读取等功能。

Distribution

log 的 partitions 分布在 kafka 集群中不一样的 broker 上，每一个 broker 能够请求备份其余broker 上 partition 上的数据。kafka 集群支持配置一个 partition 备份的数量。针对每一个partition，都有一个 broker 起到“leader”的做用，0个多个其余的 broker 做为“follwers”的做用。

leader 处理全部的针对这个 partition 的读写请求，而 followers 被动复制 leader 的结果。若是这个 leader 失效了，其中的一个 follower 将会自动的变成新的 leader。

每一个 broker 都是本身所管理的 partition 的 leader，同时又是其余 broker 所管理 partitions 的 followers，kafka 经过这种方式来达到负载均衡。

Producers

生产者将消息发送到topic中去，同时负责选择将message发送到topic的哪个partition中。经过round-robin作简单的负载均衡。也能够根据消息中的某一个关键字来进行区分。一般第二种方式使用的更多。

Consumers

传统的消息传递模式有2种：队列( queuing)和（ publish-subscribe）。

在queuing模式中，多个consumer从服务器中读取数据，消息只会到达一个consumer。

在 publish-subscribe 模型中，消息会被广播给全部的consumer。

Kafka基于这2种模式提供了一种consumer的抽象概念：consumer group

每一个consumer都要标记本身属于哪个consumer group。发布到topic中的message中message会被传递到consumer group中的一个consumer 实例。consumer实例能够运行在不一样的进程上，也能够在不一样的物理机器上。

若是全部的consumer都位于同一个consumer group 下，这就相似于传统的queue模式，并在众多的consumer instance之间进行负载均衡。

若是全部的consumer都有着本身惟一的consumer group，这就相似于传统的publish-subscribe模型。

更通常的状况是，一般一个topic会有几个consumer group，每一个consumer group都是一个逻辑上的订阅者（ logical subscriber ）。每一个consumer group由多个consumer instance组成，从而达到可扩展和容灾的功能。这并无什么特殊的地方，仅仅是将publish-subscribe模型中的运行在单个进程上的consumers中的consumer替换成一个consumer group。以下图所示：

说明：由2个broker组成的kafka集群，总共有4个Parition(P0-P3)。这个集群由2个Consumer Group， A有2个 consumer instances ，而B有四个.

消费顺序

Kafka比传统的消息系统有着更强的顺序保证。在传统的状况下，服务器按照顺序保留消息到队列，若是有多个consumer来消费队列中的消息，服务器 会接受消息的顺序向外提供消息。

可是，尽管服务器是按照顺序提供消息，可是消息传递到每个consumer是异步的，这可能会致使先消费的 consumer获取到消息时间可能比后消费的consumer获取到消息的时间长，致使不能保证顺序性。

这代表，当进行并行的消费的时候，消息在多个 consumer之间可能会失去顺序性。

消息系统一般会采起一种“ exclusive consumer”的概念，来确保同一时间内只有一个consumer可以从队列中进行消费，可是这实际上意味着在消息处理的过程当中是不支持并行的。

Kafka在这方面作的更好。经过Topic中并行度的概念，即partition，Kafka能够同时提供顺序性保证和多个consumer同时消费时的负载均衡。实现的原理是经过将一个topic中的partition分配给一个consumer group中的不一样consumer instance。

经过这种方式，咱们能够保证一个partition在同一个时刻只有一个consumer instance在消息，从而保证顺序。

虽然一个topic中有多个partition，可是一个consumer group中同时也有多个consumer instance，经过合理的分配依然可以保证负载均衡。

须要注意的是，一个consumer group中的consumer instance的数量不能比一个Topic中的partition的数量多。若是多了，它将分配不上分区消息。

Kafka只在partition的范围内保证消息消费的局部顺序性，不能在同一个topic中的多个partition中保证总的消费顺序性。一般来讲，这已经能够知足大部分应用的需求。

可是，若是的确有在整体上保证消费的顺序的需求的话，那么咱们能够经过将topic的partition数量设置为1，将consumer group中的consumer instance数量也设置为1。

可是这样作，Kafka 的吞吐量就会降低。

kafka的安装与使用

安装前的环境准备

因为Kafka是用Scala语言开发的，运行在JVM上，所以在安装Kafka以前须要先安装JDK。

# yum install java-1.8.0-openjdk* -y
复制代码

kafka依赖zookeeper，因此须要先安装zookeeper

# wget http://mirror.bit.edu.cn/apache/zookeeper/stable/zookeeper-3.4.12.tar.gz
# tar -zxvf zookeeper-3.4.12.tar.gz
# cd zookeeper-3.4.12
# cp conf/zoo_sample.cfg conf/zoo.cfg 启动zookeeper
# bin/zkServer.sh start
# bin/zkCli.sh
# ls / #查看zk的根目录相关节点
复制代码

第一步：下载安装包

下载1.1.0 release版本，并解压：

# wget https://archive.apache.org/dist/kafka/1.1.0/kafka_2.11-1.1.0.tgz
# tar -xzf kafka_2.11-1.1.0.tgz
# cd kafka_2.11-1.1.0
复制代码

第二步：启动服务

如今来启动kafka服务： 启动脚本语法：kafka-server-start.sh [-daemon] server.properties

能够看到，server.properties的配置路径是一个强制的参数，-daemon表示之后台进程运行，不然ssh客户端退出后，就会中止服务。(注意，在启动kafka时会使用linux主机名关联的ip地址，因此须要把主机名和linux的ip映射配置到本地host里，用vim /etc/hosts)

# bin/kafka-server-start.sh -daemon config/server.properties
复制代码

咱们进入zookeeper目录经过zookeeper客户端查看下zookeeper的目录树

# bin/zkCli.sh
# ls / #查看zk的根目录kafka相关节点
# ls /brokers/ids #查看kafka节点
复制代码

第三步：建立主题

如今咱们来建立一个名字为“test”的Topic，这个topic只有一个partition，而且备份因子也设置为1：

# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
复制代码

如今咱们能够经过如下命令来查看kafka中目前存在的topic

# bin/kafka-topics.sh --list --zookeeper localhost:2181
复制代码

除了咱们经过手工的方式建立Topic，咱们能够配置broker，当producer发布一个消息某个指定的Topic，可是这个Topic并不存在时，就自动建立。

第四步：发送消息

kafka自带了一个producer命令客户端，能够从本地文件中读取内容，或者咱们也能够以命令行中直接输入内容，并将这些内容以消息的形式发送到kafka集群中。在默认状况下，每个行会被当作成一个独立的消息。

首先咱们要运行发布消息的脚本，而后在命令中输入要发送的消息的内容：

# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>this is a msg
>this is a another msg
复制代码

第五步：消费消息

对于consumer，kafka一样也携带了一个命令行客户端，会将获取到内容在命令中进行输出：

# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning #老版本

# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --consumer-property group.id=testGroup --consumer-property client.id=consumer-1 --topic test #新版本
复制代码

若是你是经过不一样的终端窗口来运行以上的命令，你将会看到在producer终端输入的内容，很快就会在consumer的终端窗口上显示出来。

Kafka实操的一些命令

启动zk 
bin/zkServer.sh start
启动kafka
bin/kafka-server-start.sh config/server.properties &
中止kafka 若是无论用 就是用kill -9 
bin/kafka-server-stop.sh
1.建立主题
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
2.列出主题
bin/kafka-topics.sh --list --zookeeper localhost:2181
3.生产消息
bin/kafka-console-producer.sh --broker-list localhost:9092 -topic test
4.消费消息
bin/kafka-console-consumer.sh --zookeeper localhost:2181 -topic test --from-beginning
5.删除主题
 1. 删除 kafka 主题 
    bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic sceniccenter-base-ticket
 2. 在kafka 数据目录删除主题文件夹
 3. 删除 zookeeper 上的 记录
    1）登陆zookeeper客户端：命令：./zkCli.sh 
        2）找到topic所在的目录：ls /brokers/topics
        3）找到要删除的topic，执行命令：rmr /brokers/topics/【topic name】便可，此时topic被完全删除。    
    另外被标记为 marked for deletion 的topic你能够在zookeeper客户端中经过命令得到： ls /admin/delete_topics/【topic name】
  总结  完全删除topic：
     一、删除kafka存储目录（server.properties文件log.dirs配置，默认为"/tmp/kafka-logs"）相关topic目录
     二、若是配置了delete.topic.enable=true直接经过命令删除，若是命令删除不掉，直接经过 zookeeper-client 删除掉broker下的topic便可。
6.查看toplic 的分区等状况 
bin/kafka-topics --describe --zookeeper hadoop1:2181 --topic wwcc1
Topic:wwcc1    PartitionCount:3    ReplicationFactor:3    Configs:
    Topic: wwcc1    Partition: 0    Leader: 127    Replicas: 127,128,129    Isr: 127,128,129
    Topic: wwcc1    Partition: 1    Leader: 128    Replicas: 128,129,127    Isr: 128,129,127
    Topic: wwcc1    Partition: 2    Leader: 129    Replicas: 129,127,128    Isr: 129,127,128

复制代码

一些学习地址

系统架构中为何要用消息中间件-中华石杉：juejin.im/post/5c0fba…

官方地址：kafka.apache.org/

中文教程： orchome.com/kafka/index

消息中间件MQ详解：blog.51cto.com/leexide/210…