您的位置：首页 > 其它

storm学习（六）——kafka原理及安装

2018-08-27 23:36 92 查看

首先回顾一下storm的内容：

Storm的数据源一方面来自kafka，kafka的数据送给storm中的spout，供storm使用，下面详细介绍kafka。

Kafka介绍：https://blog.csdn.net/ychenfeng/article/details/74980531

1、Kafka是什么

在流式计算中，Kafka一般用来缓存数据，Storm通过消费Kafka的数据进行计算。

KAFKA + STORM +REDIS

Apache Kafka是一个开源消息系统，由Scala写成。是由Apache软件基金会开发的一个开源消息系统项目。
Kafka最初是由LinkedIn开发，并于2011年初开源。2012年10月从Apache Incubator毕业。该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台。
Kafka是一个分布式消息队列：生产者、消费者的功能。它提供了类似于JMS的特性，但是在设计实现上完全不同，此外它并不是JMS规范的实现。
Kafka对消息保存时根据Topic进行归类，发送消息者称为Producer,消息接受者称为Consumer,此外kafka集群有多个kafka实例组成，每个实例(server)成为broker。
无论是kafka集群，还是producer和consumer都依赖于zookeeper集群保存一些meta信息，来保证系统可用性

2、JMS是什么

2.1、JMS的基础

JMS是什么：JMS是Java提供的一套技术规范

JMS干什么用：用来异构系统集成通信，缓解系统瓶颈，提高系统的伸缩性增强系统用户体验，使得系统模块化和组件化变得可行并更加灵活

通过什么方式：生产消费者模式（生产者、服务器、消费者）

jdk，kafka，activemq……

2.2、JMS消息传输模型

1、点对点模式（一对一，消费者主动拉取数据，消息收到后消息清除）

点对点模型通常是一个基于拉取或者轮询的消息传送模型，这种模型从队列中请求信息，而不是将消息推送到客户端。这个模型的特点是发送到队列的消息被一个且只有一个接收者接收处理，即使有多个消息监听者也是如此。

2、发布/订阅模式（一对多，数据生产后，推送给所有订阅者）

发布订阅模型则是一个基于推送的消息传送模型。发布订阅模型可以有多种不同的订阅者，临时订阅者只在主动监听主题时才接收消息，而持久订阅者则监听主题的所有消息，即时当前订阅者不可用，处于离线状态。

queue.put（object）数据生产

queue.take(object) 数据消费

2.3、JMS核心组件

Destination：消息发送的目的地，也就是前面说的Queue和Topic。

Message ：从字面上就可以看出是被发送的消息。

StreamMessage：Java 数据流消息，用标准流操作来顺序的填充和读取。

MapMessage：一个Map类型的消息；名称为 string 类型，而值为 Java 的基本类型。

TextMessage：普通字符串消息，包含一个String。

ObjectMessage：对象消息，包含一个可序列化的Java 对象

BytesMessage：二进制数组消息，包含一个byte[]。

XMLMessage: 一个XML类型的消息。

最常用的是TextMessage和ObjectMessage。

Producer：消息的生产者，要发送一个消息，必须通过这个生产者来发送。

MessageConsumer：与生产者相对应，这是消息的消费者或接收者，通过它来接收一个消息。

通过与ConnectionFactory可以获得一个connection

通过connection可以获得一个session会话。

Destimation即为消息队列queue。

2.4、常见的类JMS消息服务器

2.4.1、JMS消息服务器 ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的。

主要特点：

多种语言和协议编写客户端。语言: Java, C, C++, C#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)
对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性
通过了常见J2EE服务器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的测试,其中通过JCA 1.5 resource adaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上
支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
支持通过JDBC和journal提供高速的消息持久化
从设计上保证了高性能的集群,客户端-服务器,点对点
支持Ajax
支持与Axis的整合
可以很容易得调用内嵌JMS provider,进行测试

2.4.2、分布式消息中间件 Metamorphosis

Metamorphosis (MetaQ) 是一个高性能、高可用、可扩展的分布式消息中间件，类似于LinkedIn的Kafka，具有消息存储顺序写、吞吐量大和支持本地和XA事务等特性，适用于大吞吐量、顺序消息、广播和日志数据传输等场景，在淘宝和支付宝有着广泛的应用，现已开源。

主要特点：

生产者、服务器和消费者都可分布
消息存储顺序写
性能极高,吞吐量大
支持消息顺序
支持本地和XA事务
客户端pull，随机读,利用sendfile系统调用，zero-copy ,批量拉数据
支持消费端事务
支持消息广播模式
支持异步发送消息
支持http协议
支持消息重试和recover
数据迁移、扩容对用户透明
消费状态保存在客户端
支持同步和异步复制两种HA
支持group commit

