您好,登錄后才能下訂單哦!
前言
眾所周知,國內(nèi)知名框架 Dubbo 底層使用的是 Netty 作為網(wǎng)絡(luò)通信,那么內(nèi)部到底是如何使用的呢?今天我們就來一探究竟。
1. dubbo 的 Consumer 消費者如何使用 Netty
注意:此次代碼使用了從 github 上 clone 的 dubbo 源碼中的 dubbo-demo 例子。
代碼如下:
System.setProperty("java.net.preferIPv4Stack", "true");
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-consumer.xml"});
context.start();
// @1
DemoService demoService = (DemoService) context.getBean("demoService"); // get remote service proxy
int a = 0;
while (true) {
try {
Thread.sleep(1000);
System.err.println( ++ a + " ");
String hello = demoService.sayHello("world"); // call remote method
System.out.println(hello); // get result
} catch (Throwable throwable) {
throwable.printStackTrace();
}
}
當(dāng)代碼執(zhí)行到 @1 的時候,會調(diào)用 Spring 容器的 getBean 方法,而 dubbo 擴展了 FactoryBean,所以,會調(diào)用 getObject 方法,該方法會創(chuàng)建代理對象。
這個過程中會調(diào)用 DubboProtocol 實例的 getClients(URL url) 方法,當(dāng)這個給定的 URL 的 client 沒有初始化則創(chuàng)建,然后放入緩存,代碼如下:
這個 initClient 方法就是創(chuàng)建 Netty 的 client 的。
最終調(diào)用的就是抽象父類 AbstractClient 的構(gòu)造方法,構(gòu)造方法中包含了創(chuàng)建 Socket 客戶端,連接客戶端等行為。
public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
doOpen();
connect();
}
doOpent 方法用來創(chuàng)建 Netty 的 bootstrap :
protected void doOpen() throws Throwable {
NettyHelper.setNettyLoggerFactory();
bootstrap = new ClientBootstrap(channelFactory);
bootstrap.setOption("keepAlive", true);
bootstrap.setOption("tcpNoDelay", true);
bootstrap.setOption("connectTimeoutMillis", getTimeout());
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
public ChannelPipeline getPipeline() {
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("decoder", adapter.getDecoder());
pipeline.addLast("encoder", adapter.getEncoder());
pipeline.addLast("handler", nettyHandler);
return pipeline;
}
});
}
connect 方法用來連接提供者:
protected void doConnect() throws Throwable {
long start = System.currentTimeMillis();
ChannelFuture future = bootstrap.connect(getConnectAddress());
boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
if (ret && future.isSuccess()) {
Channel newChannel = future.getChannel();
newChannel.setInterestOps(Channel.OP_READ_WRITE);
}
}
上面的代碼中,調(diào)用了 bootstrap 的 connect 方法,熟悉的 Netty 連接操作。當(dāng)然這里使用的是 jboss 的 netty3,稍微有點區(qū)別。當(dāng)連接成功后,注冊寫事件,準備開始向提供者傳遞數(shù)據(jù)。
當(dāng) main 方法中調(diào)用 demoService.sayHello(“world”) 的時候,最終會調(diào)用 HeaderExchangeChannel 的 request 方法,通過 channel 進行請求。
public ResponseFuture request(Object request, int timeout) throws RemotingException {
Request req = new Request();
req.setVersion("2.0.0");
req.setTwoWay(true);
req.setData(request);
DefaultFuture future = new DefaultFuture(channel, req, timeout);
channel.send(req);
return future;
}
send 方法中最后調(diào)用 jboss Netty 中繼承了 NioSocketChannel 的 NioClientSocketChannel 的 write 方法。完成了一次數(shù)據(jù)的傳輸。
2. dubbo 的 Provider 提供者如何使用 Netty
Provider demo 代碼:
System.setProperty("java.net.preferIPv4Stack", "true");
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-provider.xml"});
context.start();
System.in.read(); // press any key to exit
Provider 作為被訪問方,肯定是一個 Server 模式的 Socket。如何啟動的呢?
當(dāng) Spring 容器啟動的時候,會調(diào)用一些擴展類的初始化方法,比如繼承了 InitializingBean,ApplicationContextAware,ApplicationListener 。
而 dubbo 創(chuàng)建了 ServiceBean 繼承了一個監(jiān)聽器。Spring 會調(diào)用他的 onApplicationEvent 方法,該類有一個 export 方法,用于打開 ServerSocket 。
然后執(zhí)行了 DubboProtocol 的 createServer 方法,然后創(chuàng)建了一個 NettyServer 對象。NettyServer 對象的 構(gòu)造方法同樣是 doOpen 方法和。代碼如下:
protected void doOpen() throws Throwable {
NettyHelper.setNettyLoggerFactory();
ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
bootstrap = new ServerBootstrap(channelFactory);
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
channels = nettyHandler.getChannels();
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
public ChannelPipeline getPipeline() {
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("decoder", adapter.getDecoder());
pipeline.addLast("encoder", adapter.getEncoder());
pipeline.addLast("handler", nettyHandler);
return pipeline;
}
});
channel = bootstrap.bind(getBindAddress());
}
該方法中,看到了熟悉的 boss 線程,worker 線程,和 ServerBootstrap,在添加了編解碼 handler 之后,添加一個 NettyHandler,最后調(diào)用 bind 方法,完成綁定端口的工作。和我們使用 Netty 是一摸一樣。
3. 總結(jié)
可以看到,dubbo 使用 Netty 還是挺簡單的,消費者使用 NettyClient,提供者使用 NettyServer,Provider 啟動的時候,會開啟端口監(jiān)聽,使用我們平時啟動 Netty 一樣的方式。
而 Client 在 Spring getBean 的時候,會創(chuàng)建 Client,當(dāng)調(diào)用遠程方法的時候,將數(shù)據(jù)通過 dubbo 協(xié)議編碼發(fā)送到 NettyServer,然后 NettServer 收到數(shù)據(jù)后解碼,并調(diào)用本地方法,并返回數(shù)據(jù),完成一次完美的 RPC 調(diào)用。
--------------------- 作者:SJYUA 來源:CSDN 原文:版權(quán)聲明:本文為博主原創(chuàng)文章,轉(zhuǎn)載請附上博文鏈接!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。