九、ApacheDubbo3协议解析（三）Dubbo协议解析⼊⼝讲解先⼊⼝类：
服务端
1. org.fig.bootstrap.DubboBootstrap#start
2. org.fig.bootstrap.DubboBootstrap#exportServices
3. org.fig.ServiceConfig#export
4. org.fig.ServiceConfig#doExport
5. org.fig.ServiceConfig#doExportUrls
6. org.fig.ServiceConfig#doExportUrlsFor1Protocol
7. org.istry.integration.RegistryProtocol#export
8. org.istry.integration.RegistryProtocol#doLocalExport
9. org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#export
10. org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#openServer
11. org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#createServer
12. org.hange.Exchangers#bind(org.apache.dubbomon.URL,
org.hange.ExchangeHandler)
13. org.hange.support.header.HeaderExchanger#bind
14. org.ing.Transporters#bind(org.apache.dubbomon.URL,
org.ing.ChannelHandler…)
15. org.ansportty4.NettyTransporter#bind
16. org.ansportty4.NettyServer#NettyServer
17. org.ansport.AbstractServer#AbstractServer
18. org.ansportty4.NettyServer#doOpen
服务端部分我们到了NettyServer的doOpen⽅法 @Override
protected void doOpen()throws Throwable{
bootstrap =new ServerBootstrap();
bossGroup = NettyEventLoopFactory.eventLoopGroup(1,"NettyServerBoss");
workerGroup = NettyEventLoopFactory.eventLoopGroup(
getUrl().getPositiveParameter(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS),
"NettyServerWorker");
final NettyServerHandler nettyServerHandler =new NettyServerHandler(getUrl(),this);
channels = Channels();
.
channel(NettyEventLoopFactory.serverSocketChannelClass())
.option(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
.childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
.childHandler(new ChannelInitializer<SocketChannel>(){
@Override
protected void initChannel(SocketChannel ch)throws Exception {
// FIXME: should we use getTimeout()?
int idleTimeout = IdleTimeout(getUrl());
NettyCodecAdapter adapter =new NettyCodecAdapter(getCodec(),getUrl(), NettyServer.this);
if(getUrl().getParameter(SSL_ENABLED_KEY,false)){
ch.pipeline().addLast("negotiation",
SslHandlerInitializer.sslServerHandler(getUrl(), nettyServerHandler));
}
ch.pipeline()
.addLast("decoder", Decoder())
.addLast("encoder", Encoder())
.addLast("server-idle-handler",new IdleStateHandler(0,0, idleTimeout, MILLISECONDS))
.addLast("handler", nettyServerHandler);
}
});
// bind
ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
channelFuture.syncUninterruptibly();
channel = channelFuture.channel();
}
客户端
1. org.fig.bootstrap.DubboBootstrap#start
2. org.fig.bootstrap.DubboBootstrap#referServices
3. org.fig.utils.ReferenceConfigCache#get(org.fig.ReferenceConfigBase)
4. org.fig.ReferenceConfig#get
5. org.fig.ReferenceConfig#init
6. org.fig.ReferenceConfig#createProxy
7. org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#refer
8. org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#protocolBindingRefer
9. org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#getClients
10. org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#initClient
bind9
11. org.hange.Exchangers#connect(org.apache.dubbomon.URL,
org.hange.ExchangeHandler)
12. org.hange.support.header.HeaderExchanger#connect
13. org.ing.Transporters#connect(org.apache.dubbomon.URL,
org.ing.ChannelHandler…)
14. org.ansportty4.NettyTransporter#connect
内部会计控制规范15. org.ansportty4.NettyClient#NettyClient
16. org.ansport.AbstractClient#AbstractClient
17. org.ansportty4.NettyClient#doOpen
客户端部分我们到了NettyClient的doOpen⽅法
@Override
protected void doOpen()throws Throwable {
final NettyClientHandler nettyClientHandler =new NettyClientHandler(getUrl(),this);
bootstrap =new Bootstrap();
灿都
.option(ChannelOption.SO_KEEPALIVE,true)
.
option(ChannelOption.TCP_NODELAY,true)7月份CPI同比上涨
.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
//.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout())
.channel(socketChannelClass());
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Math.max(3000,getConnectTimeout()));
bootstrap.handler(new ChannelInitializer<SocketChannel>(){
@Override
protected void initChannel(SocketChannel ch)throws Exception {
int heartbeatInterval = Heartbeat(getUrl());
if(getUrl().getParameter(SSL_ENABLED_KEY,false)){
ch.pipeline().addLast("negotiation", SslHandlerInitializer.sslClientHandler(getUrl(), nettyClientHandler));
}
NettyCodecAdapter adapter =new NettyCodecAdapter(getCodec(),getUrl(), NettyClient.this);
ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
//我们看到此处为解码
.addLast("decoder", Decoder())
//我们看到此处为编码
.addLast("encoder", Encoder())
.addLast("client-idle-handler",new IdleStateHandler(heartbeatInterval,0,0, MILLISECONDS))
.addLast("handler", nettyClientHandler);
String socksProxyHost = Property(SOCKS_PROXY_HOST);
if(socksProxyHost != null){
int socksProxyPort = Integer.Property(SOCKS_PROXY_PORT, DEFAULT_SOCKS_PROXY_PORT));
Socks5ProxyHandler socks5ProxyHandler =new Socks5ProxyHandler(new InetSocketAddress(socksProxyHost, socksProxyPort)); ch.pipeline().addFirst(socks5ProxyHandler);
}
}
});
}
由以上代码我们看到了解码与编码的调⽤，我们就来研究下此处代码
final public class NettyCodecAdapter {
private final ChannelHandler encoder =new InternalEncoder();
private final ChannelHandler decoder =new InternalDecoder();
public ChannelHandler getEncoder(){
return encoder;
}
public ChannelHandler getDecoder(){
return decoder;
}
private class InternalDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf input, List<Object> out)throws Exception { ChannelBuffer message =new NettyBackedChannelBuffer(input);
统一登录
NettyChannel channel = OrAddChannel(ctx.channel(), url, handler);
// decode object.
do{
int saveReaderIndex = aderIndex();
Object msg = codec.decode(channel, message);
if(msg == Codec2.DecodeResult.NEED_MORE_INPUT){
break;
}else{
//is it possible to go here ?
if(saveReaderIndex == aderIndex()){
throw new IOException("Decode without read data.");
}
if(msg != null){
dota半人马
out.add(msg);
}
}
}adable());
}
}
}
我们先看编码部分
豫公网安备41160202000603
站长QQ:729038198
关于我们
投诉建议