2.4.3、分布式消息中间件 RocketMQ

RocketMQ 是一款分布式、队列模型的消息中间件，具有以下特点：

能够保证严格的消息顺序
提供丰富的消息拉取模式
高效的订阅者水平扩展能力
实时的消息订阅机制
亿级消息堆积能力
Metaq3.0 版本改名，产品名称改为RocketMQ

2.4.4、其他MQ

.NET消息中间件 DotNetMQ
基于HBase的消息队列 HQueue
Go 的 MQ 框架 KiteQ
AMQP消息服务器 RabbitMQ
MemcacheQ 是一个基于 MemcacheDB 的消息队列服务器。

3、为什么需要消息队列（重要）

消息系统的核心作用就是三点：解耦，异步和并行。

以用户注册的案列来说明消息系统的作用

3.1、用户注册的一般流程

问题：随着后端流程越来越多，每步流程都需要额外的耗费很多时间，从而会导致用户更长的等待延迟。

3.2、用户注册的并行执行

问题：系统并行的发起了4个请求，4个请求中，如果某一个环节执行1分钟，其他环节再快，用户也需要等待1分钟。如果其中一个环节异常之后，整个服务挂掉了。

3.3、用户注册的最终一致

保证主流程的正常执行、执行成功之后，发送MQ消息出去。
需要这个destination的其他系统通过消费数据再执行，最终一致。

4、Kafka核心组件

Topic ：消息根据Topic进行归类
Producer：发送消息者
Consumer：消息接受者
broker：每个kafka实例(server)
Zookeeper：依赖集群保存meta信息。

5、Kafka集群部署

5.1、集群部署的基本流程

下载安装包、解压安装包、修改配置文件、分发安装包、启动集群

5.2、集群部署的基础环境准备

安装前的准备工作（zk集群已经部署完毕）

1、关闭防火墙

chkconfig iptables off && setenforce 0

2、创建用户

groupadd realtime &&　useradd realtime　&& usermod -a -G realtime realtime

3、创建工作目录并赋权

mkdir /export

mkdir /export/servers

chmod 755 -R /export

4、切换到realtime用户下

su realtime

5.3 Kafka集群部署

5.3.1、下载安装包

http://kafka.apache.org/downloads.html

在linux中使用wget命令下载安装包

wget http://mirrors.hust.edu.cn/apache/kafka/0.8.2.2/kafka_2.11-0.8.2.2.tgz

5.3.2、解压安装包

tar -zxvf /usr/apps/kafka/kafka_2.11-0.8.2.2.tgz -C /usr/apps/kafka/

cd /usr/apps/kafka/

ln -s kafka_2.11-0.8.2.2 kafka

5.3.3、修改配置文件

修改配置文件：

export KAFKA_HOME=/usr/apps/kafka/kafka_2.12-2.0.0

export PATH=$PATH:$KAFKA_HOME/bin

修改config文件：

cp /usr/apps/kafka/kafka_2.12-2.0.0/config/server.properties

/usr/apps/kafka/kafka_2.12-2.0.0/config/server.properties.bak

vi /usr/apps/kafka/kafka_2.12-2.0.0/config/server.properties

输入以下内容：

5.3.4、分发安装包

scp -r /usr/apps/kafka hdp-node-2:/export/servers

scp -r /usr/apps/kafka hdp-node-3:/export/servers

然后分别在各机器上创建软连

cd /usr/apps/kafka/

ln -s kafka_2.11-0.8.2.2 kafka

5.3.5、再次修改配置文件（重要）

依次修改各服务器上配置文件的的broker.id，分别是0,1,2不得重复。

5.3.6、启动集群

依次在各节点上启动kafka

./kafka-server-start.sh /usr/apps/kafka/kafka_2.12-2.0.0/config/server.properties &

5.4、Kafka常用操作命令

1、查看当前服务器中的所有topic

./kafka-topics.sh --list --zookeeper hdp-node-1:2181

2、创建topic

./kafka-topics.sh --create --zookeeper hdp-node-1:2181 --replication-factor 1 --partitions 3 --topic first

解释：

--partitions：分区数；

--replication-factor：备份的副本数；

3、删除topic

./kafka-topics.sh --delete --zookeeper hdp-node-1:2181 --topic first

需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启。

4、通过shell命令发送消息

./kafka-console-producer.sh --broker-list hdp-node-1:9092 --topic first

5、通过shell消费消息

./kafka-console-consumer.sh --bootstrap-server hdp-node-1:9092 --from-beginning --topic first

6、查看消费位置

./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper hdp-node-1:2181 --group testGroup

7、查看某个Topic的详情

./kafka-topics.sh --topic test --describe --zookeeper hdp-node-1:2181

6、Kafka生产者Java API

Kafka设置的文件为：

主题是test；

