Netty 在 Dubbo 中是如何应用的?

作者:莫那·鲁道 https://www.cnblogs.com/stateis0/p/9062171.html

众所周知,国内知名框架 Dubbo 底层使用的是 Netty 作为网络通信,那么内部到底是如何使用的呢?今天我们就来一探究竟。

1. dubbo 的 Consumer 消费者如何使用 Netty


注意:此次代码使用了从 github 上 clone 的 dubbo 源码中的 dubbo-demo 例子。

代码如下:

1System.setProperty("java.net.preferIPv4Stack", "true"); 2   ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-consumer.xml"}); 3   context.start(); 4    // @1 5   DemoService demoService = (DemoService) context.getBean("demoService"); // get remote service proxy 6   int a = 0; 7   while (true) { 8       try { 9           Thread.sleep(1000); 10           System.err.println( ++ a + " "); 11 12           String hello = demoService.sayHello("world"); // call remote method 13           System.out.println(hello); // get result 14 15       } catch (Throwable throwable) { 16           throwable.printStackTrace(); 17       } 18   }

当代码执行到 @1 的时候,会调用 Spring 容器的 getBean 方法,而 dubbo 扩展了 FactoryBean,所以,会调用 getObject 方法,该方法会创建代理对象。

这个过程中会调用 DubboProtocol 实例的 getClients(URL url) 方法,当这个给定的 URL 的 client 没有初始化则创建,然后放入缓存,代码如下:

这个 initClient 方法就是创建 Netty 的 client 的。

最终调用的就是抽象父类 AbstractClient 的构造方法,构造方法中包含了创建 Socket 客户端,连接客户端等行为。

1public AbstractClient(URL url, ChannelHandler handler) throws RemotingException { 2   doOpen(); 3   connect(); 4}

doOpent 方法用来创建 Netty 的 bootstrap :

1protected void doOpen() throws Throwable { 2   NettyHelper.setNettyLoggerFactory(); 3   bootstrap = new ClientBootstrap(channelFactory); 4   bootstrap.setOption("keepAlive", true); 5   bootstrap.setOption("tcpNoDelay", true); 6   bootstrap.setOption("connectTimeoutMillis", getTimeout()); 7   final NettyHandler nettyHandler = new NettyHandler(getUrl(), this); 8   bootstrap.setPipelineFactory(new ChannelPipelineFactory() { 9       public ChannelPipeline getPipeline() { 10           NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this); 11           ChannelPipeline pipeline = Channels.pipeline(); 12           pipeline.addLast("decoder", adapter.getDecoder()); 13           pipeline.addLast("encoder", adapter.getEncoder()); 14           pipeline.addLast("handler", nettyHandler); 15           return pipeline; 16       } 17   }); 18}

connect 方法用来连接提供者:

1protected void doConnect() throws Throwable { 2   long start = System.currentTimeMillis(); 3   ChannelFuture future = bootstrap.connect(getConnectAddress()); 4   boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS); 5   if (ret && future.isSuccess()) { 6       Channel newChannel = future.getChannel(); 7       newChannel.setInterestOps(Channel.OP_READ_WRITE); 8   } 9}

上面的代码中,调用了 bootstrap 的 connect 方法,熟悉的 Netty 连接操作。当然这里使用的是  jboss 的 netty3,稍微有点区别。点击这篇:教你用 Netty 实现一个简单的 RPC。当连接成功后,注册写事件,准备开始向提供者传递数据。

当 main 方法中调用 demoService.sayHello(“world”) 的时候,最终会调用 HeaderExchangeChannel 的 request 方法,通过 channel 进行请求。

1public ResponseFuture request(Object request, int timeout) throws RemotingException { 2   Request req = new Request(); 3   req.setVersion("2.0.0"); 4   req.setTwoWay(true); 5   req.setData(request); 6   DefaultFuture future = new DefaultFuture(channel, req, timeout); 7   channel.send(req); 8   return future; 9}

send 方法中最后调用 jboss  Netty 中继承了  NioSocketChannel 的 NioClientSocketChannel 的 write 方法。完成了一次数据的传输。

2. dubbo 的 Provider 提供者如何使用 Netty

Provider demo 代码:

1System.setProperty("java.net.preferIPv4Stack", "true"); 2ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-provider.xml"}); 3context.start(); 4System.in.read(); // press any key to exit

Provider 作为被访问方,肯定是一个 Server 模式的 Socket。如何启动的呢?

当 Spring 容器启动的时候,会调用一些扩展类的初始化方法,比如继承了 InitializingBean,ApplicationContextAware,ApplicationListener 。

而 dubbo 创建了 ServiceBean 继承了一个监听器。Spring 会调用他的 onApplicationEvent 方法,该类有一个 export 方法,用于打开 ServerSocket 。

然后执行了 DubboProtocol 的 createServer 方法,然后创建了一个NettyServer 对象。NettyServer 对象的 构造方法同样是  doOpen 方法和。

代码如下:

1protected void doOpen() throws Throwable { 2   NettyHelper.setNettyLoggerFactory(); 3   ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true)); 4   ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true)); 5   ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS)); 6   bootstrap = new ServerBootstrap(channelFactory); 7 8   final NettyHandler nettyHandler = new NettyHandler(getUrl(), this); 9   channels = nettyHandler.getChannels(); 10   bootstrap.setPipelineFactory(new ChannelPipelineFactory() { 11       public ChannelPipeline getPipeline() { 12           NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this); 13           ChannelPipeline pipeline = Channels.pipeline(); 14           pipeline.addLast("decoder", adapter.getDecoder()); 15           pipeline.addLast("encoder", adapter.getEncoder()); 16           pipeline.addLast("handler", nettyHandler); 17           return pipeline; 18       } 19   }); 20   channel = bootstrap.bind(getBindAddress()); 21}

该方法中,看到了熟悉的 boss 线程,worker 线程,和 ServerBootstrap,在添加了编解码 handler  之后,添加一个 NettyHandler,最后调用 bind 方法,完成绑定端口的工作。和我们使用 Netty 是一摸一样。

3. 总结


可以看到,dubbo 使用 Netty 还是挺简单的,消费者使用 NettyClient,提供者使用 NettyServer,Provider  启动的时候,会开启端口监听,使用我们平时启动 Netty 一样的方式。

而 Client 在 Spring getBean 的时候,会创建 Client,当调用远程方法的时候,将数据通过 dubbo 协议编码发送到 NettyServer,然后 NettServer 收到数据后解码,并调用本地方法,并返回数据,完成一次完美的 RPC 调用。

好,关于 dubbo 如何使用 Netty 就简短的介绍到这里。

推荐去我的博客阅读更多:

1.Java JVM、集合、多线程、新特性系列教程

2.Spring MVC、Spring Boot、Spring Cloud 系列教程

3.Maven、Git、Eclipse、Intellij IDEA 系列工具教程

4.Java、后端、架构、阿里巴巴等大厂最新面试题

觉得不错,别忘了点赞+转发哦!

点赞
收藏

评论区

加载中...

相关推荐

MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1

文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s

Oracle 分组与拼接字符串同时使用

SELECTT.,ROWNUMIDFROM(SELECTT.EMPLID,T.NAME,T.BU,T.REALDEPART,T.FORMATDATE,SUM(T.S0)S0,MAX(UPDATETIME)CREATETIME,LISTAGG(TOCHAR(

MySQL部分从库上面因为大量的临时表tmp_table造成慢查询

背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_

皕杰报表之UUID

​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为

手写Java HashMap源码

HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22

2020年前端实用代码段,为你的工作保驾护航

有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )