實(shí)現(xiàn)一個(gè)簡(jiǎn)單Dubbo完整過(guò)程詳解

更新時(shí)間：2023年07月13日 10:42:13 作者：pq217

這篇文章主要為大家介紹了實(shí)現(xiàn)一個(gè)簡(jiǎn)單Dubbo完整過(guò)程詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

Dubbo

Dubbo最早的定位是rpc框架，即遠(yuǎn)程服務(wù)調(diào)用，解決的是跨服務(wù)之間的方法調(diào)用問(wèn)題，本文還是在這個(gè)定位基礎(chǔ)上嘗試手寫(xiě)一個(gè)簡(jiǎn)單的Dubbo

需求

首先要搭建測(cè)試的項(xiàng)目結(jié)構(gòu)，兩個(gè)服務(wù)consumer和provider，分別代表調(diào)用方和提供方，二者功能依賴(lài)于interface，其中暴露接口

interface包中定義一個(gè)接口

// interface
public interface HelloService {
    String sayHello(String name);
}

provider實(shí)現(xiàn)

// provider
public class HelloServiceImpl implements HelloService {
    public String sayHello(String name) {
        return "hello "+name;
    }
}

consumer調(diào)用

// consumer
public class Consumer {
    public static void main(String[] args) {
        // todo 獲取不到HelloService的實(shí)現(xiàn)
        HelloService helloService = null;
        System.out.println(helloService.sayHello("pq"));
    }
}

當(dāng)前的需求即consumer服務(wù)調(diào)用provider服務(wù)里sayHello方法的實(shí)現(xiàn)，顯然當(dāng)前無(wú)法實(shí)現(xiàn)，這是一種遠(yuǎn)程發(fā)放調(diào)用，我們?cè)谛陆ㄒ粋€(gè)Module命名為dubbo，意圖通過(guò)依賴(lài)它來(lái)實(shí)現(xiàn)遠(yuǎn)程方法的調(diào)用

dubbo

網(wǎng)絡(luò)

由于跨服務(wù)了，所以遠(yuǎn)程調(diào)用必然是要走網(wǎng)絡(luò)的，dubbo使用了netty，我們也用netty來(lái)實(shí)現(xiàn)通訊

首先定義網(wǎng)絡(luò)請(qǐng)求的數(shù)據(jù)，遠(yuǎn)程調(diào)用需要的信息：哪個(gè)類(lèi)，哪個(gè)方法，什么參數(shù)，我們把這些信息封裝一下

// dubbo
@Data
@AllArgsConstructor
public class Invocation implements Serializable {
    private String className;
    private String methodName;
    private Class&lt;?&gt;[] paramTypes;
    private Object[] args;
}

服務(wù)端

provider作為服務(wù)的提供方，需要依靠netty搭建一個(gè)服務(wù)器，當(dāng)接受到請(qǐng)求(Invocation對(duì)象)時(shí)，可以根據(jù)className，methodName等信息找到對(duì)應(yīng)的本地方法進(jìn)行調(diào)用

所以provider首先要維護(hù)一個(gè)map存儲(chǔ)className和class的對(duì)應(yīng)關(guān)系，這樣在收到請(qǐng)求時(shí)可以通過(guò)className找到對(duì)應(yīng)的類(lèi)，再通過(guò)反射獲取對(duì)應(yīng)的方法進(jìn)行調(diào)用

在我們的dubbo框架中封裝這么一個(gè)map結(jié)構(gòu)供provider使用

// dubbo
public class LocalRegister {
    private static Map&lt;String, Object&gt; map = new HashMap&lt;String, Object&gt;();
    public static void register(String className, Object impl) {
        map.put(className, impl);
    }
    public static Object get(String className) {
        return map.get(className);
    }
}

然后再做一個(gè)處理請(qǐng)求netty服務(wù)供provider使用

// dubbo
public class NettyServer {
    public void start(Integer port) {
        try {
            final ServerBootstrap bootstrap = new ServerBootstrap();
            EventLoopGroup bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("bossGroup", true));
            EventLoopGroup workerGroup = new NioEventLoopGroup(10, new DefaultThreadFactory("workerGroup", true));
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer&lt;SocketChannel&gt;() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            channel.pipeline().addLast("decoder", new ObjectDecoder(ClassResolvers
                                    .weakCachingConcurrentResolver(this.getClass()
                                            .getClassLoader())));
                            channel.pipeline().addLast("encoder", new ObjectEncoder());
                            channel.pipeline().addLast("handler", new RequestHandler());
                        }
                    });
            ChannelFuture cf = bootstrap.bind(port).sync();
            cf.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

