经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Zookeeper » 查看文章
Dubbo 我手写几行代码，就把通信模式给你解释清楚！
来源：cnblogs　　作者：小傅哥　　时间：2023/7/12 10:58:48　　对本文有异议

作者：小傅哥

博客：https://bugstack.cn

原文：https://bugstack.cn/md/road-map/road-map.html

沉淀、分享、成长，让自己和他人都能有所收获！??

本文的宗旨在于通过简单干净实践的方式教会读者，为什么要使用Dubbo、怎么使用Dubbo、Dubbo通信的原理是什么。在学习本文后，你可以避开很多关于 Dubbo 使用时的坑，也能更清楚自己的编码是在做什么。

本文涉及的工程：

一、为什么使用

随着互联网场景中所要面对的用户规模和体量的增加，系统的也需要做相应的拆分设计和实现。随之而来的，以前的一套系统，现在成了多个微服务。如；电商系统，以前就在一个工程中写就可以了，现在需要拆分出，用户、支付、商品、配送、活动、风控等各个模块。那么这些模块拆分后，如何高效的通信呢？

  • 关于通信，就引入了 RPC 框架，而 Dubbo 就是其中的一个实现方式。
  • 那为啥用 Dubbo 呢？其实核心问题就一个，为了提高通信效率。因为 Dubbo 的底层通信是 Socket 而不是 HTTP 所以通信的性能会更好。同时 Dubbo 又有分布式的高可用设计，在一组部署了交易服务的实例宕机后，会被从注册中心摘除，之后流量会打到其他服务上。

二、要怎么使用

Dubbo 的使用分为2方，一个是接口的提供方，另外一个是接口的调用方。接口的提供方需要提供出被调用方使用接口的描述性信息。这个信息包括；接口名称、接口入参、接口出参，只有让调用方拿到这些信息以后，它才能依托于这样的接口信息做一个代理操作，并在代理类中使用 Socket 完成双方的信息交互。

所以你看上去调用 RPC 接口好像和使用 HTTP 也没啥区别，无非就是引入了 POM 配置，之后再配置了注解就可以使用了。但其实，它是把你的 Jar 当做代理的必要参数使用了。本文也会介绍，具体是怎么代理的

三、使用的案例

对于编程的学习来说，其实最开始的那一下，不是搞明白所有原理，而是先让自己可以看到运行出来的效果。哎，之后就去分析原理，这样会舒服的多。

所以小傅哥这里提供了一套简单的 Dubbo 使用案例，只要你满足最基本的配置条件，就可以运行出效果；

  1. JDK 1.8
  2. Maven 3.x - jdk1.8支持的就可以
  3. Dubbo 3.1.4 - POM 中已经配置，与2.x最大的使用上的区别就是一些注解的使用
  4. Zookeeper 3.4.x - 如果你只是按照本文中的直连模式测试，那么不安装 Zookeeper 也可以

1. 接口提供方

工程案例创建结构，采用的是 DDD 结构。但和 DDD 一点关系没有。如果你对工程创建有疑惑，可以参考 《Java 简明教程》之 DDD 架构

1.1 接口定义

源码：cn.bugstack.dev.tech.dubbo.api.IUserService

  1. public interface IUserService {
  3.     Response<UserResDTO> queryUserInfo(UserReqDTO reqDTO);
  5. }
  • 接口定义平平无奇，但第1个坑暗藏玄机！
  • 也就是，所有的 Dubbo 接口，出入参，默认都需要继承 Serializable 接口。也就是 UserReqDTO、UserResDTO、Response 这3个类，都得继承 Serializable 序列化接口。

1.2 接口实现

