数据库路由中间件MyCat - 源代码篇（5）
3. 连接模块
如之前所述，MyCat的连接分为前端和后端，下面是连接基本相关类图：
3.1 ClosableConnection：
代码语言：javascript复制public interface ClosableConnection {
String getCharset();
//关闭连接
void close(String reason);
boolean isClosed();
public void idleCheck();
long getStartupTime();
String getHost();
int getPort();
int getLocalPort();
long getNetInBytes();
long getNetOutBytes();
}根据字面意思，一个可以关闭的连接需要实现关闭方法-_-，并且需要原因判断是否是正常关闭。MySQL的通信都需要指定字符集。MyCat服务器建立ServerSocket时输入的端口为服务器在其上面监听客户的连接,当有客户连接时,在随机选择一个没用的端口与客户端通信;建立客户socket时输入的为服务端的监听端口,在本地选择一个未用端口与服务器通信,至于服务器怎么知道和客户端的哪个端口通信,和客户端怎么知道和服务端的哪个端口通信(因为这两个端口都是随机生成的),tcp是采用”三次握手”建立连接,而udp则是每次发送信息时将端口号放在ip报文的数据段里面。所以，连接里面需要提供获得监听端口和服务端口的方法。此外，还需要检查连接是否为空闲状态（idle）。最后，需要一些统计数据。
3.2 NIOConnection：
代码语言：javascript复制public interface NIOConnection extends ClosableConnection {
//connected
void register() throws IOException;
//处理数据
void handle(byte[] data);
// 写出一块缓冲数据
void write(ByteBuffer buffer);
}所有NIO的通信需要在多路复用选择器上注册channel，这里有个对应的register()方法需要实现。然后，读取和写入数据都需要通过缓冲。缓冲区(Buffer)就是在内存中预留指定大小的存储空间用来对输入/输出(I/O)的数据作临时存储，这部分预留的内存空间就叫做缓冲区，使用缓冲区有这么两个好处： 1. 减少实际的物理读写次数 2. 缓冲区在创建时就被分配内存，这块内存区域一直被重用，可以减少动态分配和回收内存的次数 读取到的数据需要经过处理，这里对应的就是handle(byte[])方法。
3.3 AbstractConnection：
从上面的实体图，我们发现，AbstractConnection其实就是把Java的NetworkChannel进行封装，同时需要依赖其他几个类来完成他所需要的操作，如下：
其中，NIOProcessor是对AbstractConnection实现NIO读写的方法类，NIOHandler是处理AbstractConnection读取的数据的处理方法类，NIOSocketWR是执行以上方法的线程类。
3.3.1 NIOProcessor：
NIOProcessor的构建方法：
代码语言：javascript复制public NIOProcessor(String name, BufferPool bufferPool,
NameableExecutor executor) throws IOException {
this.name = name;
this.bufferPool = bufferPool;
this.executor = executor;
this.frontends = new ConcurrentHashMap();
this.backends = new ConcurrentHashMap();
this.commands = new CommandCount();
}调用位置：
MyCatServer.java
代码语言：javascript复制...
bufferPool = new BufferPool(processBuferPool, processBufferChunk,
socketBufferLocalPercent / processorCount);
businessExecutor = ExecutorUtil.create("BusinessExecutor",
threadPoolSize);
...
for (int i = 0; i < processors.length; i ) {
processors[i] = new NIOProcessor("Processor" i, bufferPool,
businessExecutor);
}
...每个MyCat实例会初始化processors个NIOProcessor，每个NIOProcessor公用同一个bufferPool和businessExecutor。 bufferPool是缓冲池，BufferPool这个类负责缓冲统一管理 businessExecutor如之前所述，是业务线程池。 NIOProcessor被池化，很简单，就是保存到数组中，通过MyCatServer的nextProcessor()方法轮询获取一个NIOProcessor，之后每个AbstractConnection通过setNIOProcessor方法，设置NIOProcessor。
代码语言：javascript复制public NIOProcessor nextProcessor() {
int i = nextProcessor;
if (i >= processors.length) {
i = nextProcessor = 0;
}
return processors[i];
}可以看出，每个AbstractConnection依赖于一个NIOProcessor，每个NIOProcessor保存着多个AbstractConnection。AbstractConnection分为FrontendConnection和BackendConnection被分别保存在NIOProcessor的frontends和backends这两个ConcurrentHashMap中。 用ConcurrentHashMap是因为NIOAcceptor和NIOConnector线程以及RW线程池都会访问这两个变量。 NIOProcessor其实主要负责连接资源的管理： * MyCat会定时检查前端和后端空闲连接，并清理和回收资源：* MyCatServer.java:
代码语言：javascript复制// 处理器定时检查任务
private TimerTask processorCheck() {
return new TimerTask() {
@Override
public void run() {
timerExecutor.execute(new Runnable() {
@Override
public void run() {
try {
for (NIOProcessor p : processors) {
p.checkBackendCons();
}
} catch (Exception e) {
LOGGER.warn("checkBackendCons caught err:" e);
}
}
});
timerExecutor.execute(new Runnable() {
@Override
public void run() {
try {
for (NIOProcessor p : processors) {
p.checkFrontCons();
}
} catch (Exception e) {
LOGGER.warn("checkFrontCons caught err:" e);
}
}
});
}
};
}检查前端连接，回收空闲资源：
代码语言：javascript复制 /**
* 定时执行该方法，回收部分资源。
*/
public void checkFrontCons() {
frontendCheck();
}
private void frontendCheck() {
Iterator> it = frontends.entrySet()
.iterator();
while (it.hasNext()) {
FrontendConnection c = it.next().getValue();
// 删除空连接
if (c == null) {
it.remove();
this.frontendsLength.decrementAndGet();
continue;
}
// 清理已关闭连接，否则空闲检查。
if (c.isClosed()) {
c.cleanup();
it.remove();
this.frontendsLength.decrementAndGet();
} else {
// very important ,for some data maybe not sent
checkConSendQueue(c);
c.idleCheck();
}
}
}在关闭前端连接时，会清理连接占用的缓存资源： FrontendConnection.java：
代码语言：javascript复制protected void cleanup() {
//回收读缓冲
if (readBuffer != null) {
recycle(readBuffer);
this.readBuffer = null;
this.readBufferOffset = 0;
}
//回收写缓冲
if (writeBuffer != null) {
recycle(writeBuffer);
this.writeBuffer = null;
}
//回收压缩协议栈编码解码队列
if(!decompressUnfinishedDataQueue.isEmpty())
{
decompressUnfinishedDataQueue.clear();
}
if(!compressUnfinishedDataQueue.isEmpty())
{
compressUnfinishedDataQueue.clear();
}
//回收写队列
ByteBuffer buffer = null;
while ((buffer = writeQueue.poll()) != null) {
recycle(buffer);
}
}后端连接检查，除了要清理已关闭的连接，还有要检查SQL执行时间是否超时：
代码语言：javascript复制 /**
* 定时执行该方法，回收部分资源。
*/
public void checkBackendCons() {
backendCheck();
}
// 后端连接检查
private void backendCheck() {
long sqlTimeout = MycatServer.getInstance().getConfig().getSystem().getSqlExecuteTimeout() * 1000L;
Iterator> it = backends.entrySet().iterator();
while (it.hasNext()) {
BackendConnection c = it.next().getValue();
// 删除空连接
if (c == null) {
it.remove();
continue;
}
// SQL执行超时的连接关闭
if (c.isBorrowed()
&& c.getLastTime() < TimeUtil.currentTimeMillis()
- sqlTimeout) {
LOGGER.warn("found backend connection SQL timeout ,close it "
c);
c.close("sql timeout");
}
// 清理已关闭连接，否则空闲检查。
if (c.isClosed()) {
it.remove();
} else {
// very important ,for some data maybe not sent
if (c instanceof AbstractConnection) {
checkConSendQueue((AbstractConnection) c);
}
c.idleCheck();
}
}
}同时，在检查连接是否关闭时，需要检查写队列是否为空。写队列不为空，证明还有请求没有响应。需要将写队列的剩余请求异步写出，通过NIOSocketWR。
代码语言：javascript复制 private void checkConSendQueue(AbstractConnection c) {
// very important ,for some data maybe not sent
if (!c.writeQueue.isEmpty()) {
c.getSocketWR().doNextWriteCheck();
}
}
