上一节，我们对blockmanager的主要写入方法做了一个整理，知道了blockmananger的主要写入逻辑，以及对于块信息的管理。但是，由于spark的整个存储模块是在是很庞大，而且很多细节的逻辑错综复杂，如果对于每个细节都刨根问底，一来精力有限，二来感觉也没有太大的必要，当然如果时间允许肯定是越详细越好，在这里，我的分析的主要目的是理清存储模块的重点逻辑，希望能够提纲契领地把各个模块的脉络领出来，建立起对spark-core中各模块的整体认知，这样我们在遇到一些问题的时候就能够很快地知道应该从何处下手，从哪个具体的模块去找问题。
好了废话不多说，本节接着上一节。上一篇，我们分析了blockmanager的几个主要的存储方法，发现blockmanager主要依靠内部的两个组件memorystore和diskstore来进行实际的数据写入和块的管理。
本节，我们就来看一下memorystore这个组件。
不过，我还是延续我一贯的风格，从外部对一个类的方法调用为切入点分析这个类的作用和逻辑。
所以，我们先来看一下上一节对于memorystore的主要的方法调用的总结：
memorystore.putiteratorasvalues memorystore.putiteratorasbytes memorystore.putbytes
这个方法主要是用于存储级别是非序列化的情况，即直接以java对象的形式将数据存放在jvm堆内存上。我们都知道，在jvm堆内存上存放大量的对象并不是什么好事，gc压力大，挤占内存，可能引起频繁的gc，但是也有明显的好处，就是省去了序列化和反序列化耗时，而且直接从堆内存取数据显然比任何其他方式（磁盘和直接内存）都要快很多，所以对于内存充足且要缓存的数据量本省不是很大的情况，这种方式也不失为一种不错的选择。
我要评论private[storage] def putiteratorasvalues[t](
blockid: blockid,
values: iterator[t],
classtag: classtag[t]): either[partiallyunrollediterator[t], long] = {
// 用于存储java对象的容器
val valuesholder = new deserializedvaluesholder[t](classtag)
putiterator(blockid, values, classtag, memorymode.on_heap, valuesholder) match {
// 存储成功
case right(storedsize) => right(storedsize)
// 存储失败的情况
case left(unrollmemoryusedbythisblock) =>
// valuesholder内部的数组和vector会相互转换
// 数据写入完成后会将vector中的数据转移到数组中
val unrollediterator = if (valuesholder.vector != null) {
valuesholder.vector.iterator
} else {
valuesholder.arrayvalues.toiterator
}
// 返回写入一半的迭代器、
// 外部调用者一半会选择关闭这个迭代器以释放被使用的内存
left(new partiallyunrollediterator(
this,
memorymode.on_heap,
unrollmemoryusedbythisblock,
unrolled = unrollediterator,
rest = values))
}
}
这个方法的逻辑很简单，作用也比较单一，主要是对实际存储方法putiterator的返回结果做处理，如果失败的话，就封装一个partiallyunrollediterator返回给外部调用这个，调用这个一般需要将这个写入一半的迭代器关闭。
这个方法看似很长，其实逻辑相对简单，主要做的事就是把数据一条一条往valuesholder中写，并周期性地检查内存，如果内存不够就通过内存管理器memorymanager申请内存，每次申请当前内存量的1.5倍。
最后，将valuesholder中的数据转移到一个数组中(其实数据在sizetrackingvector中也是以数组的形式存储，只不过sizetrackingvector对象内部处理数组还有一些其他的簿记量，更为关键的是我们需要将存储的数据以同一的接口进行包装，以利于memorystore进行同一管理)。最后还有关键的一步，就是释放展开内存，重新申请存储内存。
此外，这个过程中有使用到memorymanager，具体的方法调用是：
memorymanager.acquireunrollmemory(blockid, memory, memorymode)
------------------------------分割线------------------------------
private def putiterator[t](
blockid: blockid,
values: iterator[t],
classtag: classtag[t],
memorymode: memorymode,
valuesholder: valuesholder[t]): either[long, long] = {
require(!contains(blockid), s"block $blockid is already present in the memorystore")
// number of elements unrolled so far
var elementsunrolled = 0
// whether there is still enough memory for us to continue unrolling this block
var keepunrolling = true
// initial per-task memory to request for unrolling blocks (bytes).
// 用于数据在内存展开的初始的内存使用量
val initialmemorythreshold = unrollmemorythreshold
// how often to check whether we need to request more memory
// 检查内存的频率，每写这么多条数据就会检查一次是否需要申请额外的内存
val memorycheckperiod = conf.get(unroll_memory_check_period)
// memory currently reserved by this task for this particular unrolling operation
// 内存阈值，开始时等于初始阈值
var memorythreshold = initialmemorythreshold
// memory to request as a multiple of current vector size
// 内存增长因子，每次申请的内存是当前内存的这个倍数
val memorygrowthfactor = conf.get(unroll_memory_growth_factor)
// keep track of unroll memory used by this particular block / putiterator() operation
// 当前的块使用的内存大小
var unrollmemoryusedbythisblock = 0l
// request enough memory to begin unrolling
// 首先进行初始的内存申请,向memorymanager申请内存
keepunrolling =
reserveunrollmemoryforthistask(blockid, initialmemorythreshold, memorymode)
if (!keepunrolling) {
logwarning(s"failed to reserve initial memory threshold of " +
s"${utils.bytestostring(initialmemorythreshold)} for computing block $blockid in memory.")
} else {
// 如果成功申请到内存，则累加记录
unrollmemoryusedbythisblock += initialmemorythreshold
}
// unroll this block safely, checking whether we have exceeded our threshold periodically
// 循环将每条数据写入容器中valuesholder
while (values.hasnext && keepunrolling) {
valuesholder.storevalue(values.next())
// 如果写入数据的条数达到一个周期，那么就检查一下是否需要申请额外的内存
if (elementsunrolled % memorycheckperiod == 0) {
// 通过valuesholder获取已经写入的数据的评估大小
// 注意，这里的数据大小只是估计值，并不是十分准确
// 具体如何进行估算的可以看valuesholder内部实现
val currentsize = valuesholder.estimatedsize()
// if our vector's size has exceeded the threshold, request more memory
// 如果已写入的数据大小超过了当前阈值
if (currentsize >= memorythreshold) {
// 这里每次申请的内存量都是不一样的
// 每次申请的内存是当前已使用内存的1.5倍（默认）
val amounttorequest = (currentsize * memorygrowthfactor - memorythreshold).tolong
keepunrolling =
reserveunrollmemoryforthistask(blockid, amounttorequest, memorymode)
if (keepunrolling) {
// 记录累积申请的内存量
unrollmemoryusedbythisblock += amounttorequest
}
// new threshold is currentsize * memorygrowthfactor
// 目前已经向内存管理器申请的内存量
memorythreshold += amounttorequest
}
}
// 记录插入的数据条数
elementsunrolled += 1
}
// make sure that we have enough memory to store the block. by this point, it is possible that
// the block's actual memory usage has exceeded the unroll memory by a small amount, so we
// perform one final call to attempt to allocate additional memory if necessary.
// 如果keepunrolling为true，说明顺利地将所有数据插入，
// 并未遇到申请内存失败的情况
if (keepunrolling) {
// 将内部的数据转移到一个数组中
val entrybuilder = valuesholder.getbuilder()
// 数据在内存中的精确大小
val size = entrybuilder.precisesize
// 实际的大小可能大于申请的内存量
// 因此根据实际大小还要再申请额外的内存
if (size > unrollmemoryusedbythisblock) {
val amounttorequest = size - unrollmemoryusedbythisblock
keepunrolling = reserveunrollmemoryforthistask(blockid, amounttorequest, memorymode)
if (keepunrolling) {
unrollmemoryusedbythisblock += amounttorequest
}
}
if (keepunrolling) {
// 获取memoryentry对象，该对象是对插入数据的包装
val entry = entrybuilder.build()
// synchronize so that transfer is atomic
memorymanager.synchronized {
// 这一步主要是释放申请的展开内存
// 然后申请存储内存
// 这里需要弄清楚展开内存的概念
// 展开状态指的是对象在内存中处于一种比较松散的状态，这样的状态方便做一些管理如统计大小等
// 而随后将对象转移到数组中，处于一种比较紧实的状态，数组相对来说占用的额外内存是比较小的
// 一个数组只是一个对象，只有一个对象头，可以用来管理大量的对象
releaseunrollmemoryforthistask(memorymode, unrollmemoryusedbythisblock)
// 申请存储内存
val success = memorymanager.acquirestoragememory(blockid, entry.size, memorymode)
assert(success, "transferring unroll memory to storage memory failed")
}
// 放入map中管理起来
entries.synchronized {
entries.put(blockid, entry)
}
loginfo("block %s stored as values in memory (estimated size %s, free %s)".format(blockid,
utils.bytestostring(entry.size), utils.bytestostring(maxmemory - blocksmemoryused)))
right(entry.size)
} else {
// we ran out of space while unrolling the values for this block
logunrollfailuremessage(blockid, entrybuilder.precisesize)
// 如果失败，返回已经申请的展开内存
left(unrollmemoryusedbythisblock)
}
} else {
// we ran out of space while unrolling the values for this block
logunrollfailuremessage(blockid, valuesholder.estimatedsize())
left(unrollmemoryusedbythisblock)
}
}
我们再看另一个方法。套路基本和putiteratorasvalues是一样一样的。
最大的区别在于valuesholder类型不同。非序列化形式存储使用的是deserializedmemoryentry，而序列化形式存储使用的是serializedmemoryentry。
private[storage] def putiteratorasbytes[t](
blockid: blockid,
values: iterator[t],
classtag: classtag[t],
memorymode: memorymode): either[partiallyserializedblock[t], long] = {
require(!contains(blockid), s"block $blockid is already present in the memorystore")
// initial per-task memory to request for unrolling blocks (bytes).
val initialmemorythreshold = unrollmemorythreshold
// 字节数组的块大小，默认是1m
val chunksize = if (initialmemorythreshold > int.maxvalue) {
logwarning(s"initial memory threshold of ${utils.bytestostring(initialmemorythreshold)} " +
s"is too large to be set as chunk size. chunk size has been capped to " +
s"${utils.bytestostring(int.maxvalue)}")
int.maxvalue
} else {
initialmemorythreshold.toint
}
// 字节数组的容器
val valuesholder = new serializedvaluesholder[t](blockid, chunksize, classtag,
memorymode, serializermanager)
putiterator(blockid, values, classtag, memorymode, valuesholder) match {
case right(storedsize) => right(storedsize)
case left(unrollmemoryusedbythisblock) =>
// 部分展开，部分以序列化形式存储的block
left(new partiallyserializedblock(
this,
serializermanager,
blockid,
valuesholder.serializationstream,
valuesholder.redirectablestream,
unrollmemoryusedbythisblock,
memorymode,
valuesholder.bbos,
values,
classtag))
}
}
我们再来看另一个被外部调用用来插入数据的方法。很简单，不说了。
def putbytes[t: classtag](
blockid: blockid,
size: long,
memorymode: memorymode,
_bytes: () => chunkedbytebuffer): boolean = {
require(!contains(blockid), s"block $blockid is already present in the memorystore")
// 首先向内存管理器申请内存
// 这里申请的是存储内存，因为要插入的字节数组，
// 所以不需要再展开，也就不需要申请展开内存
if (memorymanager.acquirestoragememory(blockid, size, memorymode)) {
// we acquired enough memory for the block, so go ahead and put it
val bytes = _bytes()
assert(bytes.size == size)
// 这里直接构建了一个serializedmemoryentry
// 并放到map中管理起来
val entry = new serializedmemoryentry[t](bytes, memorymode, implicitly[classtag[t]])
entries.synchronized {
entries.put(blockid, entry)
}
loginfo("block %s stored as bytes in memory (estimated size %s, free %s)".format(
blockid, utils.bytestostring(size), utils.bytestostring(maxmemory - blocksmemoryused)))
true
} else {
false
}
}
通过对上面的三个方法，其实主要是前两个方法的分析，我们发现，除了对内存进行簿记管理之外，以及通过内存管理器申请内存之外，插入数据最主要的工作其实都是有valuesholder对象来完成的。
valuesholder特质有两个实现类：deserializedvaluesholder和serializedvaluesholder。
deserializedvaluesholder对象内部有两个成员：vector，是一个sizetrackingvector；arrayvalues，是一个存放值的数组，用于在所有数据插入后，将主句转移到一个数组中，方便包装成一个memoryentry对象。大部分工作是有sizetrackingvector完成的。
private class deserializedvaluesholder[t] (classtag: classtag[t]) extends valuesholder[t] {
// underlying vector for unrolling the block
var vector = new sizetrackingvector[t]()(classtag)
var arrayvalues: array[t] = null
override def storevalue(value: t): unit = {
vector += value
}
override def estimatedsize(): long = {
vector.estimatesize()
}
override def getbuilder(): memoryentrybuilder[t] = new memoryentrybuilder[t] {
// we successfully unrolled the entirety of this block
arrayvalues = vector.toarray
vector = null
override val precisesize: long = sizeestimator.estimate(arrayvalues)
override def build(): memoryentry[t] =
deserializedmemoryentry[t](arrayvalues, precisesize, classtag)
}
}
上面提到的sizetrackingvector继承了这个特质，除了这个特质，还集成了primitivevector类，但是primitivevector类基本上就是对一个数组的简单包装。
sizetrackingvector最重要的功能：追踪对象的大小，就是在sizetracker特之中实现的。
我大致说一下这个特质是如何实现对象大小跟踪和估算的，代码实现也并不复杂，感兴趣的可以看一看，限于篇幅这里就不贴了。
可见这么做并不是什么精确，但是由于是抽样，而且抽样周期越往后面越长，所以对于数据插入的效率影响很小，而且这种不精确性其实在后续的内存检查过程中是有考虑到的。在所有数据插入完的收尾工作中，会对对象大小做一次精确计算。此外，熟悉spark内存管理的同学应该知道，其实spark一般会配置一个安全因子（一般是0.9），也就是说只是用配置的内存大小的90%，就是为了尽可能地减少这种不精确的内存估算造成oom的可能性。
private class serializedvaluesholder[t](
blockid: blockid,
chunksize: int,
classtag: classtag[t],
memorymode: memorymode,
serializermanager: serializermanager) extends valuesholder[t] {
val allocator = memorymode match {
case memorymode.on_heap => bytebuffer.allocate _
// 调用unsafe的本地方法申请直接内存
// 这个方法之所以没有调用bytebuffer.allocatedirect方法
// 是因为这个方法分配的直接内存大小收到参数maxdirectmemorysize限制
// 所以这里绕过bytebuffer.allocatedirect方法，通过反射和unsafe类创建直接内存对象
case memorymode.off_heap => platform.allocatedirectbuffer _
}
val redirectablestream = new redirectableoutputstream
val bbos = new chunkedbytebufferoutputstream(chunksize, allocator)
redirectablestream.setoutputstream(bbos)
val serializationstream: serializationstream = {
val autopick = !blockid.isinstanceof[streamblockid]
val ser = serializermanager.getserializer(classtag, autopick).newinstance()
// 包装压缩流和序列化流
ser.serializestream(serializermanager.wrapforcompression(blockid, redirectablestream))
}
// 写入方法，写入的对象经过序列化，压缩，
// 然后经过chunkedbytebufferoutputstream被分割成一个个的字节数组块
override def storevalue(value: t): unit = {
serializationstream.writeobject(value)(classtag)
}
override def estimatedsize(): long = {
bbos.size
}
override def getbuilder(): memoryentrybuilder[t] = new memoryentrybuilder[t] {
// we successfully unrolled the entirety of this block
serializationstream.close()
override def precisesize(): long = bbos.size
override def build(): memoryentry[t] =
serializedmemoryentry[t](bbos.tochunkedbytebuffer, memorymode, classtag)
}
}
大概看一下，主要的逻辑很简单，这里面有几个注意点：
memorystore.scala这个文件中乍看代码有八百多行，但是其实很大部分代码是一些辅助类，比较核心的写入逻辑也就是前面提到的几个方法，再加上核心的两个类deserializedvaluesholder和serializedvaluesholder实现了以对象或字节数组的形式存储数据。
如对本文有疑问， 点击进行留言回复！！
Spring Cloud Alibaba系列，nacos作为服务配置中心
Spring Cloud Alibaba系列（一）nacos作为服务注册中心
网友评论