2. kafka - 爱码网

一. kafka是什么

https://www.jianshu.com/p/014af2b34159

Kafka是一个分布式消息队列，它将数据分区保存，并将每个分区保存成多份以提高数据可靠性。Kafka是在大数据背景下产生的，用以应对海量数据的处理场景，具有高性能、良好的扩展性、数据持久性等特点。

Kafka架构由生产者、代理和消费者三类组件构成。生产者将数据推送给代理，消费者从代理商拉取数据进行处理，而代理通过ZooKeeper进行协调和管理。生产者和消费者可根据业务需要自定义开发，多个代理构成一个可靠的分布式消息存储系统，避免数据丢失。代理中的消息被划分成若干个主题，同属于一个主题的的所有数据被分成多个分区，以实现负载分摊和数据并行处理。Kafka基本架构如下：

2. kafka

https://segmentfault.com/a/1190000013834998?utm_source=tag-newest

Kafka是运行在一个集群上，所以它可以拥有一个或多个服务节点；
Kafka集群将消息存储在特定的文件中，对外表现为Topics；
每条消息记录都包含一个key,消息内容以及时间戳；

Kafka为了拥有更强大的功能，提供了四大核心接口：

- Producer API允许了应用可以向Kafka中的topics发布消息；
- Consumer API允许了应用可以订阅Kafka中的topics,并消费消息；
- Streams API允许应用可以作为消息流的处理者，比如可以从topicA中消费消息，处理的结果发布到topicB中；
- Connector API提供Kafka与现有的应用或系统适配功能，比如与数据库连接器可以捕获表结构的变化；

它们与Kafka集群的关系可以用下图表示：

2. kafka

二. Kafka的使用场景：

- 日志收集：一个公司可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。
- 消息系统：解耦和生产者和消费者、缓存消息等。
- 用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。
- 运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。
- 流式处理：比如spark streaming和storm
- 事件源

2. kafka

三.相关术语

https://www.jianshu.com/p/1d496ebd0edf

https://www.jianshu.com/p/446e012a2d3b

Message(消息)：Kafka中的一条记录或数据单位。每条消息都有一个键和对应的一个值，有时还会有可选的消息头。在发送消息时，我们可以省略掉key部分，而直接使用value部分。key的主要作用则是根据一定的策略，将此消息发送到指定的分区中，这样就可以确保包含同一key值的消息全部都写入到同一个分区中。
Topic(主题)：topic是那些被发布的数据记录或消息的一种类别。消费者通过订阅topic，来读取写给它们的数据。首先，主题是一个逻辑上的概念，它用于从逻辑上来归类与存储消息本身。多个生产者可以向一个Topic发送消息，同时也可以有多个消费者消费一个Topic中的消息。Topic还有分区和副本的概念，后续介绍。Topic与消息这两个概念之间密切相关，Kafka中的每一条消息都归属于某一个Topic，而一个Topic下面可以有任意数量的消息。正是借助于Topic这个逻辑上的概念，Kafka将各种各样的消息进行了分门别类，使得不同的消息归属于不同的Topic，这样就可以很好地实现不同系统的生产者可以向同一个Broker发送消息，而不同系统的消费者则可以根据Topic的名字从Broker中拉取消息。Topic是一个字符串。通过Topic这样一个逻辑上的概念，我们就很好地实现了生产者与消费者之间有针对性的发送与拉取。一个topic是对一组消息的归纳。对每个topic，Kafka 对它的日志进行了分区，如下图所示：

Topic partition(主题分区)：不同的topic被分为不同的分区，而每一条消息都会被分配一个offset，通常每个分区都会被复制至少一到两次。➢ 每一条消息发送到 broker 时，会根据 partition 的规则选择存储到哪一个 partition。如果 partition 规则设置合理，那么所有的消息会均匀的分布在不同的 partition 中，这样就有点类似数据库的分库分表的概念，把数据做了分片处理。

2. kafka