源码：cn.bugstack.dev.tech.dubbo.trigger.rpc.UserService

  1. @Slf4j
  2. @DubboService(version = "1.0.0")
  3. public class UserService implements IUserService {
  5.     @Override
  6.     public Response<UserResDTO> queryUserInfo(UserReqDTO reqDTO) {
  7.         log.info("查询用户信息 userId: {} reqStr: {}", reqDTO.getUserId(), JSON.toJSONString(reqDTO));
  8.         try {
  9.             // 1. 模拟查询【你可以从数据库或者Redis缓存获取数据】
  10.             UserResDTO resDTO = UserResDTO.builder()
  11.                     .userId(reqDTO.getUserId())
  12.                     .userName("小傅哥")
  13.                     .userAge(20)
  14.                     .build();
  16.             // 2. 返回结果
  17.             return Response.<UserResDTO>builder()
  18.                     .code(Constants.ResponseCode.SUCCESS.getCode())
  19.                     .info(Constants.ResponseCode.SUCCESS.getInfo())
  20.                     .data(resDTO).build();
  21.         } catch (Exception e) {
  22.             log.error("查询用户信息失败 userId: {} reqStr: {}", reqDTO.getUserId(), JSON.toJSONString(reqDTO), e);
  23.             return Response.<UserResDTO>builder()
  24.                     .code(Constants.ResponseCode.UN_ERROR.getCode())
  25.                     .info(Constants.ResponseCode.UN_ERROR.getInfo())
  26.                     .build();
  27.         }
  28.     }
  30. }
  • 接口实现平平无奇，但第2个坑暗藏玄机！
  • Dubbo 的实现接口，需要被 Dubbo 自己管理。所以 Dubbo 提供了 @DubboService 注解。有些小卡拉米，使用的是不是 Spring 的 @Service 呀？尤其是以前的 Dubbo 版本 2.7.* 它的注解也是 @Service 也不留神就用成了 Spring 的 @Service。一个小bug，又调了一上午。

1.3 工程配置

application.yml

  1. dubbo:
  2.   application:
  3.     name: xfg-dev-tech-dubbo
  4.     version: 1.0.0
  5.   registry:
  6.     address: zookeeper://127.0.0.1:2181 # N/A - 无zookeeper可配置 N/A 走直连模式测试
  7.   protocol:
  8.     name: dubbo
  9.     port: 20881
  10.   scan:
  11.     base-packages: cn.bugstack.dev.tech.dubbo.api
  • 配置信息平平无奇，但第3个坑暗藏玄机！
  • base-packages 扫描的是哪里配置了 Dubbo 的 API 入口，给它入口就行，它会自己找到实现类。但！你要知道 Java 的 Spring 应用能扫描到，能被 Spring 管理，那么 pom 要直接或者间接的引导到定义了 Dubbo 的模块。
  • 再有一个问题，Spring 应用开发，讲究约定大于配置。你 Application 应用，的包名应该是可以覆盖到其他包名的。比如 Application 都配置到 cn.bugstack.dev.tech.dubbo.a.b.c.d.* 去了，它默认就扫不到 cn.bugstack.dev.tech.dubbo.api 了。一个小bug，一下午又过去了。
  • 注意：address：如果配置的是 N/A 就是不走任何注册中心，就是个直连，主要用于本地验证的。如果你配置了 zookeeper://127.0.0.1:2181 就需要先安装一个 zookeeper 另外，即使你配置了注册中心的方式，也可以直连测试。

1.4 应用构建

以上信息都准备了，一群小卡拉米开始掉到第4个坑里了！

你有2个应用，一个Dubbo接口提供方、一个Dubbo接口使用方。那么你在给你另外一个应用使用接口的时候，你在 InelliJ IDEA 的 Maven 中执行 Install 了吗？

Install 是干啥的？它是为了让你使用了同一个本地 Maven 配置的应用，可以引入到对方提供的 Jar 包。你 Install 以后，这个 Jar 包就会进入到本地 Maven 仓库了。如果是公司里开发，会有专门的自己家部署的，私有Maven中心仓库，就可以通过 deploy 把本地 Jar 发布上去，那么公司里的伙伴，也就都可以引用了。

  • 你要先点击 root 下的 install 操作，这样就会自动构建了。
  • 如果你电脑配置有点低，也会出现一些气人怪相，比如就刷不进去，install 了也引用不了。记得要 clean 清空下，也可以直接到 maven 文件件去清空。

2. 接口使用方

有些小卡拉米觉得前面的抗都扫干净了，就完事了。没有接下来还有坑，让你一搞搞一天，半夜也睡不好。

2.1 POM 引入

  1. <dependency>
  2.     <groupId>cn.bugstack</groupId>
  3.     <artifactId>xfg-dev-tech-dubbo-api</artifactId>
  4.     <version>1.0-SNAPSHOT</version>
  5. </dependency>
  • POM 的配置，就是把 Jar 包给引用进来。因为 Dubbo 需要根据这个接口，做一个代理操作。不引入，你代码就爆红啦！爆红啦！??

