大数据(9)：Kafka

一、Kafka 简介

用官网的话来说：Apache Kafka is an open-source distributed event streaming platform……
Apache Kafka 是一个分布式流处理平台……这到底意味着什么呢?

• 消息系统： Kafka 和传统的消息系统（消息中间件）都具备系统解耦、冗余存储、流量削峰、缓冲、异步通信、扩展性、可恢复性等功能。与此同时，Kafka 还提供了大多数消息系统难以实现的消息顺序性保障及回溯消费的功能。

• 存储系统： Kafka 把消息持久化到磁盘，相比于其他基于内存存储的系统而言，降低了数据丢失的风险。因此可以把 Kafka 作为长期的数据存储系统来使用，只需要把对应的数据保留策略设置为“永久”或启用主题的日志压缩功能。

• 流式处理平台： Kafka 不仅为每个流行的流式处理框架提供了可靠的数据来源，还提供了一个完整的流式处理类库，比如窗口、连接、变换和聚合等各类操作。

二、为什么使用消息系统？

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋等问题

1. 应用解耦

在系统交互时，我们很难一次性就设计出完善的接口，会随着业务发展，交互接口不断的变迁，变得越来越复杂，各个系统之间的关系错综复杂，这时就需要考虑引入一种基于数据的接口层（消息队列）。

生产者只需要把数据放入消息队列中，不关心消费者；而消费者只需要去队列中取数据，不关心生产者。只要保证数据结构不变，各系统可以独立的扩展或修改自己的处理过程，解除了两者之间的耦合。

解耦的同时也提高了系统的稳定性（某个组件失效不会影响其他部分正常运行）和扩展性（可以横向扩展系统以增加处理消息的能力）。

2. 异步处理

如果业务逻辑涉及多个步骤，而这些步骤关联性不强，如果我们串行执行，总耗时=每个步骤耗时之和，如果我们让每个步骤并行处理，总耗时< 每个步骤耗时之和。

例如电商系统中，下单后发送短信，没必要等短信发送成功后，才返回下单成功，只需要下单后，发送消息到消息队列就返回成功，短信模块则默默的从消息队列中读取数据，再默默的发送短信就可以了。

在这里我们引入消息队列，将每个处理步骤发送到消息队列，并且针对每个处理步骤都有对应的线程去监听，这样就能达到串行化转为异步并行，从而提高系统的的吞吐量。

3. 流量削峰

在秒杀或抢购活动中，一般会因为流量暴增，应用因处理不过来而挂掉，此时一般会引入消息队列，流量先进入消息队列，应用再根据自己的实际处理能力来消费这些消息，从而达到缓解流量暴增对系统构成的压力。

三、消息队列模式

消息队列包括两种模式，点对点/队列 (point to point, queue) 模式 和 发布-订阅/主题 (publish-subscribe, topic) 模式。

点对点模式

1、每条消息只有一个接收者 (Consumer) 收到，一旦被消费，消息就不再在消息队列中；
2、发送者和接收者间没有依赖性，发送者发送消息之后，不管有没有接收者，都不会影响到发送者下次发送消息；
3、接收者在成功接收消息之后需向队列应答成功，以便消息队列删除当前接收的消息；

点对点模式

发布/订阅模式

在发布-订阅消息系统中，消息被持久化到一个 topic 中。与点对点模式不同的是，消费者可以订阅一个或多个 topic，消费者可以消费该 topic 中所有的数据，同一条数据可以被多个消费者消费，数据被消费后不会立马删除。在发布-订阅消息系统中，消息的生产者称为发布者，消费者称为订阅者。

发布订阅模式

Kafka 将这两种模式统一起来，有一个消费者组的概念。每个消费者可以加入一个消费者组，同一个组中的消费者共同竞争一个 topic 中的消息。

如果所有的消费者都在一个组中，这就成为了传统的队列模式。如果所有的消费者在不同的组中，这就成为了发布-订阅模式，所有的消息都被分发到每个消费者中。

四、Quick Start

1、下载
官网下载 Kafka 并解压

$ tar -xzf kafka_2.13-2.7.0.tgz
$ cd kafka_2.13-2.7.0

2、Kafka 所需环境
注意：Kafka 需要 Java 8+ 环境才能运行，同时还需要安装 Zookeeper，如果你还不了解 Zookeeper 可以看看 《大数据(7)：ZooKeeper》这篇文章。
进入 Zookeeper 所在目录，启动 Zookeeper

# Start the ZooKeeper service
# Note: Soon, ZooKeeper will no longer be required by Apache Kafka.
$ bin/zookeeper-server-start.sh config/zookeeper.properties

在启动前，还需要进行一些配置。进入 kafka 配置目录 kafka/config，编辑 server.properties 文件，需要注意以下几个配置：

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

# A comma separated list of directories under which to store log files
log.dirs=/Users/Downloads/soft_tmp/kafka

# Zookeeper connection string (see zookeeper docs for details).
# e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
zookeeper.connect=localhost:2181

3、启动 Kafka
进入 Kafka 目录，启动

# Start the Kafka broker service
$ bin/kafka-server-start.sh config/server.properties

4、创建主题
Kafka是一个分布式事件流平台，可跨多台计算机读取，写入，存储和处理事件（也称为记录或消息）。在发送消息之前，需要先创建主题（topic），所有数据都是存储在 topic 中。

$ bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --partitions 2 --replication-factor 1 --topic test

这句的意思是，联系 localhost:9092 创建一个分区为 2，副本为 1 的主题，主题的名称是 test。