Serizlizer.class，kafka.serializer.stringEncoder为序列化类，如果写了自定义的序列化，可以替换掉第二个；

Broker.list中定义了kafka对应的broker对应的服务器list；

另外，配置中还有其他的可选数据配置：

7、Kafka消费者Java API

消费者可以通过建立线程池，线程池中通过迭代器取出一个个线程，然后运行。

8、Kafka整体结构图

Kafka名词解释和工作方式

1）由于每天系统产生的数据比较多，因此需要对数据进行分组：partition0、1、2、3，这几个partition分布在几个不同的服务器上；

2）生产者的数据根据生产者规则，将数据分到不同的partition上，一般会有两种方式：取模和随机；

各个组件的介绍为：

Producer ：消息生产者，就是向kafka broker发消息的客户端。Producer将消息发布到它指定的topic中，并负责决定发布到哪个分区。通常简单的由负载均衡机制随机选择分区，但也可以通过特定的分区函数选择分区。使用的更多的是第二种。。
Consumer ：消息消费者，向kafka broker取消息的客户端
Topic ：咋们可以理解为一个队列。
Consumer Group （CG）：这是kafka用来实现一个topic消息的广播（发给所有的consumer）和单播（发给任意一个consumer）的手段。一个topic可以有多个CG。topic的消息会复制（不是真的复制，是概念上的）到所有的CG，但每个partion只会把消息发给该CG中的一个consumer。如果需要实现广播，只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic。
Broker ：一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。
Partition：为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序。
Offset：kafka的存储文件都是按照offset.kafka来命名，用offset做名字的好处是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。当然the first offset就是00000000000.kafka

9、Consumer与topic关系

本质上kafka只支持Topic；

1、每个group中可以有多个consumer，每个consumer属于一个consumer group；

通常情况下，一个group中会包含多个consumer，这样不仅可以提高topic中消息的并发消费能力，而且还能提高"故障容错"性，如果group中的某个consumer失效那么其消费的partitions将会有其他consumer自动接管。

2、对于Topic中的一条特定的消息，只会被订阅此Topic的每个group中的其中一个consumer消费，此消息不会发送给一个group的多个consumer；

那么一个group中所有的consumer将会交错的消费整个Topic，每个group中consumer消息消费互相独立，我们可以认为一个group是一个"订阅"者。

3、在kafka中,一个partition中的消息只会被group中的一个consumer消费(同一时刻)；

一个Topic中的每个partitions，只会被一个"订阅者"中的一个consumer消费，不过一个consumer可以同时消费多个partitions中的消息。

4、kafka的设计原理决定,对于一个topic，同一个group中不能有多于partitions个数的consumer同时消费，否则将意味着某些consumer将无法得到消息。

kafka只能保证一个partition中的消息被某个consumer消费时是顺序的；事实上，从Topic角度来说,当有多个partitions时,消息仍不是全局有序的。

kafka如何保证数据的完全生产

ack机制：broker表示发来的数据已确认接收无误，表示数据已经保存到磁盘。

0：不等待broker返回确认消息

1：等待topic中某个partition leader保存成功的状态反馈

-1：等待topic中某个partition 所有副本都保存成功的状态反馈

broker如何保存数据

在理论环境下，broker按照顺序读写的机制，可以每秒保存600M的数据。主要通过pagecache机制，尽可能的利用当前物理机器上的空闲内存来做缓存。

当前topic所属的broker，必定有一个该topic的partition，partition是一个磁盘目录。partition的目录中有多个segment组合(index,log)

10、Kafka消息的分发

Producer客户端负责消息的分发

kafka集群中的任何一个broker都可以向producer提供metadata信息,这些metadata中包含"集群中存活的servers列表"/"partitions leader列表"等信息；
当producer获取到metadata信息之后, producer将会和Topic下所有partition leader保持socket连接；
消息由producer直接通过socket发送到broker，中间不会经过任何"路由层"，事实上，消息被路由到哪个partition上由producer客户端决定；

比如可以采用"random""key-hash""轮询"等,如果一个topic中有多个partitions,那么在producer端实现"消息均衡分发"是必要的。

在producer端的配置文件中,开发者可以指定partition路由的方式。

Producer消息发送的应答机制

设置发送数据是否需要服务端的反馈,有三个值0,1,-1；

0: producer不会等待broker发送ack；

1: 当leader接收到消息之后发送ack；

-1: 当所有的follower都同步消息成功后发送ack；

request.required.acks=0

11、Consumer的负载均衡

当一个group中,有consumer加入或者离开时,会触发partitions均衡.均衡的最终目的,是提升topic的并发消费能力，步骤如下：