對(duì)應(yīng)的handler如下

// dubbo
public class RequestHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Invocation invocation = (Invocation) msg;
        // 根據(jù)className獲取寄存的服務(wù)對(duì)象
        Object serviceImpl = LocalRegister.get(invocation.getClassName());
        // 通過(guò)methodName等信息獲取對(duì)應(yīng)的方法
        Method method = serviceImpl.getClass().getMethod(invocation.getMethodName(), invocation.getParamTypes());
        // 調(diào)用方法
        Object result = method.invoke(serviceImpl, invocation.getArgs());
        // 返回服務(wù)結(jié)果
        ctx.writeAndFlush(result);
    }
}

provider啟動(dòng)類(lèi)Starter

// provider
public class Starter {
    public static void main(String[] args) {
        // 存儲(chǔ)服務(wù)于名字映射關(guān)系
        HelloServiceImpl helloService = new HelloServiceImpl();
        String className = HelloService.class.getName();
        LocalRegister.register(className, helloService);
        // 開(kāi)啟netty服務(wù)
        NettyServer nettyServer = new NettyServer();
        System.out.println("provider 端口號(hào)9001");
        nettyServer.start(9001);
    }
}

代理

consumer只能拿到到HelloService接口，那么實(shí)例化的方法可以采用jdk動(dòng)態(tài)代理生成代理實(shí)現(xiàn)，而代理的實(shí)際執(zhí)行方式是通過(guò)netty網(wǎng)絡(luò)發(fā)送請(qǐng)求給provider，

首先還是在dubbo框架中封裝一個(gè)netty的客戶端供consumer發(fā)起請(qǐng)求

// dubbo
@Setter
public class NettyClient {
    /**
     * 管道上下文
     */
    private volatile ChannelHandlerContext channelHandlerContext;
    /**
     * 返回消息暫存
     */
    private Object message;
    public void start(String hostName, Integer port) {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer&lt;SocketChannel&gt;() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            channel.pipeline().addLast("decoder", new ObjectDecoder(ClassResolvers
                                    .weakCachingConcurrentResolver(this.getClass()
                                            .getClassLoader())));
                            channel.pipeline().addLast("encoder", new ObjectEncoder());
                            channel.pipeline().addLast(new ResponseHandler(NettyClient.this));
                        }
                    });
            bootstrap.connect(hostName, port).sync();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /**
     * 發(fā)送遠(yuǎn)程調(diào)用
     * @param hostName
     * @param port
     * @param invocation
     * @return
     */
    public synchronized String send(String hostName, Integer port, Invocation invocation) {
        start(hostName, port);
        try {
            wait();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 發(fā)送數(shù)據(jù)
        channelHandlerContext.writeAndFlush(invocation);
        // 等待
        try {
            wait();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 返回?cái)?shù)據(jù)
        return message.toString();
    }
}

其中的ResponseHandler入下

// dubbo
public class ResponseHandler extends ChannelInboundHandlerAdapter {
    private final NettyClient client;
    public ResponseHandler(NettyClient client) {
        this.client = client;
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        synchronized (client) {
            client.notify();
        }
        client.setChannelHandlerContext(ctx);
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        client.setMessage(msg);
        synchronized (client) {
            client.notify();
        }
    }
}

然后在我們的dubbo框架中實(shí)現(xiàn)創(chuàng)建代理

// dubbo
public class ProxyFactory {
    /**
     * 根據(jù)接口創(chuàng)建代理 jdk動(dòng)態(tài)代理
     * @param interfaceClass
     * @param <T>
     * @return
     */
    public static <T> T getProxy(final Class<T> interfaceClass) {
        return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class[]{interfaceClass}, new InvocationHandler() {
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 請(qǐng)求封裝成對(duì)象
                Invocation invocation = new Invocation(interfaceClass.getName(), method.getName(), method.getParameterTypes(), args);
                NettyClient nettyClient = new NettyClient();
                // 發(fā)起網(wǎng)絡(luò)請(qǐng)求
                String response = nettyClient.send("127.0.0.1", 9001, invocation);
                return response;
            }
        });
    }
}

最后回到consumer添加啟動(dòng)類(lèi)，通過(guò)代理創(chuàng)建HelloService的實(shí)現(xiàn)，嘗試調(diào)用provider的sayHello方法

// consumer
public class Consumer {
    public static void main(String[] args) {
        HelloService helloService = ProxyFactory.getProxy(HelloService.class);
        System.out.println(helloService.sayHello("pq"));
    }
}