2.2 消费配置

源码：application.yml

  1. dubbo:
  2.   application:
  3.     name: xfg-dev-tech-dubbo
  4.     version: 1.0.0
  5.   registry:
  6.      address: zookeeper://127.0.0.1:2181
  7. #    address: N/A
  8.   protocol:
  9.     name: dubbo
  10.     port: 20881
  • 配置了 zookeeper 你就用第一个，代码中对应 @DubboReference(interfaceClass = IUserService.class, version = "1.0.0")
  • 配置了 N/A 你就用第二个，代码中必须指定直连。@DubboReference(interfaceClass = IUserService.class, url = "dubbo://127.0.0.1:20881", version = "1.0.0")

2.3 代码配置

源码：cn.bugstack.dev.tech.dubbo.consumer.test.ApiTest

  1. // 直连模式；@DubboReference(interfaceClass = IUserService.class, url = "dubbo://127.0.0.1:20881", version = "1.0.0")
  2. @DubboReference(interfaceClass = IUserService.class, version = "1.0.0")
  3. private IUserService userService;
  5. @Test
  6. public void test_userService() {
  7.     UserReqDTO reqDTO = UserReqDTO.builder().userId("10001").build();
  8.     Response<UserResDTO> resDTO = userService.queryUserInfo(reqDTO);
  9.     log.info("测试结果 req: {} res: {}", JSON.toJSONString(reqDTO), JSON.toJSONString(resDTO));
  10. }

测试结果

  1. 2023-07-08 15:37:22.291  INFO 62481 --- [           main] c.b.d.tech.dubbo.consumer.test.ApiTest   : 测试结果 req: {"userId":"10001"} res: {"code":"0000","data":{"userAge":20,"userId":"10001","userName":"小傅哥"},"info":"成功"}
  2. 2023-07-08 15:37:22.324  INFO 62481 --- [tor-Framework-0] o.a.c.f.imps.CuratorFrameworkImpl        : backgroundOperationsLoop exiting
  • 如果不出啥意外，到这你就可以直接启动运行了。并看到测试结果。
  • 但别忘记了，你启动的时候，需要先启动 xfg-dev-tech-dubbo 让接口提供方跑起来。

四、原理的分析

都说 Jar 是提供可描述性信息的，对方才能代理调用。那么这个过程是怎么干的呢，总不能一问这个，就让小卡拉米们去手写 Dubbo 呀！所以小傅哥会通过最简单模型结构，让你了解这个 Dubbo 通信的原理，方便小卡拉米们上手。

  • 如果所示，接口使用方，对接口进行代理。什么是代理呢，代理就是用一个包装的结构，代替原有的操作。在这个包装的结构里，你可以自己扩展出任意的方法。
  • 那么，这里的代理。就是根据接口的信息，创建出一个代理对象，在代理对象中，提供 Socket 请求。当调用这个接口的时候，就可以对接口提供方的，发起 Socket 请求了。
  • 而 Socket 接收方，也就是接口提供方。他收到信息以后，根据接口的描述性内容，进行一个反射调用。这下就把信息给请求出来，之后再通过 Socket 返回回去就可以了。

好，核心的原理就这么点。接下来，我们从代码中看看。

1. 接口代理 - 提供方

