Dubbo 我手寫幾行代碼，就把通信模式給你解釋清楚！

作者：小傅哥 2023-07-10 08:59:37

隨著互聯網場景中所要面對的用戶規模和體量的增加，系統的也需要做相應的拆分設計和實現。隨之而來的，以前的一套系統，現在成了多個微服務。

本文的宗旨在于通過簡單干凈實踐的方式教會讀者，為什么要使用Dubbo、怎么使用Dubbo、Dubbo通信的原理是什么。在學習本文后，你可以避開很多關于 Dubbo 使用時的坑，也能更清楚自己的編碼是在做什么。

本文涉及的工程：

xfg-dev-tech-dubbo：https://gitcode.net/KnowledgePlanet/road-map/xfg-dev-tech-dubbo
xfg-dev-tech-dubbo-test：https://gitcode.net/KnowledgePlanet/road-map/xfg-dev-tech-dubbo-test

一、為什么使用

隨著互聯網場景中所要面對的用戶規模和體量的增加，系統的也需要做相應的拆分設計和實現。隨之而來的，以前的一套系統，現在成了多個微服務。如；電商系統，以前就在一個工程中寫就可以了，現在需要拆分出，用戶、支付、商品、配送、活動、風控等各個模塊。那么這些模塊拆分后，如何高效的通信呢？

圖片

關于通信，就引入了 RPC 框架，而 Dubbo 就是其中的一個實現方式。
那為啥用 Dubbo 呢？其實核心問題就一個，為了提高通信效率。因為 Dubbo 的底層通信是 Socket 而不是 HTTP 所以通信的性能會更好。同時 Dubbo 又有分布式的高可用設計，在一組部署了交易服務的實例宕機后，會被從注冊中心摘除，之后流量會打到其他服務上。

二、要怎么使用

圖片

Dubbo 的使用分為2方，一個是接口的提供方，另外一個是接口的調用方。接口的提供方需要提供出被調用方使用接口的描述性信息。這個信息包括；接口名稱、接口入參、接口出參，只有讓調用方拿到這些信息以后，它才能依托于這樣的接口信息做一個代理操作，并在代理類中使用 Socket 完成雙方的信息交互。

所以你看上去調用 RPC 接口好像和使用 HTTP 也沒啥區別，無非就是引入了 POM 配置，之后再配置了注解就可以使用了。但其實，它是把你的 Jar 當做代理的必要參數使用了。本文也會介紹，具體是怎么代理的

三、使用的案例

對于編程的學習來說，其實最開始的那一下，不是搞明白所有原理，而是先讓自己可以看到運行出來的效果。哎，之后就去分析原理，這樣會舒服的多。

所以小傅哥這里提供了一套簡單的 Dubbo 使用案例，只要你滿足最基本的配置條件，就可以運行出效果；

JDK 1.8
Maven 3.x - jdk1.8支持的就可以
Dubbo 3.1.4 - POM 中已經配置，與2.x最大的使用上的區別就是一些注解的使用
Zookeeper 3.4.x - 如果你只是按照本文中的直連模式測試，那么不安裝 Zookeeper 也可以

1. 接口提供方

工程案例創建結構，采用的是 DDD 結構。但和 DDD 一點關系沒有。如果你對工程創建有疑惑，可以參考《Java 簡明教程》之 DDD 架構

圖片

1.1 接口定義

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

public interface IUserService {

    Response<UserResDTO> queryUserInfo(UserReqDTO reqDTO);

}

接口定義平平無奇，但第1個坑暗藏玄機！
也就是，所有的 Dubbo 接口，出入參，默認都需要繼承 Serializable 接口。也就是 UserReqDTO、UserResDTO、Response 這3個類，都得繼承 Serializable 序列化接口。

1.2 接口實現

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

@Slf4j
@DubboService(version = "1.0.0")
public class UserService implements IUserService {