假如topic1,具有如下partitions: P0,P1,P2,P3
加入group中,有如下consumer: C1,C2
首先根据partition索引号对partitions排序: P0,P1,P2,P3
根据consumer.id排序: C0,C1
计算倍数: M = [P0,P1,P2,P3].size / [C0,C1].size，本例值M=2(向上取整)
然后依次分配partitions: C0 = [P0,P1],C1=[P2,P3],即Ci = [P(i * M),P((i + 1) * M -1)]

由上面的计算，C的数量需要小于等于P的数量，如果C的数量多于P，多出来的C的个数是空闲的，主要是分配时难以分配partition给到C。

partition如何分布在不同的broker上：

int i = 0

list{kafka01,kafka02,kafka03}

for(int i=0;i<5;i++){

brIndex = i%broker;

hostName = list.get(brIndex)

}

发布消息通常有两种模式：队列模式（queuing）和发布-订阅模式(publish-subscribe)。队列模式中，consumers可以同时从服务端读取消息，每个消息只被其中一个consumer读到；发布-订阅模式中消息被广播到所有的consumer中。Consumers可以加入一个consumer 组，共同竞争一个topic，topic中的消息将被分发到组中的一个成员中。同一组中的consumer可以在不同的程序中，也可以在不同的机器上。如果所有的consumer都在一个组中，这就成为了传统的队列模式，在各consumer中实现负载均衡。如果所有的consumer都不在不同的组中，这就成为了发布-订阅模式，所有的消息都被分发到所有的consumer中。更常见的是，每个topic都有若干数量的consumer组，每个组都是一个逻辑上的“订阅者”，为了容错和更好的稳定性，每个组由若干consumer组成。这其实就是一个发布-订阅模式，只不过订阅者是个组而不是单个consumer。

12、Kafka核心特性

12.1、压缩

我们上面已经知道了Kafka支持以集合（batch）为单位发送消息，在此基础上，Kafka还支持对消息集合进行压缩，Producer端可以通过GZIP或Snappy格式对消息集合进行压缩。Producer端进行压缩之后，在Consumer端需进行解压。压缩的好处就是减少传输的数据量，减轻对网络传输的压力，在对大数据处理上，瓶颈往往体现在网络上而不是CPU（压缩和解压会耗掉部分CPU资源）。
那么如何区分消息是压缩的还是未压缩的呢，Kafka在消息头部添加了一个描述压缩属性字节，这个字节的后两位表示消息的压缩采用的编码，如果后两位为0，则表示消息未被压缩。

12.2、消息可靠性

在消息系统中，保证消息在生产和消费过程中的可靠性是十分重要的，在实际消息传递过程中，可能会出现如下三中情况：

一个消息发送失败

一个消息被发送多次

最理想的情况：exactly-once ,一个消息发送成功且仅发送了一次

有许多系统声称它们实现了exactly-once，但是它们其实忽略了生产者或消费者在生产和消费过程中有可能失败的情况。比如虽然一个Producer成功发送一个消息，但是消息在发送途中丢失，或者成功发送到broker，也被consumer成功取走，但是这个consumer在处理取过来的消息时失败了。
从Producer端看：Kafka是这么处理的，当一个消息被发送后，Producer会等待broker成功接收到消息的反馈（可通过参数控制等待时间），如果消息在途中丢失或是其中一个broker挂掉，Producer会重新发送（我们知道Kafka有备份机制，可以通过参数控制是否等待所有备份节点都收到消息）。
从Consumer端看：前面讲到过partition，broker端记录了partition中的一个offset值，这个值指向Consumer下一个即将消费message。当Consumer收到了消息，但却在处理过程中挂掉，此时Consumer可以通过这个offset值重新找到上一个消息再进行处理。Consumer还有权限控制这个offset值，对持久化到broker端的消息做任意处理。

12.3、备份机制

备份机制是Kafka0.8版本的新特性，备份机制的出现大大提高了Kafka集群的可靠性、稳定性。有了备份机制后，Kafka允许集群中的节点挂掉后而不影响整个集群工作。一个备份数量为n的集群允许n-1个节点失败。在所有备份节点中，有一个节点作为lead节点，这个节点保存了其它备份节点列表，并维持各个备份间的状体同步。下面这幅图解释了Kafka的备份机制:

12.4、Kafka高效性相关设计

12.4.1 消息的持久化