源码：cn.bugstack.dev.tech.dubbo.trigger.socket.RpcServerSocket

  1. @Slf4j
  2. @Service
  3. public class RpcServerSocket implements Runnable {
  5.     private ApplicationContext applicationContext;
  7.     public RpcServerSocket(ApplicationContext applicationContext) {
  8.         this.applicationContext = applicationContext;
  9.         new Thread(this).start();
  10.     }
  12.     @Override
  13.     public void run() {
  14.         EventLoopGroup bossGroup = new NioEventLoopGroup();
  15.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  17.         try {
  18.             ServerBootstrap b = new ServerBootstrap();
  19.             b.group(bossGroup, workerGroup)
  20.                     .channel(NioServerSocketChannel.class)
  21.                     .option(ChannelOption.SO_BACKLOG, 128)
  22.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  23.                         @Override
  24.                         public void initChannel(SocketChannel channel) {
  25.                             channel.pipeline().addLast(new ObjectEncoder());
  26.                             channel.pipeline().addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
  27.                             channel.pipeline().addLast(new SimpleChannelInboundHandler<Map<String, Object>>() {
  29.                                 @Override
  30.                                 protected void channelRead0(ChannelHandlerContext channelHandlerContext, Map<String, Object> request) throws Exception {
  31.                                     // 解析参数
  32.                                     Class<?> clazz = (Class<?>) request.get("clazz");
  33.                                     String methodName = (String) request.get("methodName");
  34.                                     Class<?>[] paramTypes = (Class<?>[]) request.get("paramTypes");
  35.                                     Object[] args = (Object[]) request.get("args");
  37.                                     // 反射调用
  38.                                     Method method = clazz.getMethod(methodName, paramTypes);
  39.                                     Object invoke = method.invoke(applicationContext.getBean(clazz), args);
  41.                                     // 封装结果
  42.                                     Map<String, Object> response = new HashMap<>();
  43.                                     response.put("data", invoke);
  45.                                     log.info("RPC 请求调用 clazz:{} methodName:{}, response:{}", clazz.getName(), methodName, JSON.toJSON(response));
  46.                                     // 回写数据
  47.                                     channelHandlerContext.channel().writeAndFlush(response);
  48.                                 }
  49.                             });
  50.                         }
  51.                     });
  53.             ChannelFuture f = b.bind(22881).sync();
  54.             f.channel().closeFuture().sync();
  55.         } catch (Exception e) {
  56.             e.printStackTrace();
  57.         } finally {
  58.             bossGroup.shutdownGracefully();
  59.             workerGroup.shutdownGracefully();
  60.         }
  61.     }
  63. }

这段代码主要提供的功能包括；

  1. Netty Socket 启动一个服务端
  2. 注入 ApplicationContext applicationContext 用于在接收到请求接口信息后，获取对应的 Bean 对象。
  3. 根据请求来的 Bean 对象，以及参数的必要信息。进行接口的反射调用。
  4. 最后一步，就是把接口反射请求的信息，再通过 Socket 返回回去。

2. 接口反射 - 调用方

打开工程：xfg-dev-tech-dubbo-test

源码：cn.bugstack.dev.tech.dubbo.consumer.config.RPCProxyBeanFactory

  1. @Slf4j
  2. @Component("rpcProxyBeanFactory")
  3. public class RPCProxyBeanFactory implements FactoryBean<IUserService>, Runnable {
  5.     private Channel channel;
  7.     // 缓存数据，实际RPC会对每次的调用生成一个ID来标记获取
  8.     private Object responseCache;
  10.     public RPCProxyBeanFactory() throws InterruptedException {
  11.         new Thread(this).start();
  12.         while (null == channel) {
  13.             Thread.sleep(150);
  14.             log.info("Rpc Socket 链接等待...");
  15.         }
  16.     }
  18.     @Override
  19.     public IUserService getObject() throws Exception {
  21.         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
  22.         Class<?>[] classes = {IUserService.class};
  23.         InvocationHandler handler = new InvocationHandler() {
  24.             @Override
  25.             public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  27.                 if (Object.class.equals(method.getDeclaringClass())) {
  28.                     return method.invoke(this, args);
  29.                 }
  31.                 Map<String, Object> request = new HashMap<>();
  32.                 request.put("clazz", IUserService.class);
  33.                 request.put("methodName", method.getName());
  34.                 request.put("paramTypes", method.getParameterTypes());
  35.                 request.put("args", args);
  37.                 channel.writeAndFlush(request);
  39.                 // 模拟超时等待，一般RPC接口请求，都有一个超时等待时长。
  40.                 Thread.sleep(350);
  42.                 return responseCache;
  43.             }
  44.         };
  46.         return (IUserService) Proxy.newProxyInstance(classLoader, classes, handler);
  47.     }
  49.     @Override
  50.     public Class<?> getObjectType() {
  51.         return IUserService.class;
  52.     }
  54.     @Override
  55.     public void run() {
  56.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  57.         try {
  58.             Bootstrap b = new Bootstrap();
  59.             b.group(workerGroup)
  60.                     .channel(NioSocketChannel.class)
  61.                     .option(ChannelOption.AUTO_READ, true)
  62.                     .handler(new ChannelInitializer<SocketChannel>() {
  64.                         @Override
  65.                         protected void initChannel(SocketChannel channel) throws Exception {
  66.                             channel.pipeline().addLast(new ObjectEncoder());
  67.                             channel.pipeline().addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
  68.                             channel.pipeline().addLast(new SimpleChannelInboundHandler<Map<String, Object>>() {
  70.                                 @Override
  71.                                 protected void channelRead0(ChannelHandlerContext channelHandlerContext, Map<String, Object> data) throws Exception {
  72.                                     responseCache = data.get("data");
  73.                                 }
  74.                             });
  75.                         }
  76.                     });
  77.             ChannelFuture channelFuture = b.connect("127.0.0.1", 22881).syncUninterruptibly();
  78.             this.channel = channelFuture.channel();
  79.             channelFuture.channel().closeFuture().syncUninterruptibly();
  80.         } finally {
  81.             workerGroup.shutdownGracefully();
  82.         }
  83.     }
  85. }

