作者简介
路路，热爱技术、乐于分享的技术人，目前主要从事数据库相关技术的研究。
前言
本文主要分为两部分，一是内存管理模块结构，二是内存管理模块源码解析。
1. 网络读写
2. 查询结果集处理
简单说明一下这两点：
第一点比较好理解，网络读的时候需要从通道中读取字节流到内存中，以便进一步处理，网络写的时候也需要将字节流从内存中写入到通道中，以便发送信息。
第二点就比较复杂一些了，DBLE 获取到 MySQL 端的数据后，会经过 DBLE 然后再发送给客户端，如果是简单查询的话，DBLE 端只是起到一个中转的作用，会利用少量内存用于将结果转发给客户端。如果是复杂查询的话，DBLE 会先对结果集进行处理，比如连接、分组排序等操作，然后再将结果发送给客户端，这里涉及到的内存使用就复杂了，对于复杂查询 DBLE 会先使用堆内存，如果使用的堆内存超过限制，则会写内存映射文件，如果写内存映射文件个数再次达到上限，则会写硬盘。是不是很复杂？放心，这一块本文不会介绍，原因我就不说了（太复杂了）……
我们看一下 DBLE 官方内存结构图：
个人觉得上图有一点理解上的疑惑， DirectByteBufferPool 占用的内存应该位于物理内存中，JVM 中只是有着堆外内存的引用对象而已。
下图个人觉得可能更好理解。
结合 DBLE 管理内存的主要两个作用，网络读写主要用到了堆外内存，即 DirectByteBufferPool 管理的部分。结果集处理根据情况有所不同，本文只讨论简单情况，对于简单情况来讲，会使用到堆外内存或堆内存，下文源码中会提到具体情况。
DirectByteBufferPool 类即为管理堆外内存的内存池类，可以看到它创建了 ByteBuffePage 类，并且与该类有着一对多的关系；ByteBufferPage 类为 DBLE 抽象出的堆外内存页的概念，内存页中又有内存块的概念，内存块是内存分配的最小单元。采用此种内存概念主要是为了减少内存碎片问题。DBLE 官方对于这两个类的内部结构以图的形式表示的很清楚了：
上图中的参数 bufferPoolPageNumber、bufferPoolPageSize 和 bufferPoolChunkSize 可在 Server.xml 中配置 ，bufferPoolPageSize 默认为 2M，bufferPoolPageNumber 默认为 Java 虚拟机的可用的处理器数量 * 20， bufferPoolChunkSize 的参数默认为 4k，该参数最好设为 bufferPoolPageSize 的约数，否则最后一个会造成浪费。
1. 如果不指定分配大小，则默认分配一个最小单元(最小单元由 bufferPoolChunkSize 决定）；
2. 如果指定分配大小，则分配放得下分配大小的最小单元的整数倍(向上取整) ，举个例子，如果需要 10k 的大小，则在 bufferPoolChunkSize 参数默认为 4k 的情况下，会分配 3 个内存 chunk；
bufferPoolPageNumber -1 页(上次分配过的记为第 N 页)，然后对单页加锁在每个页中从头寻找未被使用的连续 M 个最小单元 ；（4）如果找不到可存放的单页(比如大于 bufferPoolPageSize )，直接分配 On-Heap 内存。
内存池初始化的代码调用在 DbleServer#startup 方法中：
//通过配置的相关参数初始化内存池
bufferPool = new DirectByteBufferPool(bufferPoolPageSize, bufferPoolChunkSize, bufferPoolPageNumber);
DirectByteBufferPool 类的构造方法中：
public DirectByteBufferPool(int pageSize, short chunkSize, short pageCount) {
//初始化内存总页数
allPages = new ByteBufferPage[pageCount];
//内存页中的内存块大小
this.chunkSize = chunkSize;
//每个内存页的大小
this.pageSize = pageSize;
//内存页总数
this.pageCount = pageCount;
//记录上次分配过的页
prevAllocatedPage = new AtomicInteger(0);
//循环初始化内存页
for (int i = 0; i < pageCount; i++) {
allPages[i] = new ByteBufferPage(ByteBuffer.allocateDirect(pageSize), chunkSize);
}
//记录堆外内存的使用大小
memoryUsage = new ConcurrentHashMap<>();
}
ByteBufferPage 类的构造方法中：
public ByteBufferPage(ByteBuffer buf, int chunkSize) {
//内存块大小
this.chunkSize = chunkSize;
//计算总内存块数
chunkCount = buf.capacity() / chunkSize;
//非常巧妙的位图结构，用于标记内存块的使用情况
chunkAllocateTrack = new BitSet(chunkCount);
//内存页的大小
this.buf = buf;
}
上述代码完成了堆外内存的初始化。
DirectByteBufferPool#allocate 方法中：
public ByteBuffer allocate(int size) {
//计算需要分配的chunk数
final int theChunkCount = size / chunkSize + (size % chunkSize == 0 ? 0 : 1);
//本次分配的开始页数，为上次分配过的页数（N）+1
int selectedPage = prevAllocatedPage.incrementAndGet() % allPages.length;
//从N+1页到bufferPoolPageNumber-1页遍历分配，allocateBuffer方法下面会进一步分析
ByteBuffer byteBuf = allocateBuffer(theChunkCount, selectedPage, allPages.length);
//没有分配成功，则从0页到N页继续遍历分配
if (byteBuf == null) {
byteBuf = allocateBuffer(theChunkCount, 0, selectedPage);
}
final long threadId = Thread.currentThread().getId();
//分配成功的话，则记录分配到的内存大小
if (byteBuf != null) {
if (memoryUsage.containsKey(threadId)) {
memoryUsage.put(threadId, memoryUsage.get(threadId) + byteBuf.capacity());
} else {
memoryUsage.put(threadId, (long) byteBuf.capacity());
}
}
//如果遍历完所有页还没有分配成功，则直接分配堆上内存
if (byteBuf == null) {
//ByteBuffer.allocate方法为分配JVM堆内存
return ByteBuffer.allocate(size);
}
return byteBuf;
}
继续看下 DirectByteBufferPool#allocateBuffer 方法：
private ByteBuffer allocateBuffer(int theChunkCount, int startPage, int endPage) {
//从指定页数开始遍历分配内存，分配成功则标记当前分配的页数，然后直接返回
for (int i = startPage; i < endPage; i++) {
//调用了ByteBufferPage类的allocateChunk方法进行内存块的分配，该方法下面也会进一步分析
ByteBuffer buffer = allPages[i].allocateChunk(theChunkCount);
if (buffer != null) {
prevAllocatedPage.getAndSet(i);
return buffer;
}
}
return null;
}
ByteBufferPage#allocateChunk 方法，该方法比较长，分配页中的连续内存块逻辑就在此方法中：
public ByteBuffer allocateChunk(int theChunkCount) {
//对页加状态锁，防止并发操作异常
if (!allocLockStatus.compareAndSet(false, true)) {
return null;
}
int startChunk = -1;
int continueCount = 0;
try {
//下面的逻辑为在页中找到符合内存分配大小的连续内存块
for (int i = 0; i < chunkCount; i++) {
if (!chunkAllocateTrack.get(i)) {
if (startChunk == -1) {
startChunk = i;
continueCount = 1;
if (theChunkCount == 1) {
break;
}
} else {
if (++continueCount == theChunkCount) {
break;
}
}
} else {
startChunk = -1;
continueCount = 0;
}
}
//成功找到后则返回分配的内存块，并且标记相应的内存块位置
if (continueCount == theChunkCount) {
int offStart = startChunk * chunkSize;
int offEnd = offStart + theChunkCount * chunkSize;
buf.limit(offEnd);
buf.position(offStart);
ByteBuffer newBuf = buf.slice();
markChunksUsed(startChunk, theChunkCount);
return newBuf;
} else {
//分配失败返回null
return null;
}
} finally {
//解锁
allocLockStatus.set(false);
}
}
到这里，内存分配的逻辑就大概讲完了。
DirectByteBufferPool#recycle 方法中：
public void recycle(ByteBuffer theBuf) {
//堆内存的回收
if (!(theBuf instanceof DirectBuffer)) {
theBuf.clear();
return;
}
final long size = theBuf.capacity();
//堆外内存的回收
boolean recycled = false;
DirectBuffer thisNavBuf = (DirectBuffer)theBuf;
int chunkCount = theBuf.capacity() / chunkSize;
DirectBuffer parentBuf = (DirectBuffer) thisNavBuf.attachment();
int startChunk = (int) ((thisNavBuf.address() - parentBuf.address()) / this.chunkSize);
//遍历页然后将对应块标记为未使用即可
for (ByteBufferPage allPage : allPages) {
if ((recycled = allPage.recycleBuffer((ByteBuffer) parentBuf, startChunk, chunkCount))) {
break;
}
}
final long threadId = Thread.currentThread().getId();
if (memoryUsage.containsKey(threadId)) {
memoryUsage.put(threadId, memoryUsage.get(threadId) - size);
}
if (!recycled) {
LOGGER.info("warning ,not recycled buffer " + theBuf);
}
}
内存的分配与回收到这里也讲完了，还记得开始时候说的内存的主要作用吗？一个是网络读写时使用，一个是结果集暂存时使用。
网络读写时使用的例子可以参考这里，NIOSocketWR#asyncRead 方法：
public void asyncRead() throws IOException {
ByteBuffer theBuffer = con.readBuffer;
if (theBuffer == null) {
//这里分配了从通道读取数据时候的内存
theBuffer = con.processor.getBufferPool().allocate(con.processor.getBufferPool().getChunkSize());
con.readBuffer = theBuffer;
}
int got = channel.read(theBuffer);
con.onReadData(got);
}
暂存结果集使用的例子可以参考这里，SingleNodeHandler#rowResponse 方法：
//省略无关内容
......
//这里分配内存，将结果集写入，然后待发送到客户端
buffer = session.getSource().writeToBuffer(row, allocBuffer());
......
当然内存分配的代码调用远不止上面举例的两处，这里只是和内存管理的作用做个呼应。其他地方大家可以自己看。
社区近期动态
10.26 DBLE 用户见面会 北京站
爱可生开源社区将在 2019 年 10 月 26 日迎来在北京的首场 DBLE 用户见面会，以线下互动分享的会议形式跟大家见面。
时间：10月26日 9:00 – 12:00 AM
地点：HomeCafe 上地店（北京市海淀区上地二街一号龙泉湖酒店对面）
重要提醒：
1. 同日下午还有 dbaplus 社群举办的沙龙：聚焦数据中台、数据架构与优化。
2. 爱可生开源社区会在每年10.24日开源一款高质量产品。本次在 dbaplus 沙龙会议上，爱可生的资深研发工程师闫阿龙，将为大家带来《金融分布式事务实践及txle概述》，并在现场开源。
「3306π」成都站 Meetup
知数堂将在 2019 年 10 月 26 日在成都举办线下会议，本次会议中邀请了五位数据库领域的资深研发/DBA进行主题分享。
时间：2019年10月26日 13:00-18:00
地点：成都市高新区天府三街198号腾讯成都大厦A座多功能厅
Mycat 问题免费诊断
诊断范围支持：
Mycat 的故障诊断、源码分析、性能优化
服务支持渠道：
技术交流群，进群后可提问
QQ群（669663113）
社区通道，邮件&电话
osc@actionsky.com
现场拜访，线下实地，1天免费拜访
关注“爱可生开源社区”公众号，回复关键字“Mycat”，获取活动详情。
社区技术内容征稿
征稿内容：
格式：.md/.doc/.txt
主题：MySQL、分布式中间件DBLE、数据传输组件DTLE相关技术内容
要求：原创且未发布过
奖励：作者署名；200元京东E卡+社区周边
投稿方式：
邮箱：osc@actionsky.com
格式：[投稿]姓名+文章标题
以附件形式发送，正文需注明姓名、手机号、微信号，以便小编及时联系