Dubbo框架的1個(gè)核心設(shè)計(jì)點(diǎn)

Java領(lǐng)域要說(shuō)讓我最服氣的RPC框架當(dāng)屬Dubbo，原因有許多，但是最吸引我的還是它把遠(yuǎn)程調(diào)用這個(gè)事情設(shè)計(jì)得很有藝術(shù)。

1、Dubbo優(yōu)點(diǎn)較多，我只鐘情其一

1.1、優(yōu)點(diǎn)

業(yè)內(nèi)對(duì)于微服務(wù)之間調(diào)用的框架選擇較多，主流是Spring Cloud的Rest方式 和 Dubbo方式，我使用Dubbo方式居多。Dubbo工業(yè)級(jí)可用，穩(wěn)定又高效，深受各大公司研發(fā)同學(xué)的喜愛(ài)。

Dubbo的優(yōu)點(diǎn)較多，比如：

• 高性能 ：Dubbo 使用的是基于 Netty 的自定義通信協(xié)議，提供了高效的二進(jìn)制數(shù)據(jù)傳輸，使得遠(yuǎn)程服務(wù)調(diào)用性能大幅提升。
• 模塊化設(shè)計(jì) ：Dubbo 的架構(gòu)非常模塊化，主要由五大核心模塊組成：遠(yuǎn)程調(diào)用模塊（RPC）、集群模塊、負(fù)載均衡模塊、容錯(cuò)模塊和注冊(cè)中心模塊。
• 每個(gè)部件都支持多協(xié)議 ：每個(gè)部件都支持多種協(xié)議，比如注冊(cè)中心，支持ZK、Redis、Nacos等等。
• 負(fù)載均衡和容錯(cuò) ：Dubbo 提供了多種容錯(cuò)機(jī)制，比如失敗重試、失敗轉(zhuǎn)移等。還支持多種負(fù)載均衡，比如隨機(jī)、輪詢、一致性哈希等。
• 服務(wù)注冊(cè)和發(fā)現(xiàn) ：Dubbo引入了注冊(cè)中心的概念，實(shí)現(xiàn)了服務(wù)的自動(dòng)注冊(cè)和發(fā)現(xiàn)。
• SPI 擴(kuò)展機(jī)制 ：在 背八股文 場(chǎng)景下，Dubbo被提及最多的就是使用了類(lèi)似Java的SPI機(jī)制，提高了擴(kuò)展性，這一點(diǎn)仁者見(jiàn)仁智者見(jiàn)智吧。

1.2、鐘情其一

但是，Dubbo最吸引人的， 半支煙 覺(jué)得反而倒是它的RPC調(diào)用。Dubbo的定位是一個(gè)RPC框架，這是它的核心和立足之地，所以Dubbo將RPC的調(diào)用過(guò)程透明化，使得開(kāi)發(fā)者可以專(zhuān)注于業(yè)務(wù)邏輯，而不用關(guān)注底層通信問(wèn)題。

一個(gè)RPC框架只有聚焦于先做好它的RPC調(diào)用過(guò)程這個(gè)模塊，才會(huì)有人關(guān)注，其余的優(yōu)點(diǎn)都是在這之后，慢慢迭代而來(lái)。

作者將RPC調(diào)用的這個(gè)過(guò)程，抽象成 一種協(xié)議消息的傳輸機(jī)制 ，再通過(guò) 控制好線程的等待和喚醒 ，來(lái)實(shí)現(xiàn)遠(yuǎn)程方法調(diào)用。這一設(shè)計(jì)思路真是美妙，充分體驗(yàn)了作者的智慧。

2、RPC簡(jiǎn)易示例

學(xué)Dubbo，首先就是要學(xué)習(xí)作者這種設(shè)計(jì)理念和思路。基于此，來(lái)實(shí)現(xiàn)一個(gè)簡(jiǎn)易的遠(yuǎn)程方法調(diào)用，將Dubbo的RPC過(guò)程簡(jiǎn)易化。

2.1、示例步驟

簡(jiǎn)易的RPC過(guò)程步驟如下，大致分5步，依舊使用Netty作用Socket通訊工具。

1. 使用2個(gè)Java進(jìn)程來(lái)模擬2個(gè)系統(tǒng)之間的調(diào)用，A進(jìn)程 和 B進(jìn)程。
2. A進(jìn)程的某個(gè)方法，使用網(wǎng)絡(luò)請(qǐng)求調(diào)用B進(jìn)程的某個(gè)方法。
3. 然后A進(jìn)程的方法就處于等待狀態(tài)。
4. 等B進(jìn)程的方法執(zhí)行完之后，在利用網(wǎng)絡(luò)通知到A進(jìn)程。
5. 然后A進(jìn)程的方法被喚醒，繼續(xù)往下執(zhí)行。

2.2、示例代碼

• B進(jìn)程作為服務(wù)端，啟動(dòng)網(wǎng)絡(luò)服務(wù)