这段代码主要提供的功能包括；

  1. 实现 FactoryBean<IUserService> 为的是把这样一个代理对象，交给 Spring 容器管理。
  2. 实现 Runnable 接口，并在接口中，创建 Netty 的 Socket 客户端。客户端中接收来自服务端的消息，并临时存放到缓存中。注意 Dubbo 中这块的处理会复杂一些，以及请求同步响应通信，这样才能把各个接口的调动记录下来
  3. getObject() 对象中，提供代理操作。代理里，就可以自己想咋搞咋搞了。而 Dubbo 也是在代理里，提供了如此的操作，对接口提供方发送请求消息，并在超时时间内返回接口信息。因为反射调用，需要你提供类、方法、入参类型、入参内容，所以我们要把这些信息传递给接口提供方。

3. 服务测试 - 消费验证

  • 启动 xfg-dev-tech-dubbo
  • 测试 xfg-dev-tech-dubbo-test
  1. @Resource(name = "rpcProxyBeanFactory")
  2. private IUserService proxyUserService;
  4. @Test
  5. public void test_proxyUserService(){
  6.     UserReqDTO reqDTO = UserReqDTO.builder().userId("10001").build();
  7.     Response<UserResDTO> resDTO = proxyUserService.queryUserInfo(reqDTO);
  8.     log.info("测试结果 req: {} res: {}", JSON.toJSONString(reqDTO), JSON.toJSONString(resDTO));
  9. }

测试结果

  1. 2023-07-08 16:14:51.322  INFO 74498 --- [           main] c.b.d.tech.dubbo.consumer.test.ApiTest   : 测试结果 req: {"userId":"10001"} res: {"code":"0000","data":{"userAge":20,"userId":"10001","userName":"小傅哥"},"info":"成功"}
  • 这里我们给 IUserService 注入一个自己代理好的对象，之后就可以调用验证了。
  • 好啦，到这我们就把关于 Dubbo 的事交代明白了，以上内容较多。小卡拉米需要细细的品味吸收！

原文链接：https://www.cnblogs.com/xiaofuge/p/17543083.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Zookeeper热门文章

  • java Zookeeper简述
  • zookeeper-操作与应用场景-《每日五分钟搞定大数据》
  • ZooInspector 连接不到 Zookeeper 的解决方法
  • IDEA zookeeper插件的使用
  • 解决dubbo错误ip及ip乱入问题的方法
  • zookeeper集群搭建
  • 大数据开发phoenix连接hbase流程详解
  • 快来看看 Dubbo 是如何实现服务的动态发现的
  • zookeeper特性与节点说明
  • zookeeper【1】配置管理

    • Zookeeper推荐文章

  • Zookeeper安装部署
  • 得物 ZooKeeper SLA 也可以 99.99%
  • zookeeper源码(06)ZooKeeperServer及子类
  • Dubbo 我手写几行代码，就把通信模式给你解释清楚！
  • Rpc-实现Zookeeper注册中心
  • Dubbo 中 Zookeeper 注册中心原理分析
  • Java&nbsp;zookeeper服务的使用详解
  • 【入门】ZooKeeper 相关概念总结
  • zookeeper特性与节点说明
  • 快来看看 Dubbo 是如何实现服务的动态发现的

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号