Offset(偏移量)：单个分区中的每一条消息都被分配一个offset，它是一个单调递增的整型数，可用来作为分区中消息的唯一标识符。kafka 通过 offset 保证消息在分区内的顺序，offset 的顺序不跨分区，即 kafka 只保证在同一个分区内的消息有序。
Producer(生产者)：producer将消息发布到Kafka的topics上。producer决定向topic分区的发布方式，如：轮询的随机方法、或基于消息键(key)的分区算法。
Broker(代理)：Kafka以分布式系统或集群的方式运行。那么群集中的每个节点称为一个broker。在消息队列领域中，它指的其实就是消息队列产品本身，比如说在Kafka这个领域下，Broker其实指的就是一台Kafka Server。换句话说，我们可以将部署的一个Kafka Server看作是一个Broker，就是这样简单。那么从流程上来说，生产者会将消息发送给Broker，然后消费者再从Broker中拉取消息。

Consumer(消费者)：consumer通过订阅topic partition，来读取Kafka的各种topic消息。然后，消费类应用处理会收到消息，以完成指定的工作。
Consumer group(消费组)：在Kafka中，多个消费者可以共同构成一个消费者组，而一个消费者只能从属于一个消费者组。消费者组最为重要的一个功能是实现广播与单播的功能。consumer可以按照consumer group进行逻辑划分。topic partition被均衡地分配给组中的所有consumers。因此，在同一个consumer group中，所有的consumer都以负载均衡的方式运作。换言之，同一组中的每一个consumer都能看到每一条消息。如果某个consumer处于“离线”状态的话，那么该分区将会被分配给同组中的另一个consumer。这就是所谓的“再均衡(rebalance)”。当然，如果组中的consumer多于分区数，则某些consumer将会处于闲置的状态。相反，如果组中的consumer少于分区数，则某些consumer会获得来自一个以上分区的消息。

由两个机器组成的集群拥有4个分区 (P0-P3) 2个consumer组. A组有两个consumer, B组有4个. 相比传统的消息系统，Kafka可以很好的保证有序性。

Lag(延迟)：当consumer的速度跟不上消息的产生速度时，consumer就会因为无法从分区中读取消息，而产生延迟。延迟表示为分区头后面的offset数量。从延迟状态(到“追赶上来”)恢复正常所需要的时间，取决于consumer每秒能够应对的消息速度。其公式如下：

time = messages / (consume rate per second - produce rate per second)

四.原理解析

https://www.jianshu.com/p/26d4bf0ccb1d

1. 消息分发

2.1 kafka 消息分发策略

消息是 kafka 中最基本的数据单元.
在 kafka 中，一条消息由 key、value 两部分构成，在发送一条消息时，我们可以指定这个 key，那么 producer 会根据 key 和 partition 机制来判断当前这条消息应该发送并存储到哪个 partition 中.
我们可以根据需要进行扩展 producer 的 partition 机制。

2.2 消息默认的分发机制

默认情况下，kafka 采用的是 hash 取模的分区算法。
如果 Key 为 null，则会随机分配一个分区。这个随机是在这个参数metadata.max.age.ms的时间范围内随机选择一个。
对于这个时间段内，如果 key 为 null，则只会发送到唯一的分区。该值默认情况下10 分钟更新一次。
关于 Metadata，简单理解就是 Topic/Partition 和 broker 的映射关系，每一个 topic 的每一个 partition，需要知道对应的 broker 列表是什么，leader 是谁、follower 是谁。这些信息都是存储在 Metadata 这个类里面。

2.3 消费端如何消费指定的分区

通过下面的代码，就可以消费指定该 topic 下的 0 号分区。其他分区的数据就无法接收

//消费指定分区的时候，不需要再订阅 
//kafkaConsumer.subscribe(Collections.singleto nList(topic));

//消费指定的分区
TopicPartition topicPartition=new TopicPartition(topic,0); 
kafkaConsumer.assign(Arrays.asList(topicPartit ion));

View Code