上篇我们了解到服务启动过程中，ProviderBootStrap.init()和ProviderBootStrap.startup(providerConfig);分别启动http协议服务和default（TCP）协议的netty服务。本篇我们分析下端口监听和消息处理。
我们以ProviderBootStrap.startup(providerConfig)来切入TCP协议NettyServer实现。
关键代码如下：
public static ServerConfig startup(ProviderConfig<?> providerConfig) {
ServerConfig serverConfig = providerConfig.getServerConfig();
synchronized (ProviderBootStrap.class) {
//Server接口两个实现：NettyServer和JettyHttpServer
List<Server> servers = ExtensionLoader.newExtensionList(Server.class);
for (Server s : servers) {
if (!s.isStarted()) {
//NettyServer实现支持default协议
if (s.support(serverConfig)) {
//调用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) {
//预启动请求处理线程池，ThreadPoolExcutor.prestartAllCoreThreads()
server.getRequestProcessor().getRequestProcessThreadPool().prestartAllCoreThreads();
return server.getServerConfig();
}
return null;
}
}
跟进AbstractServer.start
public void start(ServerConfig serverConfig) {
//请求处理线程池RequestThreadPoolProcessor
requestProcessor = RequestProcessorFactory.selectProcessor();
//调用NettyServer.doStart
doStart(serverConfig);
if (requestProcessor != null) {
//根据上篇ServerConfig的线程池参数配置创建DynamicThreadPool线程池
//底层实现为new ThreadPoolExecutor(...)
requestProcessor.start(serverConfig);
}
this.serverConfig = serverConfig;
}
接下来我们看下NettyServer实现。
该pigeon版本基于Netty3.9.2.Final，本篇不对netty服务做详细说明，感兴趣的朋友可以阅读下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;
}
因为本篇在于介绍消息处理，我们重点看最后一个处理器NettyServerHandler，跟踪代码如下：
//NettyServerHandler.messageReceived
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent message) {
CodecEvent codecEvent = (CodecEvent) (message.getMessage());
//请求信息
InvocationRequest request = (InvocationRequest) codecEvent.getInvocation();
ProviderContext invocationContext = new DefaultProviderContext(request, new NettyServerChannel(ctx.getChannel()));
try {
//请求处理
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) {
//最终调用RequestThreadPoolProcessor.doProcessRequest
return requestProcessor.processRequest(request, providerContext);
}
RequestThreadPoolProcessor.doProcessRequest就是我们所有关注的最终的业务处理逻辑，几点说明如下：
1、将业务处理逻辑封装在Callable任务中；
2、将Callable任务提交到线程池执行；
3、ProviderProcessHandlerFactory中将不同消息类型对应的Filter集合以ServiceInvocationHandler封装成责任链模式的消息处理器链
代码如下：
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 {
//根据消息类型获取对应的Filter责任链
ServiceInvocationHandler invocationHandler = ProviderProcessHandlerFactory
.selectInvocationHandler(providerContext.getRequest().getMessageType());
if (invocationHandler != null) {
providerContext.setThread(Thread.currentThread());
//执行责任链
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));
//提交线程池执行
return pool.submit(requestExecutor);
} catch (RejectedExecutionException e) {
requestContextMap.remove(request);
throw new RejectedException(getProcessorStatistics(pool), e);
}
}
关于ProviderProcessHandlerFactory中不同消息类型对应的Filter集合如下：
public static void init() {
//业务消息
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);
}
以业务消息对应的Filter集合为例子，来看下责任链的封装以及Filter的执行顺序
//Filter逆序封装，顺序执行
//在ServiceInvocationHandler.handle中执行filter.invoke，在filter.invoke中执行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;
}
关于TraceFilter、MonitorProcessFilter、ExceptionProcessFilter等这里限于篇幅不具体分析代码，这些Filter负责的功能从Filter命名可以猜个七七八八。我们重点看看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的方法映射关系
method = ServiceMethodFactory.getMethod(request);
}
。。。
Object returnObj = null;
try {
//基于反射调用
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!");
}
可以看到我们根据请求信息InvocationRequest从ServiceMethodFactory找到对应的service bean方法，然后基于反射执行。
而ServiceMethodFactory在上篇s服务初始化过程中缓存了每一个service bean的方法映射关系。
至此我们理清了TCP协议NettyServer的消息处理机制。
关于http协议的JettyHttpServer实现不再赘述了，感兴趣的朋友可以自行了解。
网友评论