上篇我們了解到服務(wù)啟動(dòng)過程中，ProviderBootStrap.init()和ProviderBootStrap.startup(providerConfig);分別啟動(dòng)http協(xié)議服務(wù)和default（TCP）協(xié)議的netty服務(wù)。本篇我們分析下端口監(jiān)聽和消息處理。
我們以ProviderBootStrap.startup(providerConfig)來切入TCP協(xié)議NettyServer實(shí)現(xiàn)砌创。
關(guān)鍵代碼如下：
public static ServerConfig startup(ProviderConfig<?> providerConfig) {
ServerConfig serverConfig = providerConfig.getServerConfig();
synchronized (ProviderBootStrap.class) {
//Server接口兩個(gè)實(shí)現(xiàn)：NettyServer和JettyHttpServer
List<Server> servers = ExtensionLoader.newExtensionList(Server.class);
for (Server s : servers) {
if (!s.isStarted()) {
//NettyServer實(shí)現(xiàn)支持default協(xié)議
if (s.support(serverConfig)) {
//調(diào)用AbstractServer.start
s.start(serverConfig);
s.addService(providerConfig);
serversMap.put(s.getProtocol() + serverConfig.getPort(), s);
logger.warn("pigeon " + s + "[version:" + VersionUtils.VERSION + "] has been started");
break;
}
}
}
server = serversMap.get(serverConfig.getProtocol() + serverConfig.getPort());
if (server != null) {
//預(yù)啟動(dòng)請(qǐng)求處理線程池，ThreadPoolExcutor.prestartAllCoreThreads()
server.getRequestProcessor().getRequestProcessThreadPool().prestartAllCoreThreads();
return server.getServerConfig();
}
return null;
}
}
跟進(jìn)AbstractServer.start
public void start(ServerConfig serverConfig) {
//請(qǐng)求處理線程池RequestThreadPoolProcessor
requestProcessor = RequestProcessorFactory.selectProcessor();
//調(diào)用NettyServer.doStart
doStart(serverConfig);
if (requestProcessor != null) {
//根據(jù)上篇ServerConfig的線程池參數(shù)配置創(chuàng)建DynamicThreadPool線程池
//底層實(shí)現(xiàn)為new ThreadPoolExecutor(...)
requestProcessor.start(serverConfig);
}
this.serverConfig = serverConfig;
}
接下來我們看下NettyServer實(shí)現(xiàn)鲫懒。
該pigeon版本基于Netty3.9.2.Final嫩实，本篇不對(duì)netty服務(wù)做詳細(xì)說明，感興趣的朋友可以閱讀下netty系列文章窥岩。
public NettyServer() {
this.bootstrap = new ServerBootstrap(channelFactory);
this.bootstrap.setPipelineFactory(new NettyServerPipelineFactory(this));
this.bootstrap.setOption("child.tcpNoDelay", true);
this.bootstrap.setOption("child.keepAlive", true);
this.bootstrap.setOption("child.reuseAddress", true);
this.bootstrap.setOption("child.connectTimeoutMillis", 1000);
}
@Override
public void doStart(ServerConfig serverConfig) {
if (!started) {
if (serverConfig.isAutoSelectPort()) {
int availablePort = getAvailablePort(serverConfig.getPort());
this.port = availablePort;
}
InetSocketAddress address = null;
if (this.ip == null) {
address = new InetSocketAddress(this.port);
} else {
address = new InetSocketAddress(this.ip, this.port);
}
channel = this.bootstrap.bind(address);
serverConfig.setActualPort(this.port);
this.started = true;
}
}
//在NettyServerPipelineFactory中定義了一些列消息處理器
public ChannelPipeline getPipeline() {
ChannelPipeline pipeline = pipeline();
pipeline.addLast("framePrepender", new FramePrepender());
pipeline.addLast("frameDecoder", new FrameDecoder());
pipeline.addLast("crc32Handler", new Crc32Handler(codecConfig));
pipeline.addLast("compressHandler", new CompressHandler(codecConfig));
pipeline.addLast("providerDecoder", new ProviderDecoder());
pipeline.addLast("providerEncoder", new ProviderEncoder());
//NettyServerHandler處理器中處理InvocationRequest
pipeline.addLast("serverHandler", new NettyServerHandler(server));
return pipeline;
}
因?yàn)楸酒谟诮榻B消息處理甲献，我們重點(diǎn)看最后一個(gè)處理器NettyServerHandler，跟蹤代碼如下：
//NettyServerHandler.messageReceived
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent message) {
CodecEvent codecEvent = (CodecEvent) (message.getMessage());
//請(qǐng)求信息
InvocationRequest request = (InvocationRequest) codecEvent.getInvocation();
ProviderContext invocationContext = new DefaultProviderContext(request, new NettyServerChannel(ctx.getChannel()));
try {
//請(qǐng)求處理
this.server.processRequest(request, invocationContext);
} catch (Throwable e) {
String msg = "process request failed:" + request;
// 心跳消息只返回正常的, 異常不返回
if (request.getCallType() == Constants.CALLTYPE_REPLY
&& request.getMessageType() != Constants.MESSAGE_TYPE_HEART) {
ctx.getChannel().write(ProviderUtils.createFailResponse(request, e));
}
}
}
@Override
public Future<InvocationResponse> processRequest(InvocationRequest request, ProviderContext providerContext) {
//最終調(diào)用RequestThreadPoolProcessor.doProcessRequest
return requestProcessor.processRequest(request, providerContext);
}
RequestThreadPoolProcessor.doProcessRequest就是我們所有關(guān)注的最終的業(yè)務(wù)處理邏輯颂翼，幾點(diǎn)說明如下：
1晃洒、將業(yè)務(wù)處理邏輯封裝在Callable任務(wù)中慨灭；
2、將Callable任務(wù)提交到線程池執(zhí)行球及；
3氧骤、ProviderProcessHandlerFactory中將不同消息類型對(duì)應(yīng)的Filter集合以ServiceInvocationHandler封裝成責(zé)任鏈模式的消息處理器鏈
代碼如下：
public Future<InvocationResponse> doProcessRequest(final InvocationRequest request,
final ProviderContext providerContext) {
requestContextMap.put(request, providerContext);
doMonitorData(request, providerContext);
Callable<InvocationResponse> requestExecutor = new Callable<InvocationResponse>() {
@Override
public InvocationResponse call() throws Exception {
providerContext.getTimeline().add(new TimePoint(TimePhase.T));
try {
//根據(jù)消息類型獲取對(duì)應(yīng)的Filter責(zé)任鏈
ServiceInvocationHandler invocationHandler = ProviderProcessHandlerFactory
.selectInvocationHandler(providerContext.getRequest().getMessageType());
if (invocationHandler != null) {
providerContext.setThread(Thread.currentThread());
//執(zhí)行責(zé)任鏈
return invocationHandler.handle(providerContext);
}
} catch (Throwable t) {
logger.error("Process request failed with invocation handler, you should never be here.", t);
} finally {
requestContextMap.remove(request);
}
return null;
}
};
final ThreadPool pool = selectThreadPool(request);
try {
checkRequest(pool, request);
providerContext.getTimeline().add(new TimePoint(TimePhase.T));
//提交線程池執(zhí)行
return pool.submit(requestExecutor);
} catch (RejectedExecutionException e) {
requestContextMap.remove(request);
throw new RejectedException(getProcessorStatistics(pool), e);
}
}
關(guān)于ProviderProcessHandlerFactory中不同消息類型對(duì)應(yīng)的Filter集合如下：
public static void init() {
//業(yè)務(wù)消息
registerBizProcessFilter(new TraceFilter());
if (Constants.MONITOR_ENABLE) {
registerBizProcessFilter(new MonitorProcessFilter());
}
registerBizProcessFilter(new WriteResponseProcessFilter());
registerBizProcessFilter(new ContextTransferProcessFilter());
registerBizProcessFilter(new ExceptionProcessFilter());
registerBizProcessFilter(new SecurityFilter());
registerBizProcessFilter(new GatewayProcessFilter());
registerBizProcessFilter(new BusinessProcessFilter());
bizInvocationHandler = createInvocationHandler(bizProcessFilters);
//心跳消息
registerHeartBeatProcessFilter(new WriteResponseProcessFilter());
registerHeartBeatProcessFilter(new HeartbeatProcessFilter());
heartBeatInvocationHandler = createInvocationHandler(heartBeatProcessFilters);
registerHealthCheckProcessFilter(new WriteResponseProcessFilter());
registerHealthCheckProcessFilter(new HealthCheckProcessFilter());
healthCheckInvocationHandler = createInvocationHandler(healthCheckProcessFilters);
registerScannerHeartBeatProcessFilter(new WriteResponseProcessFilter());
registerScannerHeartBeatProcessFilter(new ScannerHeartBeatProcessFilter());
scannerHeartBeatInvocationHandler = createInvocationHandler(scannerHeartBeatProcessFilters);
}
以業(yè)務(wù)消息對(duì)應(yīng)的Filter集合為例子，來看下責(zé)任鏈的封裝以及Filter的執(zhí)行順序
//Filter逆序封裝吃引，順序執(zhí)行
//在ServiceInvocationHandler.handle中執(zhí)行filter.invoke筹陵，在filter.invoke中執(zhí)行next.handle
private static <K, V extends ServiceInvocationFilter> ServiceInvocationHandler createInvocationHandler(
List<V> internalFilters) {
ServiceInvocationHandler last = null;
List<V> filterList = new ArrayList<V>();
filterList.addAll(internalFilters);
for (int i = filterList.size() - 1; i >= 0; i--) {
final V filter = filterList.get(i);
final ServiceInvocationHandler next = last;
last = new ServiceInvocationHandler() {
@SuppressWarnings("unchecked")
@Override
public InvocationResponse handle(InvocationContext invocationContext) throws Throwable {
return filter.invoke(next, invocationContext);
}
};
}
return last;
}
關(guān)于TraceFilter、MonitorProcessFilter镊尺、ExceptionProcessFilter等這里限于篇幅不具體分析代碼惶翻，這些Filter負(fù)責(zé)的功能從Filter命名可以猜個(gè)七七八八。我們重點(diǎn)看看BusinessProcessFilter
@Override
public InvocationResponse invoke(ServiceInvocationHandler handler, ProviderContext invocationContext)
throws Throwable {
invocationContext.getTimeline().add(new TimePoint(TimePhase.U));
InvocationRequest request = invocationContext.getRequest();
if (request.getMessageType() == Constants.MESSAGE_TYPE_SERVICE) {
鹅心。吕粗。。
InvocationResponse response = null;
ServiceMethod method = invocationContext.getServiceMethod();
if (method == null) {
//ServiceMethodFactory緩存了service bean的方法映射關(guān)系
method = ServiceMethodFactory.getMethod(request);
}
旭愧。颅筋。。
Object returnObj = null;
try {
//基于反射調(diào)用
returnObj = method.invoke(request.getParameters());
} finally {
ProviderHelper.clearContext();
}
invocationContext.getTimeline().add(new TimePoint(TimePhase.M, System.currentTimeMillis()));
if (request.getCallType() == Constants.CALLTYPE_REPLY) {
response = ProviderUtils.createSuccessResponse(request, returnObj);
}
return response;
}
throw new BadRequestException("message type[" + request.getMessageType() + "] is not supported!");
}
可以看到我們根據(jù)請(qǐng)求信息InvocationRequest從ServiceMethodFactory找到對(duì)應(yīng)的service bean方法输枯，然后基于反射執(zhí)行议泵。
而ServiceMethodFactory在上篇s服務(wù)初始化過程中緩存了每一個(gè)service bean的方法映射關(guān)系。
至此我們理清了TCP協(xié)議NettyServer的消息處理機(jī)制桃熄。
關(guān)于http協(xié)議的JettyHttpServer實(shí)現(xiàn)不再贅述了先口，感興趣的朋友可以自行了解。
轉(zhuǎn)載請(qǐng)備注原文鏈接瞳收。
