Kafka作为一个支持大数据量写入写出的消息队列,由于是基于Scala和Java实现的,而Scala和Java均需要在JVM上运行,所以如果是基于内存的方式,即JVM的堆来进行数据存储则需要开辟很大的堆来支持数据读写,从而会导致GC频繁影响性能。考虑到这些因素,kafka是使用磁盘存储数据的。
Kafka 中消息是以 topic 进行分类的,生产者生产消息,消费者消费消息,都是面向topic的。topic存储结构见下图:
由于生产者生产的消息会不断追加到 log 文件末尾,为防止 log 文件过大导致数据定位效率低下,Kafka 采取了 分片 和 索引 机制,将每个partition分为多个segment。每个 segment对应两个文件——“.index”文件和“.log”文件。
partition文件夹命名规则:
topic 名称+分区序号,举例有一个topic名称文“kafka”,这个topic有三个分区,则每个文件夹命名如下:
index和log文件的命名规则:
1)partition文件夹中的第一个segment从0开始,以后每个segement文件以上一个segment文件的最后一条消息的offset+1命名(当前日志中的第一条消息的offset值命名)。
2)数值最大为64位long大小。19位数字字符长度,没有数字用0填充。
举例,有以下三对文件:
以第二个文件为例看下对应的数据结构:
稀疏索引 需要注意下。
消息查找过程 :
找message-2589,即offset为2589:
1)先定位segment文件,在0000000000000002584中。
2)计算查找的offset在日志文件的相对偏移量
offset - 文件名的数量 = 2589 - 2584 = 5;
在index文件查找第一个参数的值,若找到,则获取到偏移量,通过偏移量到log文件去找对应偏移量的数据即可;
本例中没有找到,则找到当前索引中偏移量的上线最接近的值,即3,偏移量文246;然后到log文件中从偏移量为246数据开始向下寻找。
简单了解了kafka在数据存储方面的知识,线面我们具体分析下为什么kafka基于磁盘却快于内存。
在前面了解存储结构过程中,我们发现kafka记录log日志使用的结尾追加的方式,即 顺序写 。这样要比随机写块很多,这与磁盘的机械机构有关,顺序写之所以快,是因为其省去了大量磁头寻址的时间。
mmap,简单描述其就是将磁盘文件映射到内存, 用户通过修改内存就能修改磁盘文件。
即便是顺序写磁盘,磁盘的读写速度任然比内存慢慢的多得多,好在 *** 作系统已经帮我们解决这个问题。在Linux *** 作系统中,Linux会将磁盘中的一些数据读取到内存当中,我们称之为内存页。当需要读写硬盘的时候,都优先在内存页中进行处理。当内存页的数据比硬盘数据多的时候,就形成了 脏页 ,当脏页达到一定数量, *** 作系统会进行 刷脏 ,即将内存也数据写到磁盘。
问题:不可靠,写到 mmap 中的数据并没有被真正的写到硬盘, *** 作系统会在程序主动调用 Flush 的时候才把数据真正的写到硬盘。
零拷贝并不是不需要拷贝,而是减少不必要的拷贝次数,通常使用在IO读写过程中。
传统io过程
如上图所示,上图共经历了四次拷贝的过程:
1)数据到到内核态的read buffer
2)内核态的read buffer到用户态应用层的buffer;
3)用户态到内核态的socket buffer;
4)socket buffer到网卡的buffer(NIC)。
DMA
引入DMA技术,是指外部设备不通过CPU而直接与系统内存交换数据的接口技术,网卡等硬件设备支持DMA技术。
如上图所示,上图共经历了两次拷贝的过程。
sendfile
在内核版本 2.1 中,引入了 Sendfile 系统调用,以简化网络上和两个本地文件之间的数据传输。同时使用了DMA技术。
如上图所示,上图共经历了一次拷贝的过程。
sendfile( DMA 收集拷贝)
之前我们是把页缓存的数据拷贝到socket缓存中,实际上,我们仅仅需要把缓冲区描述符传到 socket 缓冲区,再把数据长度传过去,这样 DMA 控制器直接将页缓存中的数据打包发送到网络中就可以了。
如上图所示,最后一次的拷贝也被消除了,数据->read buffer->NIC。
kafka通过java和scala实现,而Java对sendfile是通过NIO 的 FileChannel (java.nio.channels.FileChannel )的 transferTo 和 transferFrom 方法实现零拷贝
注: transferTo 和 transferFrom 并不保证一定能使用零拷贝。实际上是否能使用零拷贝与 *** 作系统相关,如果 *** 作系统提供 sendfile 这样的零拷贝系统调用,则这两个方法会通过这样的系统调用充分利用零拷贝的优势,否则并不能通过这两个方法本身实现零拷贝。
一、Kafka的三大组件:Producer、Server、Consumer
1、Kafka的Producer写入消息
producer采用push(推)模式将消息发布到broker,每条消息,都被追加到分区中(顺序写到磁盘,比随机写内存效率高)。
· 分区的作用:方便容量扩展,可以多并发读写数据,所以我们会指定多个分区进行数据存储。
· 一般根据 event_key的hash % numPartitions来确定写入哪个分区,如果写入时没有指定key,则轮询写入每个分区;因此导致每个partition中消息是有序的,整体无序。
每条event数据写入partitionA中,并且只会写入partitionA_leader,当partitionA_leader写入完成后partitionA_flower节点再去partitionA_leader上异步拉取数据;默认ack为1,表示不会等待partitionA_flowers写入完成;如果设置ack为副本数或ack=-1,则等待副本全部写完,再写入下一条数据。
2、kafka的broker——保存消息
1、 创建topic,并指定分区和副本数
2、每个分区(partition)有一个leader,多个follower,pull数据时先寻找leader,只会读leader上的数据,leader和follower不会在一个节点上,leader节点宕机后,其中一个follower变成leader
3、 消息数据存在每个分区中,默认配置每条消息保存7天 或 分区达到1GB 后删除数据
3、Kafka的Consumer消费数据:
1、consumer采用pull(拉)模式从broker中读取数据。
2、如果一个消费者来消费同一个topic下不同分区的数据,会读完一个分区再读下一个分区
生产者(producer)A PI 只有一套 ; 但是消费者(consumer)A PI 有两套(高级A PI 和低级A PI )
一、高级API:
Zookeeper管理offset(默认从最后一个开始读新数据,可以配置从开头读)
kafka server(kafka服务)管理分区、副本
二、低级API:
开发者自己控制offset,想从哪里读就从哪里读
// SimpleConsumer是Kafka用来读数据的类
// 通过send()方法获取元数据找到leader
TopicMetadataResponse metadataResponse = simpleConsumer.send(request) //通过metadataResponse获取topic元数据,在获取topic中每个分区的元数据
// fetch 抓取数据
FetchResponse response = simpleConsumer.fetch(fetchRequest)
// 解析抓取到的数据
ByteBufferMessageSet messageAndOffsets = response.messageSet(topic, partition)
二、数据、broker状态,consumer状态的存储
一、在本地存储原始消息数据:
1、hash取模得分区、kafka中每条消息有一个Key,用来确定 每条数据存储到哪个分区中
2、轮询
3、自定义分区
二、在zookeeper存储kafka的元数据
三、存储consumer的offset数据
每个consumer有一个Key(broker+Topic+partition)的hash,再取模后 用来确定offset存到哪个系统文件中,Value是partitionMetaData。
1、使用zookeeper启动,zookeeper来存储offset
消费者 消费消息时,offset(消费到的下标)会保存在consumer本地和zookeeper中(由本地上传到zookeeper中,所以本地会保存offset)
2、使用bootstrap启动,本地存储offset(在本地可以减少两节点交互),zookeeper存储其他数据
三、某Flume对接Kafka案例
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)