分布式对象存储Ambry（4）Ambry-Server模块源代码解析（启动与整体通信工作篇）

Ambry一共有三个主模块：Ambry-Server，Ambry-Frontend还有Ambry-Admin。其中，Ambry-Server为其核心，我们从核心开始，一部一部剖析其源码，并会分析它的设计模式。
一个提供各种服务的服务器框架源代码，我们一般从它的Facade类开始入手进行第一步分析。Facade类就是指设计模式中的外观模式的核心类，这个类会包含这个框架几乎所有的模块。一般的，像Ambry这种服务器型框架，都会在整体设计模式上采用单例、桥接还有外观设计模式结合，我们寻找到这个单例Facade类，就能发现这个系统中的所有模块。
Ambry的主要Facade类就是是AmbryServer这个类，它的成员包括：



其中：

shutdownLatch：用于关闭AmbryServer，这是一种比较巧妙的运用CountDownLatch来实现Server启动和优雅退出的方式。

networkServer：这个类管理NIO通信连接还有工作调度分配方式，包括NIO通信，各种工作线程池，调度流程等等

requests：这个类保存着处理requests的方法，对于不同的request如何处理，就在这个类里面。

requestHandlerPool：执行上面处理requests的方法的线程池

scheduler：定时任务调度线程池

storeManager：存储与持久化对象管理类

replicationManager：备份管理类

properties：用于读取配置验证的工具类

clusterMap：集群拓扑管理类

registry，reporter：监控指标类

connectionPool：连接池，管理所有连接

notificationSystem：下游提醒系统

metrics：server监控指标

time：Ambry特定监控时间类

程序启动和优雅退出的关键设计 - shutdownLatch

Ambry运用了Java中的CountDownlatch来实现启动和优雅退出。首先，在AmbryServer启动之后（所有配置载入完毕，各种资源，比如线程池等等，初始化完毕），主线程会在这个shutdownLatch上执行await()方法。
同时，主线程会使用Runtime.getRuntime().addShutdownHook(shutdownHook);这个方法增加一个钩子来监听程序关闭事件，例如(kill -9，ctrl+c)：

Runtime.getRuntime().addShutdownHook(new Thread() {
public void run() {
logger.info("Received shutdown signal. Shutting down AmbryServer");
ambryServer.shutdown();
}
});

主线程主要操作是调用AmbryServer的启动还有等待方法：

     //载入相关配置
final InvocationOptions options = new InvocationOptions(args);
final Properties properties = Utils.loadProps(options.serverPropsFilePath);
final VerifiableProperties verifiableProperties = new VerifiableProperties(properties);
final ClusterMap clusterMap =
new ClusterMapManager(options.hardwareLayoutFilePath, options.partitionLayoutFilePath,
new ClusterMapConfig(verifiableProperties));
logger
3ff0
.info("Bootstrapping AmbryServer");
//利用配置新建AmbryServer
ambryServer = new AmbryServer(verifiableProperties, clusterMap, SystemTime.getInstance());
// 增加一个钩子来监听程序关闭事件，例如(kill -9，ctrl+c)
Runtime.getRuntime().addShutdownHook(new Thread() {
public void run() {
logger.info("Received shutdown signal. Shutting down AmbryServer");
ambryServer.shutdown();
}
});
//启动AmbryServer
ambryServer.startup();
//主线程进入阻塞状态
ambryServer.awaitShutdown();

这里的

ambryServer.startup();

是启动方法，

ambryServer.awaitShutdown();

是主线程进入阻塞状态。这个方法代码：

public void awaitShutdown() throws InterruptedException {
shutdownLatch.await();
}

在检测到ctrl+c或者kill信号时，

ambryServer.shutdown();

会被执行，这个方法关闭各个模块，清理资源，在方法的最后：

try {
//关闭各个模块，清理资源
} catch (Exception e) {
logger.error("Error while shutting down server", e);
} finally {
shutdownLatch.countDown();
long processingTime = SystemTime.getInstance().milliseconds() - startTime;
metrics.serverShutdownTimeInMs.update(processingTime);
logger.info("Server shutdown time in Ms " + processingTime);
}

shutdownLatch.countDown();

总会被执行，这时shutdownLatch归零，主线程阻塞停止，由于主线程接下来没有代码，程序结束。

整个Server的通信设计 - networkServer

之前Ambry在这里是BIO设计，如今改成了NIO设计。
未来可能还会设计成AIO的模式，所以在这里networkServer是一个接口。
这里我们类比下其他服务器框架的NIO通信设计（其实都是一种Reactor模式的设计）：

MyCat的NIO通信模块：

这个图有点不准确，在连接完成后，其实client的request还有response就是和NIOReactor交互，这里忘了画了。。。
简单来说，就是，client在与server通信时。Server有一个NIOAcceptor线程在监听client连接事件（通过NIOAcceptor的Selctor，这个Selector就是Java NIO中的核心，负责监听多个SocketChannel的四种事件），有新的client连接时，接受连接，之后将这个SocketChannel连接转交给NIOReactorPool这个线程池中的一个线程（其实就是在这个线程上的Selector注册读写事件监听），这个线程处理链接发来的请求，并把一些重工作（例如结果集合并与复杂结果集处理等等）交给BusinessExecutor这个线程池执行，这样可以增加整体的client处理响应吞吐量。

Netty的NIO通信模块：

类似地，Netty的核心机制也是差不多这样，Boss线程池负责监听client连接事件，有client连接时，处理连接，之后将连接转交给WORK线程池处理来自于client的请求，WORK一般会做一些请求与响应的编码解码等轻量级的工作，为了不影响整体吞吐量，将耗时的业务操作（例如访问数据库等）转交给EXECUTOR线程池去做。

Ambry的NIO通信模块：

Ambry的和之前的大同小异：

首先Client连接，触发OP_ACCPET

Ambry有一个Acceptor线程监听OP_ACCEPT事件，连接完成，将连接转交给Processors

之后每个Processor监听读写事件，形成Client发送Request，Processor返回Response的效果

Processor其实只负责将接收到的请求封装成Request，交给RequestHandler处理
同时接收Response，翻译后发送给Client

真正处理Request的是RequestHandler

我们来看下Acceptor还有Processor的源代码来加深下理解：



Acceptor还有Processor都继承AbstractServerThread，这个抽象类就是一个抽象线程类。包含startupLatch，shutdownLatch还有alive；
startupLatch还有shutdownLatch和其他地方的CountDownlatch类似，一个负责等待该线程启动初始化完成，一个负责等待该线程优雅关闭完成。alive是一个AtomicBoolean类型，通过这个类型还有名字我们很容易猜出它是表示这个线程是否还存活，对于监控很有用。

public AbstractServerThread() throws IOException {
startupLatch = new CountDownLatch(1);
shutdownLatch = new CountDownLatch(1);
alive = new AtomicBoolean(false);
}

/** * Initiates a graceful shutdown by signaling to stop and waiting for the shutdown to complete */
public void shutdown() throws InterruptedException {
alive.set(false);
shutdownLatch.await();
}

/** * Wait for the thread to completely start up */
public void awaitStartup() throws InterruptedException {
startupLatch.await();
}

/** * Record that the thread startup is complete */
protected void startupComplete() {
alive.set(true);
startupLatch.countDown();
}

/** * Record that the thread shutdown is complete */
protected void shutdownComplete() {
shutdownLatch.countDown();
}

/** * Is the server still running? */
protected boolean isRunning() {
return alive.get();
}

这里内容不少了，先小结个尾，接下来的分析会在下一篇文章中展示，敬请期待。