    @Override
    public Response<UserResDTO> queryUserInfo(UserReqDTO reqDTO) {
        log.info("查詢用戶信息 userId: {} reqStr: {}", reqDTO.getUserId(), JSON.toJSONString(reqDTO));
        try {
            // 1. 模擬查詢【你可以從數據庫或者Redis緩存獲取數據】
            UserResDTO resDTO = UserResDTO.builder()
                    .userId(reqDTO.getUserId())
                    .userName("小傅哥")
                    .userAge(20)
                    .build();

            // 2. 返回結果
            return Response.<UserResDTO>builder()
                    .code(Constants.ResponseCode.SUCCESS.getCode())
                    .info(Constants.ResponseCode.SUCCESS.getInfo())
                    .data(resDTO).build();
        } catch (Exception e) {
            log.error("查詢用戶信息失敗 userId: {} reqStr: {}", reqDTO.getUserId(), JSON.toJSONString(reqDTO), e);
            return Response.<UserResDTO>builder()
                    .code(Constants.ResponseCode.UN_ERROR.getCode())
                    .info(Constants.ResponseCode.UN_ERROR.getInfo())
                    .build();
        }
    }

}

接口實現平平無奇，但第2個坑暗藏玄機！
Dubbo 的實現接口，需要被 Dubbo 自己管理。所以 Dubbo 提供了 @DubboService 注解。有些小卡拉米，使用的是不是 Spring 的 @Service 呀？尤其是以前的 Dubbo 版本 2.7.* 它的注解也是 @Service 也不留神就用成了 Spring 的 @Service。一個小bug，又調了一上午。

1.3 工程配置

application.yml

dubbo:
  application:
    name: xfg-dev-tech-dubbo
    version: 1.0.0
  registry:
    address: zookeeper://127.0.0.1:2181 # N/A - 無zookeeper可配置 N/A 走直連模式測試
  protocol:
    name: dubbo
    port: 20881
  scan:
    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 引入

<dependency>
    <groupId>cn.bugstack</groupId>
    <artifactId>xfg-dev-tech-dubbo-api</artifactId>
    <version>1.0-SNAPSHOT</version>
</dependency>

POM 的配置，就是把 Jar 包給引用進來。因為 Dubbo 需要根據這個接口，做一個代理操作。不引入，你代碼就爆紅啦！爆紅啦！??

2.2 消費配置

源碼：application.yml

dubbo:
  application:
    name: xfg-dev-tech-dubbo
    version: 1.0.0
  registry:
     address: zookeeper://127.0.0.1:2181
#    address: N/A
  protocol:
    name: dubbo
    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

// 直連模式；@DubboReference(interfaceClass = IUserService.class, url = "dubbo://127.0.0.1:20881", version = "1.0.0")
@DubboReference(interfaceClass = IUserService.class, version = "1.0.0")
private IUserService userService;

@Test
public void test_userService() {
    UserReqDTO reqDTO = UserReqDTO.builder().userId("10001").build();
    Response<UserResDTO> resDTO = userService.queryUserInfo(reqDTO);
    log.info("測試結果 req: {} res: {}", JSON.toJSONString(reqDTO), JSON.toJSONString(resDTO));
}

測試結果

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":"成功"}
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

@Slf4j
@Service
public class RpcServerSocket implements Runnable {

    private ApplicationContext applicationContext;

    public RpcServerSocket(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
        new Thread(this).start();
    }

    @Override
    public void run() {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel channel) {
                            channel.pipeline().addLast(new ObjectEncoder());
                            channel.pipeline().addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
                            channel.pipeline().addLast(new SimpleChannelInboundHandler<Map<String, Object>>() {

                                @Override
                                protected void channelRead0(ChannelHandlerContext channelHandlerContext, Map<String, Object> request) throws Exception {
                                    // 解析參數
                                    Class<?> clazz = (Class<?>) request.get("clazz");
                                    String methodName = (String) request.get("methodName");
                                    Class<?>[] paramTypes = (Class<?>[]) request.get("paramTypes");
                                    Object[] args = (Object[]) request.get("args");

                                    // 反射調用
                                    Method method = clazz.getMethod(methodName, paramTypes);
                                    Object invoke = method.invoke(applicationContext.getBean(clazz), args);

                                    // 封裝結果
                                    Map<String, Object> response = new HashMap<>();
                                    response.put("data", invoke);

                                    log.info("RPC 請求調用 clazz:{} methodName:{}, response:{}", clazz.getName(), methodName, JSON.toJSON(response));
                                    // 回寫數據
                                    channelHandlerContext.channel().writeAndFlush(response);
                                }
                            });
                        }
                    });

            ChannelFuture f = b.bind(22881).sync();
            f.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

}