public class BProcessServer {
    private final int port;
    public BProcessServer(int port) {
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(new BProcessServerHandler());
                        }
                    });

            ChannelFuture future = bootstrap.bind(port).sync();
            System.out.println("B啟動(dòng)了服務(wù)，端口號(hào): " + port);
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new BProcessServer(8088).start();
    }
}

• B進(jìn)程接受網(wǎng)絡(luò)請(qǐng)求參數(shù)，反序列化之后，執(zhí)行對(duì)應(yīng)的方法，再將執(zhí)行結(jié)果返回：

public class BProcessServerHandler extends SimpleChannelInboundHandler {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        String reqData = msg.toString(CharsetUtil.UTF_8);
        System.out.println("B進(jìn)程接受到了請(qǐng)求數(shù)據(jù): " + reqData);

        executeMethod(ctx);
    }

    /**
     * 執(zhí)行方法
     *
     * @param ctx
     * @throws InterruptedException
     */
    private void executeMethod(ChannelHandlerContext ctx) throws InterruptedException {
        // TODO 將請(qǐng)求消息按照某種規(guī)則解析成方法名、方法參數(shù)等，其實(shí)就是反序列化的過(guò)程。
        System.out.println("對(duì)接受的數(shù)據(jù)做反序列化，然后開(kāi)始執(zhí)行 消息體里指定的方法...");

        // 模擬方法執(zhí)行
        Thread.sleep(2000);
        System.out.println("執(zhí)行完畢，返回結(jié)果...");

        // 將結(jié)果 通知給 A 進(jìn)程
        ByteBuf dataByteBuf = ctx.alloc().buffer().writeBytes("Task completed".getBytes(CharsetUtil.UTF_8));
        ctx.writeAndFlush(dataByteBuf);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

• A進(jìn)程啟動(dòng)Netty客戶端，建立與B進(jìn)程的通信，然后發(fā)起遠(yuǎn)程調(diào)用，處于等待狀態(tài)。

public class AProcessClient {

    private final String host;
    private final int port;
    private final Object lock = new Object();  // 監(jiān)視器對(duì)象

    public AProcessClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(new AProcessClientHandler(lock));
                        }
                    });

            ChannelFuture future = bootstrap.connect(host, port).sync();
            System.out.println("A進(jìn)程與B進(jìn)程建立了通信連接");

            Channel channel = future.channel();

            // 發(fā)起遠(yuǎn)程調(diào)用
            callRemoteMethod(channel);

            channel.closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * 執(zhí)行方法
     *
     * @param channel
     * @throws InterruptedException
     */
    private void callRemoteMethod(Channel channel) throws InterruptedException {
        //TODO 此處需要將調(diào)用的方法和參數(shù)，按照協(xié)議進(jìn)行序列化。這次暫且省去此過(guò)程。
        System.out.println("A進(jìn)程將 請(qǐng)求的方法和參數(shù) 進(jìn)行序列化，然后向B進(jìn)程發(fā)起網(wǎng)絡(luò)調(diào)用...");

        ByteBuf dataByteBuf = channel.alloc().buffer().writeBytes("Start call method".getBytes(CharsetUtil.UTF_8));

        channel.writeAndFlush(dataByteBuf);

        // 使用wait等待B進(jìn)程通知
        synchronized (lock) {
            System.out.println("A進(jìn)程等待B進(jìn)程的響應(yīng)...");
            lock.wait();  // 等待通知
        }

        System.out.println("A進(jìn)程收到了B進(jìn)程的響應(yīng)通知，繼續(xù)往下...");
    }

    public static void main(String[] args) throws InterruptedException {
        new AProcessClient("localhost", 8088).start();
    }
}

• A進(jìn)程接受B進(jìn)程的響應(yīng)，同時(shí)被喚醒，然后以上 lock.wait() 以后的代碼得以繼續(xù)執(zhí)行。

public class AProcessClientHandler extends SimpleChannelInboundHandler {

    private final Object lock;

    public AProcessClientHandler(Object lock) {
        this.lock = lock;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        String resData = msg.toString(CharsetUtil.UTF_8);
        System.out.println("A進(jìn)程接受到了響應(yīng)數(shù)據(jù): " + resData);

        // B 進(jìn)程任務(wù)完成，使用 notify 喚醒等待的線程
        synchronized (lock) {
            lock.notify();  // 喚醒 A 進(jìn)程
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

3、總結(jié)

Dubbo的優(yōu)秀設(shè)計(jì)思路有許多，我只鐘情其一，那就是RPC的調(diào)用過(guò)程。以上是一個(gè)簡(jiǎn)易的RPC遠(yuǎn)程調(diào)用的示例，用于理解Dubbo的原理和源碼，希望對(duì)你有幫助！

本篇完結(jié)！歡迎 關(guān)注、加V(yclxiao)交流、全網(wǎng)可搜(程序員半支煙)

原文鏈接： https://mp.weixin.qq.com/s/J0fzDH-iqGnnnjqaXMLs-A