Kafka高度依赖文件系统来存储和缓存消息，一般的人认为磁盘是缓慢的，这导致人们对持久化结构具有竞争性持怀疑态度。其实，磁盘远比你想象的要快或者慢，这决定于我们如何使用磁盘。
一个和磁盘性能有关的关键事实是：磁盘驱动器的吞吐量跟寻到延迟是相背离的，也就是所，线性写的速度远远大于随机写。比如：在一个6 7200rpm SATA RAID-5 的磁盘阵列上线性写的速度大概是600M/秒，但是随机写的速度只有100K/秒，两者相差将近6000倍。线性读写在大多数应用场景下是可以预测的，因此，操作系统利用read-ahead和write-behind技术来从大的数据块中预取数据，或者将多个逻辑上的写操作组合成一个大写物理写操作中。更多的讨论可以在ACMQueueArtical中找到，他们发现，对磁盘的线性读在有些情况下可以比内存的随机访问要快一些。
为了补偿这个性能上的分歧，现代操作系统都会把空闲的内存用作磁盘缓存，尽管在内存回收的时候会有一点性能上的代价。所有的磁盘读写操作会在这个统一的缓存上进行。
此外，如果我们是在JVM的基础上构建的，熟悉java内存应用管理的人应该清楚以下两件事情：

一个对象的内存消耗是非常高的，经常是所存数据的两倍或者更多。

随着堆内数据的增多，Java的垃圾回收会变得非常昂贵。

基于这些事实，利用文件系统并且依靠页缓存比维护一个内存缓存或者其他结构要好——我们至少要使得可用的缓存加倍，通过自动访问可用内存，并且通过存储更紧凑的字节结构而不是一个对象，这将有可能再次加倍。这么做的结果就是在一台32GB的机器上，如果不考虑GC惩罚，将最多有28-30GB的缓存。此外，这些缓存将会一直存在即使服务重启，然而进程内缓存需要在内存中重构（10GB缓存需要花费10分钟）或者它需要一个完全冷缓存启动（非常差的初始化性能）。它同时也简化了代码，因为现在所有的维护缓存和文件系统之间内聚的逻辑都在操作系统内部了，这使得这样做比one-off in-process attempts更加高效与准确。如果你的磁盘应用更加倾向于顺序读取，那么read-ahead在每次磁盘读取中实际上获取到这人缓存中的有用数据。
以上这些建议了一个简单的设计：不同于维护尽可能多的内存缓存并且在需要的时候刷新到文件系统中，我们换一种思路。所有的数据不需要调用刷新程序，而是立刻将它写到一个持久化的日志中。事实上，这仅仅意味着，数据将被传输到内核页缓存中并稍后被刷新。我们可以增加一个配置项以让系统的用户来控制数据在什么时候被刷新到物理硬盘上。

12.4.2 常数时间性能保证

消息系统中持久化数据结构的设计通常是维护者一个和消费队列有关的B树或者其它能够随机存取结构的元数据信息。B树是一个很好的结构，可以用在事务型与非事务型的语义中。但是它需要一个很高的花费，尽管B树的操作需要O(logN)。通常情况下，这被认为与常数时间等价，但这对磁盘操作来说是不对的。磁盘寻道一次需要10ms，并且一次只能寻一个，因此并行化是受限的。
直觉上来讲，一个持久化的队列可以构建在对一个文件的读和追加上，就像一般情况下的日志解决方案。尽管和B树相比，这种结构不能支持丰富的语义，但是它有一个优点，所有的操作都是常数时间，并且读写之间不会相互阻塞。这种设计具有极大的性能优势：最终系统性能和数据大小完全无关，服务器可以充分利用廉价的硬盘来提供高效的消息服务。
事实上还有一点，磁盘空间的无限增大而不影响性能这点，意味着我们可以提供一般消息系统无法提供的特性。比如说，消息被消费后不是立马被删除，我们可以将这些消息保留一段相对比较长的时间（比如一个星期）。

12.4.3、Kafka高性能吞吐揭秘：

Kafka作为时下最流行的开源消息系统，被广泛地应用在数据缓冲、异步通信、汇集日志、系统解耦等方面。相比较于RocketMQ等其他常见消息系统，Kafka在保障了大部分功能特性的同时，还提供了超一流的读写性能。

不同于Redis和MemcacheQ等内存消息队列，Kafka的设计是把所有的Message都要写入速度低容量大的硬盘，以此来换取更强的存储能力。实际上，Kafka使用硬盘并没有带来过多的性能损失，“规规矩矩”的抄了一条“近道”。

首先，说“规规矩矩”是因为Kafka在磁盘上只做Sequence I/O，由于消息系统读写的特殊性，这并不存在什么问题。关于磁盘I/O的性能，引用一组Kafka官方给出的测试数据(Raid-5，7200rpm)：

Sequence I/O: 600MB/s

Random I/O: 100KB/s

所以通过只做Sequence I/O的限制，规避了磁盘访问速度低下对性能可能造成的影响。

接下来我们再聊一聊Kafka是如何“抄近道的”。

首先，Kafka重度依赖底层操作系统提供的PageCache功能。当上层有写操作时，操作系统只是将数据写入PageCache，同时标记Page属性为Dirty。当读操作发生时，先从PageCache中查找，如果发生缺页才进行磁盘调度，最终返回需要的数据。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用。同时如果有其他进程申请内存，回收PageCache的代价又很小，所以现代的OS都支持PageCache。

