一惫皱、啟動方式
1.業(yè)務(wù)邏輯處理器定義
/**
* 自定義的業(yè)務(wù)邏輯用戶處理器
* SyncUserProcessor屬于同步，soaf bolt還提供了異步的方式AsyncUserProcessor
* 二者的區(qū)別在于旅敷，前者需要在當(dāng)前處理線程以return返回值的形式返回處理結(jié)果；而后者涂滴，有一個(gè) AsyncContext 客戶端的調(diào)用，調(diào)用 sendResponse 方法缔杉，內(nèi)部通過穿件RCMD對象搁料，可以在當(dāng)前線程，也可以在異步線程郭计，返回處理結(jié)果
* 如果一個(gè)處理器需要對多種數(shù)據(jù)模型感興趣，兩種方式沈贝，一種是使用基類的方式處理勋乾，第二種方式通過MultiInterestUserProcessor
*/
public class TestServerUserProcessorextends SyncUserProcessor {
@Override
? ? public Object handleRequest(BizContext bizCtx, TestRequest request)throws Exception {
TestResponse response =new TestResponse();
if (request !=null) {
System.out.println(request);
response.setResp("from server -> " + request.getReq());
}
return response;
}
/**
* 業(yè)務(wù)處理器感興趣的消息體，
*/
? ? @Override
? ? public String interest() {
return TestRequest.class.getName();
}
}
2.服務(wù)端啟動
//創(chuàng)建RpcServer實(shí)例学歧，指定監(jiān)聽ip和port各吨，指定是否使用鏈接管理器
RpcServer server =new RpcServer("127.0.0.1",8888,false);
//注冊業(yè)務(wù)邏輯處理器 UserProcessor
server.registerUserProcessor(new TestServerUserProcessor());
//啟動服務(wù)端
server.start();
3.server.start()原理分析
server.start()主要有兩步
=doInit();
==this.addressParser =new RpcAddressParser();//初始化地址解析器
==this.connectionEventHandler =new ConnectionEventHandler(switches());//初始化鏈接事件處理器，主要負(fù)責(zé)入站出站數(shù)據(jù)的處理
==this.connectionEventHandler.setConnectionManager(this.connectionManager);//添加初始化的連接管理器到鏈接事件處理器
==this.connectionEventHandler.setConnectionEventListener(this.connectionEventListener);//添加時(shí)間監(jiān)聽器
==initRpcRemoting();//初始化遠(yuǎn)程調(diào)用對象RpcServerRemoting
==this.bootstrap =new ServerBootstrap();
????this.bootstrap.group(bossGroup,workerGroup)
????.channel(NettyEventLoopUtil.getServerSocketChannelClass())
.? ?option(ChannelOption.SO_BACKLOG, ConfigManager.tcp_so_backlog())//tcp等待隊(duì)列大小横浑，默認(rèn)1024
????.option(ChannelOption.SO_REUSEADDR, ConfigManager.tcp_so_reuseaddr())//端口快速釋放屉更，默認(rèn)true
????.childOption(ChannelOption.TCP_NODELAY, ConfigManager.tcp_nodelay())
????.childOption(ChannelOption.SO_KEEPALIVE, ConfigManager.tcp_so_keepalive());
==initWriteBufferWaterMark();//設(shè)置出站的低水位和高水位
==if (ConfigManager.netty_buffer_pooled()) { this.bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT); } else { this.bootstrap.option(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT) .childOption(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT); }//根據(jù)配置決定是否使用對象池化
==NettyEventLoopUtil.enableTriggeredMode(bootstrap)//設(shè)置selector的epoll模式是邊緣觸發(fā)還是水平觸發(fā)瑰谜，默認(rèn)水平觸發(fā)
==this.bootstrap.childHandler(new ChannelInitializer<SocketChannel>() { @Override ????protected void initChannel(SocketChannel channel) { ChannelPipeline pipeline = ????channel.pipeline(); pipeline.addLast("decoder", codec.newDecoder()); ????pipeline.addLast("encoder", codec.newEncoder());if (idleSwitch) { ????pipeline.addLast("idleStateHandler", new IdleStateHandler(0, 0, idleTime, ????TimeUnit.MILLISECONDS));pipeline.addLast("serverIdleHandler", serverIdleHandler); } pipeline.addLast("connectionEventHandler", connectionEventHandler); pipeline.addLast("handler", rpcHandler); createConnection(channel); } /** * create connection operation<br> * <ul> * <li>If flag manageConnection be true, use {@link DefaultConnectionManager} to add a new connection, meanwhile bind it with the channel.</li> * <li>If flag manageConnection be false, just create a new connection and bind it with the channel.</li> * </ul> */ private void createConnection(SocketChannel channel) { Url url = addressParser.parse(RemotingUtil.parseRemoteAddress(channel)); if (switches().isOn(GlobalSwitch.SERVER_MANAGE_CONNECTION_SWITCH)) { connectionManager.add(new Connection(channel, url), url.getUniqueKey()); } else { new Connection(channel, url); } channel.pipeline().fireUserEventTriggered(ConnectionEventType.CONNECT); } });//添加加密萨脑、解密、心跳(原理后面講)渤早、遠(yuǎn)程請求(原理后面講)處理器，并創(chuàng)建一個(gè)和監(jiān)聽端口綁定的鏈接(業(yè)務(wù)層面的綁定)
=doStart();
==this.channelFuture =this.bootstrap.bind(new InetSocketAddress(ip(), port())).sync();//綁定指定的端口