這段代碼主要提供的功能包括；

Netty Socket 啟動一個服務端
注入 ApplicationContext applicationContext 用于在接收到請求接口信息后，獲取對應的 Bean 對象。
根據請求來的 Bean 對象，以及參數的必要信息。進行接口的反射調用。
最后一步，就是把接口反射請求的信息，再通過 Socket 返回回去。

2. 接口反射 - 調用方

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

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

@Slf4j
@Component("rpcProxyBeanFactory")
public class RPCProxyBeanFactory implements FactoryBean<IUserService>, Runnable {

    private Channel channel;

    // 緩存數據，實際RPC會對每次的調用生成一個ID來標記獲取
    private Object responseCache;

    public RPCProxyBeanFactory() throws InterruptedException {
        new Thread(this).start();
        while (null == channel) {
            Thread.sleep(150);
            log.info("Rpc Socket 鏈接等待...");
        }
    }

    @Override
    public IUserService getObject() throws Exception {

        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
        Class<?>[] classes = {IUserService.class};
        InvocationHandler handler = new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

                if (Object.class.equals(method.getDeclaringClass())) {
                    return method.invoke(this, args);
                }

                Map<String, Object> request = new HashMap<>();
                request.put("clazz", IUserService.class);
                request.put("methodName", method.getName());
                request.put("paramTypes", method.getParameterTypes());
                request.put("args", args);

                channel.writeAndFlush(request);

                // 模擬超時等待，一般RPC接口請求，都有一個超時等待時長。
                Thread.sleep(350);

                return responseCache;
            }
        };

        return (IUserService) Proxy.newProxyInstance(classLoader, classes, handler);
    }

    @Override
    public Class<?> getObjectType() {
        return IUserService.class;
    }

    @Override
    public void run() {
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.AUTO_READ, true)
                    .handler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            channel.pipeline().addLast(new ObjectEncoder());
                            channel.pipeline().addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
                            channel.pipeline().addLast(new SimpleChannelInboundHandler<Map<String, Object>>() {

                                @Override
                                protected void channelRead0(ChannelHandlerContext channelHandlerContext, Map<String, Object> data) throws Exception {
                                    responseCache = data.get("data");
                                }
                            });
                        }
                    });
            ChannelFuture channelFuture = b.connect("127.0.0.1", 22881).syncUninterruptibly();
            this.channel = channelFuture.channel();
            channelFuture.channel().closeFuture().syncUninterruptibly();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }

}

這段代碼主要提供的功能包括；

實現 FactoryBean<IUserService> 為的是把這樣一個代理對象，交給 Spring 容器管理。
實現 Runnable 接口，并在接口中，創建 Netty 的 Socket 客戶端。客戶端中接收來自服務端的消息，并臨時存放到緩存中。注意 Dubbo 中這塊的處理會復雜一些，以及請求同步響應通信，這樣才能把各個接口的調動記錄下來
getObject() 對象中，提供代理操作。代理里，就可以自己想咋搞咋搞了。而 Dubbo 也是在代理里，提供了如此的操作，對接口提供方發送請求消息，并在超時時間內返回接口信息。因為反射調用，需要你提供類、方法、入參類型、入參內容，所以我們要把這些信息傳遞給接口提供方。

3. 服務測試 - 消費驗證

啟動 xfg-dev-tech-dubbo
測試 xfg-dev-tech-dubbo-test

@Resource(name = "rpcProxyBeanFactory")
private IUserService proxyUserService;

@Test
public void test_proxyUserService(){
    UserReqDTO reqDTO = UserReqDTO.builder().userId("10001").build();
    Response<UserResDTO> resDTO = proxyUserService.queryUserInfo(reqDTO);
    log.info("測試結果 req: {} res: {}", JSON.toJSONString(reqDTO), JSON.toJSONString(resDTO));
}

測試結果

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 的事交代明白了，以上內容較多。小卡拉米需要細細的品味吸收！

責任編輯：武曉燕來源： bugstack蟲洞棧

Dubbo 通信模式