使用PageCache功能同时可以避免在JVM内部缓存数据，JVM为我们提供了强大的GC能力，同时也引入了一些问题不适用与Kafka的设计。

如果在Heap内管理缓存，JVM的GC线程会频繁扫描Heap空间，带来不必要的开销。如果Heap过大，执行一次Full GC对系统的可用性来说将是极大的挑战。
所有在JVM内的对象都不免带有一个Object Overhead(千万不可小视)，内存的有效空间利用率会因此降低。
所有的In-Process Cache在OS中都有一份同样的PageCache。所以通过将缓存只放在PageCache，可以至少让可用缓存空间翻倍。
如果Kafka重启，所有的In-Process Cache都会失效，而OS管理的PageCache依然可以继续使用。

PageCache还只是第一步，Kafka为了进一步的优化性能还采用了Sendfile技术。在解释Sendfile之前，首先介绍一下传统的网络I/O操作流程，大体上分为以下4步。

OS 从硬盘把数据读到内核区的PageCache。
用户进程把数据从内核区Copy到用户区。
然后用户进程再把数据写入到Socket，数据流入内核区的Socket Buffer上。
OS 再把数据从Buffer中Copy到网卡的Buffer上，这样完成一次发送。

整个过程共经历两次Context Switch，四次System Call。同一份数据在内核Buffer与用户Buffer之间重复拷贝，效率低下。其中2、3两步没有必要，完全可以直接在内核区完成数据拷贝。这也正是Sendfile所解决的问题，经过Sendfile优化后，整个I/O过程就变成了下面这个样子。

通过以上的介绍不难看出，Kafka的设计初衷是尽一切努力在内存中完成数据交换，无论是对外作为一整个消息系统，或是内部同底层操作系统的交互。如果Producer和Consumer之间生产和消费进度上配合得当，完全可以实现数据交换零I/O。这也就是我为什么说Kafka使用“硬盘”并没有带来过多性能损失的原因。下面是我在生产环境中采到的一些指标。

(20 Brokers, 75 Partitions per Broker, 110k msg/s)

此时的集群只有写，没有读操作。10M/s左右的Send的流量是Partition之间进行Replicate而产生的。从recv和writ的速率比较可以看出，写盘是使用Asynchronous+Batch的方式，底层OS可能还会进行磁盘写顺序优化。而在有Read Request进来的时候分为两种情况，第一种是内存中完成数据交换。

Send流量从平均10M/s增加到了到平均60M/s，而磁盘Read只有不超过50KB/s。PageCache降低磁盘I/O效果非常明显。

接下来是读一些收到了一段时间，已经从内存中被换出刷写到磁盘上的老数据。

其他指标还是老样子，而磁盘Read已经飚高到40+MB/s。此时全部的数据都已经是走硬盘了(对硬盘的顺序读取OS层会进行Prefill PageCache的优化)。依然没有任何性能问题。

Tips：

Kafka官方并不建议通过Broker端的log.flush.interval.messages和log.flush.interval.ms来强制写盘，认为数据的可靠性应该通过Replica来保证，而强制Flush数据到磁盘会对整体性能产生影响。
可以通过调整/proc/sys/vm/dirty_background_ratio和/proc/sys/vm/dirty_ratio来调优性能。
脏页率超过第一个指标会启动pdflush开始Flush Dirty PageCache。
脏页率超过第二个指标会阻塞所有的写操作来进行Flush。
根据不同的业务需求可以适当的降低dirty_background_ratio和提高dirty_ratio。

Partition：

Partition是Kafka可以很好的横向扩展和提供高并发处理以及实现Replication的基础。

1）扩展性方面：首先，Kafka允许Partition在集群内的Broker之间任意移动，以此来均衡可能存在的数据倾斜问题。其次，Partition支持自定义的分区算法，例如可以将同一个Key的所有消息都路由到同一个Partition上去。同时Leader也可以在In-Sync的Replica中迁移。由于针对某一个Partition的所有读写请求都是只由Leader来处理，所以Kafka会尽量把Leader均匀的分散到集群的各个节点上，以免造成网络流量过于集中。

2）并发方面：任意Partition在某一个时刻只能被一个Consumer Group内的一个Consumer消费(反过来一个Consumer则可以同时消费多个Partition)，Kafka非常简洁的Offset机制最小化了Broker和Consumer之间的交互，这使Kafka并不会像同类其他消息队列一样，随着下游Consumer数目的增加而成比例的降低性能。此外，如果多个Consumer恰巧都是消费时间序上很相近的数据，可以达到很高的PageCache命中率，因而Kafka可以非常高效的支持高并发读操作，实践中基本可以达到单机网卡上限。