kafka-topics.sh 不带任何参数的命令即可显示使用情况信息。例如，显示 topic 的详细信息：

$ bin/kafka-topics.sh --describe --topic test --bootstrap-server localhost:9092
Topic:quickstart-events  PartitionCount:2    ReplicationFactor:1 Configs:
    Topic: test Partition: 0    Leader: 0   Replicas: 0 Isr: 0
    Topic: test Partition: 1    Leader: 0   Replicas: 0 Isr: 0

5、发送消息
主题已经创建了，现在可以往主题中发送消息了

$ bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:9092
This is my first event
This is my second event

6、读取消息
打开另一个终端会话并运行控制台使用者客户端以读取刚刚创建的事件：

$ bin/kafka-console-consumer.sh --topic test --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event

五、Kafka 架构

一个典型的 Kafka 体系架构包括若干 Producer、若干 Broker、若干 Consumer，以及一个 ZooKeeper 集群，如下图所示。其中 ZooKeeper 是 Kafka 用来负责集群元数据的管理、控制器的选举等操作的。Producer 将消息发送到 Broker，Broker 负责将收到的消息存储到磁盘中，而 Consumer 负责从 Broker 订阅并消费消息。

Kafka 架构

整个 Kafka 体系结构中引入了以下几个关键词：

Producer：生产者，也就是发送消息的一方。生产者负责创建消息，然后将其投递到 Kafka 中。

Consumer：消费者，也就是接收消息的一方。消费者连接到 Kafka 上并接收消息，进而进行相应的业务逻辑处理。

Broker：服务代理节点。对于 Kafka 而言，Broker 可以简单地看作一个独立的 Kafka 服务节点或 Kafka 服务实例。

Consumer Group：消费者组。可以把多个消费者放入同一个组中，当然也能每个消费者自成一组。

如果多个消费者在同一组，那么组内所有消费者共享一个 group id，拥有相同 group id 的消费者，在消费某个 topic 的消息时，只有 1 个消费该 topic 中的消息。

因此，当每个消费者自成一组时，就是发布-订阅模式；当所有消费者在同一组时，就是队列模式。Kafka 通过消费者组，统一了发布订阅模式和队列模式。

六、主题 (Topic) 与分区 (Partition)

Kafka 中还有两个特别重要的概念——主题（Topic）与分区（Partition）。

Kafka 中的消息以 Topic 为单位进行归类，生产者负责将消息发送到某个的 Topic，然后，这条消息被记录和存储在这个 Topic 中，不允许被修改。消费者负责订阅并进行消费 Topic 中的数据，但消费后，这条消息并不会被删除，而是会继续保留在队列中（消息在 Topic 中能呆多久，可以修改 Kafka 的配置）。

例如当在创建 topic 时，使用了这句：

$ kafka-topics.sh --create --bootstrap-server localhost:9092 --partitions 2 --replication-factor 1 --topic test

这行命令创建了主题为 test，2 个分区，1 个副本。实际上是创建了 test-0、test-1 两个文件夹。

当生产者发送消息时，默认是面向 Topic 的，由 Topic 决定放在哪个 Partition，默认使用轮询策略。然后，消息会被顺序追加到 test-0 或 test-1 文件夹中日志文件的尾部，也就是说 Partition 在存储层面可以看作一个可追加的日志文件。

消息在被追加到分区日志文件的时候都会分配一个特定的偏移量（offset），并且一条消息只会进入一个分区中，也就是说即使在同一 Topic 中，不同 Partition 里的消息是不同的。

因此在有多个 Partition 的情况下，Kafka 中的 Partition 可以分布在不同的服务器（broker）上，因此一个 Topic 可以横跨多个 broker，以此来提供比单个 broker 更强大的性能。

所以，Topic 只是逻辑上的概念，实际上是由 Partition 组成：一个 Topic 可以有多个 Partition，而一个 Partition 只属于一个 Topic。一个 Partition 在物理上就对应一个文件夹。

topic and partition

也可以配置 Topic，让同类型的消息都在同一个 Partition。例如，处理用户消息，可以让某一个用户所有消息都在一个 Partition。在配置之后，可以确保用户1的所有消息都发到同一个分区中。

这个功能有什么用呢？

这是为了提供消息的有序性。消息在不同的 Partition 是不能保证有序的，只有一个 Partition 内的消息是有序的。

若 topic 有 n 个 partition，集群有 n 个 broker，则每个 broker 存储该 topic 的一个 partition。

若 topic 有 n 个 partition，集群有 (n + m) 个 broker，则其中 n 个 broker 存储该 topic 的一个 partition，剩下的 m 个 broker 没有 topic 的 partition 数据。

若 topic 有 n 个 partition，集群 broker 数目少于 n，那么一个 broker 存储该 topic 的一个或多个 partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致 Kafka 集群数据不均衡。

七、Kafka 总结

kafka data flow

每个 broker 有一个或多个 topic，每个 topic 有一个或多个 partition，每个 partition 有一个或多个 replication。

一个 producer 发送消息到 topic 时，实际上会对消息的 key 进行 hash 并根据 partition 个数进行取模（每条消息由 key 和 value 组成，value 为消息的实际内容），根据取模后的结果，将消息放入对应的 partition 中。若没填写 key，则将以轮询的方式放入 partition 中。

若有多个 replication (replication 个数不能大于 broker 个数)，producer 将先把消息发送给 leader，leader 再将消息以同步或异步的方式发送给 follower。