不过，Partition的数量并不是越多越好，Partition的数量越多，平均到每一个Broker上的数量也就越多。考虑到Broker宕机(Network Failure, Full GC)的情况下，需要由Controller来为所有宕机的Broker上的所有Partition重新选举Leader，假设每个Partition的选举消耗10ms，如果Broker上有500个Partition，那么在进行选举的5s的时间里，对上述Partition的读写操作都会触发LeaderNotAvailableException。

再进一步，如果挂掉的Broker是整个集群的Controller，那么首先要进行的是重新任命一个Broker作为Controller。新任命的Controller要从Zookeeper上获取所有Partition的Meta信息，获取每个信息大概3-5ms，那么如果有10000个Partition这个时间就会达到30s-50s。而且不要忘记这只是重新启动一个Controller花费的时间，在这基础上还要再加上前面说的选举Leader的时间 -_-!!!!!!

此外，在Broker端，对Producer和Consumer都使用了Buffer机制。其中Buffer的大小是统一配置的，数量则与Partition个数相同。如果Partition个数过多，会导致Producer和Consumer的Buffer内存占用过大。

Tips

Partition的数量尽量提前预分配，虽然可以在后期动态增加Partition，但是会冒着可能破坏Message Key和Partition之间对应关系的风险。
Replica的数量不要过多，如果条件允许尽量把Replica集合内的Partition分别调整到不同的Rack。
尽一切努力保证每次停Broker时都可以Clean Shutdown，否则问题就不仅仅是恢复服务所需时间长，还可能出现数据损坏或其他很诡异的问题。

Producer

Kafka的研发团队表示在0.8版本里用Java重写了整个Producer，据说性能有了很大提升。我还没有亲自对比试用过，这里就不做数据对比了。本文结尾的扩展阅读里提到了一套我认为比较好的对照组，有兴趣的同学可以尝试一下。

其实在Producer端的优化大部分消息系统采取的方式都比较单一，无非也就化零为整、同步变异步这么几种。

Kafka系统默认支持MessageSet，把多条Message自动地打成一个Group后发送出去，均摊后拉低了每次通信的RTT。而且在组织MessageSet的同时，还可以把数据重新排序，从爆发流式的随机写入优化成较为平稳的线性写入。

此外，还要着重介绍的一点是，Producer支持End-to-End的压缩。数据在本地压缩后放到网络上传输，在Broker一般不解压(除非指定要Deep-Iteration)，直至消息被Consume之后在客户端解压。

当然用户也可以选择自己在应用层上做压缩和解压的工作(毕竟Kafka目前支持的压缩算法有限，只有GZIP和Snappy)，不过这样做反而会意外的降低效率！！！！ Kafka的End-to-End压缩与MessageSet配合在一起工作效果最佳，上面的做法直接割裂了两者间联系。至于道理其实很简单，压缩算法中一条基本的原理“重复的数据量越多，压缩比越高”。无关于消息体的内容，无关于消息体的数量，大多数情况下输入数据量大一些会取得更好的压缩比。

不过Kafka采用MessageSet也导致在可用性上一定程度的妥协。每次发送数据时，Producer都是send()之后就认为已经发送出去了，但其实大多数情况下消息还在内存的MessageSet当中，尚未发送到网络，这时候如果Producer挂掉，那就会出现丢数据的情况。

为了解决这个问题，Kafka在0.8版本的设计借鉴了网络当中的ack机制。如果对性能要求较高，又能在一定程度上允许Message的丢失，那就可以设置request.required.acks=0 来关闭ack，以全速发送。如果需要对发送的消息进行确认，就需要设置request.required.acks为1或-1，那么1和-1又有什么区别呢？这里又要提到前面聊的有关Replica数量问题。如果配置为1，表示消息只需要被Leader接收并确认即可，其他的Replica可以进行异步拉取无需立即进行确认，在保证可靠性的同时又不会把效率拉得很低。如果设置为-1，表示消息要Commit到该Partition的ISR集合中的所有Replica后，才可以返回ack，消息的发送会更安全，而整个过程的延迟会随着Replica的数量正比增长，这里就需要根据不同的需求做相应的优化。

Tips

Producer的线程不要配置过多，尤其是在Mirror或者Migration中使用的时候，会加剧目标集群Partition消息乱序的情况(如果你的应用场景对消息顺序很敏感的话)。
0.8版本的request.required.acks默认是0(同0.7)。

Consumer

Consumer端的设计大体上还算是比较常规的。

通过Consumer Group，可以支持生产者消费者和队列访问两种模式。
Consumer API分为High level和Low level两种。前一种重度依赖Zookeeper，所以性能差一些且不自由，但是超省心。第二种不依赖Zookeeper服务，无论从自由度和性能上都有更好的表现，但是所有的异常(Leader迁移、Offset越界、Broker宕机等)和Offset的维护都需要自行处理。
大家可以关注下不日发布的0.9 Release。开发人员又用Java重写了一套Consumer。把两套API合并在一起，同时去掉了对Zookeeper的依赖。据说性能有大幅度提升哦

13、kafka文件存储机制

13.1、Kafka文件存储基本结构

Kafka的存储结构为：

现阶段这些数据会自动存储在多个服务器上。

在Kafka文件存储中，同一个topic下有多个不同partition，每个partition为一个目录，partiton命名规则为topic名称+有序序号，第一个partiton序号从0开始，序号最大值为partitions数量减1。
每个partition(目录，partition的数量是指一个服务器上有多少个文件分区)相当于一个巨型文件被平均分配到多个大小相等segment(段)数据文件中。但每个段segment file消息数量不一定相等，这种特性方便old segment file快速被删除。默认保留7天的数据。

每个partiton只需要支持顺序读写就行了，segment文件生命周期由服务端配置参数决定。（什么时候创建，什么时候删除）

数据有序的讨论？

一个partition的数据是否是有序的？间隔性有序，不连续

针对一个topic里面的数据，只能做到partition内部有序，不能做到全局有序。

特别加入消费者的场景后，如何保证消费者消费的数据全局有序的？伪命题。

只有一种情况下才能保证全局有序？就是只有一个partition。

13.2、Kafka Partition Segment

Segment file组成：由2大部分组成，分别为index file和data file，此2个文件一一对应，成对出现，后缀".index"和“.log”分别表示为segment索引文件、数据文件。

Segment文件命名规则：partion全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值。数值最大为64位long大小，19位数字字符长度，没有数字用0填充。

索引文件存储大量元数据，数据文件存储大量消息，索引文件中元数据指向对应数据文件中message的物理偏移地址。

3，497：当前log文件中的第几条信息，存放在磁盘上的那个地方
上述图中索引文件存储大量元数据，数据文件存储大量消息，索引文件中元数据指向对应数据文件中message的物理偏移地址。
其中以索引文件中元数据3,497为例，依次在数据文件中表示第3个message(在全局partiton表示第368772个message)、以及该消息的物理偏移地址为497。
segment data file由许多message组成， qq物理结构如下：

13.3、Kafka 查找message

读取offset=368776的message，需要通过下面2个步骤查找。

13.3.1、查找segment file

00000000000000000000.index表示最开始的文件，起始偏移量(offset)为0

00000000000000368769.index的消息量起始偏移量为368770 = 368769 + 1

00000000000000737337.index的起始偏移量为737338=737337 + 1

其他后续文件依次类推。

以起始偏移量命名并排序这些文件，只要根据offset **二分查找**文件列表，就可以快速定位到具体文件。当offset=368776时定位到00000000000000368769.index和对应log文件。

13.3.2、通过segment file查找message

当offset=368776时，依次定位到00000000000000368769.index的元数据物理位置和00000000000000368769.log的物理偏移地址

然后再通过00000000000000368769.log顺序查找直到offset=368776为止。

14、Kafka自定义Partition

package cn.itcast;

import kafka.javaapi.producer.Producer;

import kafka.producer.KeyedMessage;

import kafka.producer.ProducerConfig;

import java.util.Properties;

public class MyKafkaProducer {

  public static void main(String[] args) {

    Properties properties = new Properties();

    properties.put("metadata.broker.list","mini1:9092");

    // 默认的序列化为byte改为string

    properties.put("serializer.class","kafka.serializer.StringEncoder");

/**

   * 自定义parition的基本步骤

   * 1、实现partition类

   * 2、加一个构造器，MyPartitioner(VerifiableProperties properties)

   * 3、将自定义的parititoner加入到properties中

   * properties.put("partitioner.class","cn.itcast.MyPartitioner")

   * 4、producer.send方法中必须指定一个paritionKey

*/

   properties.put("partitioner.class","cn.itcast.MyPartitioner");

   Producer producer = new Producer(new ProducerConfig(properties));

   while (true){

      producer.send(new KeyedMessage("order4","zhang","我爱我的祖国"));

      // producer.send(new KeyedMessage("order","我爱我的祖国"));

package cn.itcast;

import kafka.producer.Partitioner;

import kafka.utils.VerifiableProperties;

/**

* http://www.cnblogs.com/wxd0108/p/6519973.html

*/

public class MyPartitioner implements Partitioner {

  public MyPartitioner(VerifiableProperties properties) {

  public int partition(Object key, int numPartitions) {

  return 2;

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理

标签：

相关文章推荐

新的分享

章节导航