使用Netty搭建Dubbo

前言

Dubbo是基于Netty搭建的RPC框架,為了更好地理解Netty在Dubbo中的應(yīng)用沫浆,仿照Dubbo搭建了一個(gè)簡易版的RPC框架眠冈。

概述

整個(gè)調(diào)用邏輯如下:
1旋膳、生產(chǎn)者服務(wù)端啟動(dòng)Netty服務(wù)端。
2熊户、消費(fèi)者客戶端通過JDK動(dòng)態(tài)代理啟動(dòng)Netty客戶端萍膛,通過注冊(cè)中心地址連接生產(chǎn)者服務(wù)端,同時(shí)將接口調(diào)用信息(接口嚷堡、方法蝗罗、參數(shù)等)先序列化再發(fā)送給生產(chǎn)者服務(wù)端。
3蝌戒、生產(chǎn)者服務(wù)端接收消息并通過反射調(diào)用相應(yīng)方法绿饵,然后返回調(diào)用結(jié)果給消費(fèi)者。
4瓶颠、消費(fèi)者接收生產(chǎn)者傳來的調(diào)用結(jié)果拟赊。

實(shí)現(xiàn)

新建DubboRequest類(相當(dāng)于POJO),作為消息載體

package com.beidao.netty.dubbo.facade.api;

import java.io.Serializable;
import java.util.Arrays;

/**
 * dubbo請(qǐng)求類
 * @author 0200759
 *
 */
public class DubboRequest implements Serializable{

    private static final long serialVersionUID = 422805234202183587L;
    private Class<?> interfaceClass;
    private String methodName;
    private Class<?>[] paramTypes;
    private Object[] args;
    
    public DubboRequest(Class<?> interfaceClass, String methodName, Class<?>[] paramTypes, Object[] args) {

        this.interfaceClass = interfaceClass;
        this.methodName = methodName;
        this.paramTypes = paramTypes;
        this.args = args;
    }

    public Class<?> getInterfaceClass() {
        return interfaceClass;
    }

    public void setInterfaceClass(Class<?> interfaceClass) {
        this.interfaceClass = interfaceClass;
    }

    public String getMethodName() {
        return methodName;
    }

    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }

    public Class<?>[] getParamTypes() {
        return paramTypes;
    }

    public void setParamTypes(Class<?>[] paramTypes) {
        this.paramTypes = paramTypes;
    }

    public Object[] getArgs() {
        return args;
    }

    public void setArgs(Object[] args) {
        this.args = args;
    }

    @Override
    public String toString() {
        return "DubboRequest{" +
                "interfaceClass=" + interfaceClass +
                ", methodName='" + methodName + '\'' +
                ", paramTypes=" + Arrays.toString(paramTypes) +
                ", args=" + Arrays.toString(args) +
                '}';
    }
}

新建Dubbo消費(fèi)者調(diào)用接口IUserFacade

package com.beidao.netty.dubbo.facade.api;

/**
 * dubbo api接口
 * @author 0200759
 *
 */
public interface IUserFacade {

    /**
     * 返回用戶名接口
     * @param string 
     * @return
     */
    public String getUserName(Long id);

}

新建Dubbo消費(fèi)者攔截器

package com.beidao.netty.dubbo.client;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;

import com.beidao.netty.dubbo.facade.api.DubboRequest;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

/**
 * dubbo消費(fèi)者攔截器
 * @author 0200759
 *
 */
public class DubboConsumerHandler implements InvocationHandler{

    private Object res;
    
    public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<SocketChannel>() {
                
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new ObjectDecoder(1024, ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
                    ch.pipeline().addLast(new ObjectEncoder());
                    ch.pipeline().addLast(new ConsumerHandler(proxy, method, args));
                }
            });
            //從注冊(cè)中心獲取服務(wù)端ip和端口
            ChannelFuture f = bootstrap.connect("127.0.0.1", 8080).sync();
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
        return res;
    }
    
    /**
     * 
     * netty-dubbo消費(fèi)者攔截器
     * @author 0200759
     *
     */
    private class ConsumerHandler extends ChannelInboundHandlerAdapter{
        private Object proxy;
        private Method method;
        private Object[] args;
        
        public ConsumerHandler(Object proxy, Method method, Object[] args) {
            this.proxy = proxy;
            this.args = args;
            this.method = method;
        }
        
        public void channelActive(ChannelHandlerContext ctx) {
             // 傳輸?shù)膶?duì)象必須實(shí)現(xiàn)序列化接口 包括其中的屬性
            DubboRequest req = new DubboRequest(proxy.getClass().getInterfaces()[0], method.getName(), method.getParameterTypes(), args);
            ctx.write(req);
            ctx.flush();
        }
        
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("調(diào)用成功");
            res = msg;
            ctx.flush();
            //收到響應(yīng)后斷開連接
            ctx.close();
        }
        
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    }

}

新建Dubbo服務(wù)代理類

package com.beidao.netty.dubbo.client;

import java.lang.reflect.Proxy;

/**
 * Dubbo代理類
 * @author 0200759
 *
 */
public class DubboProxy {

    public static Object getProxyInstance(Class<?> clazz) {
        return Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, new DubboConsumerHandler());
    }
}

新建Dubbo消費(fèi)者

package com.beidao.netty.dubbo.client;

import com.beidao.netty.dubbo.facade.api.IUserFacade;

/**
 * dubbo客戶端(消費(fèi)者)
 * @author 0200759
 *
 */
public class DubboClient {
 
    public static void main(String[] args){
        IUserFacade userFacade = (IUserFacade) DubboProxy.getProxyInstance(IUserFacade.class);
        
        System.out.println(userFacade.getUserName(520L));
        System.out.println(userFacade.getUserName(1314L));
        System.out.println(userFacade.getUserName(1314520L));
    }
}

新建 dubbo服務(wù)端實(shí)現(xiàn)類

package com.beidao.netty.dubbo.facade.impl;

import com.beidao.netty.dubbo.facade.api.IUserFacade;

/**
 * dubbo服務(wù)端實(shí)現(xiàn)類
 * @author 0200759
 *
 */
public class UserFacade implements IUserFacade {

    public String getUserName(Long id) {
        
        return "I love you, "+id;
    }

}

dubbo生產(chǎn)者攔截器

package com.beidao.netty.dubbo.sever;

import java.lang.reflect.Method;

import com.beidao.netty.dubbo.facade.api.DubboRequest;
import com.beidao.netty.dubbo.facade.api.IUserFacade;
import com.beidao.netty.dubbo.facade.impl.UserFacade;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * netty-dubbo服務(wù)端攔截器
 * @author 0200759
 *
 */
public class DubboServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("服務(wù)端收到消息:   " + msg);
        DubboRequest req = (DubboRequest) msg;
        // 1. 根據(jù)類名返回對(duì)象
        Object target = this.getInstenceByInterfaceClass(req.getInterfaceClass());
        // 2. 獲取方法名
        String methodName = req.getMethodName();
        // 3. 獲取方法參數(shù)類型
        // 4. 獲取方法
        Method method = target.getClass().getMethod(methodName, req.getParamTypes());
        // 5. 獲取參數(shù)值
        //調(diào)用方法 獲取返回值
        Object res = method.invoke(target, req.getArgs());
        // 寫回給調(diào)用端
        ctx.writeAndFlush(res);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    /**
     * 根據(jù)接口返回對(duì)應(yīng)的實(shí)例
     * @param clazz
     * @return
     */
    private Object getInstenceByInterfaceClass(Class<?> clazz) {
        if (IUserFacade.class.equals(clazz)) {
            return new UserFacade();
        }
        return null;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }
}

Dubbo生產(chǎn)者服務(wù)端

package com.beidao.netty.dubbo.sever;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

/**
 * @author 0200759
 *
 */
public class DubboServer {

    private int port;

    public DubboServer(int port) {
        this.port = port;
    }

    public void run() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(); 
        try {
            ServerBootstrap b = new ServerBootstrap(); 
            b.group(bossGroup)
                    .channel(NioServerSocketChannel.class) 
                    .childHandler(new ChannelInitializer<SocketChannel>() { 
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new ObjectDecoder(1024*1024, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
                            ch.pipeline().addLast(new ObjectEncoder());
                            ch.pipeline().addLast(new DubboServerHandler());
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 128)          
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            // Bind and start to accept incoming connections.
            ChannelFuture f = b.bind(port).sync(); // (7)

            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        new DubboServer(8080).run();
    }

}

驗(yàn)證

1粹淋、啟動(dòng)生產(chǎn)者服務(wù)端DubboServer吸祟。
2、啟動(dòng)消費(fèi)者調(diào)用端DubboClient桃移。
消費(fèi)者客戶端控制臺(tái)顯示如下:

調(diào)用成功
I love you, 520
調(diào)用成功
I love you, 1314
調(diào)用成功
I love you, 1314520

生產(chǎn)者服務(wù)端控制臺(tái)顯示如下:

服務(wù)端收到消息:   DubboRequest{interfaceClass=interface com.beidao.netty.dubbo.facade.api.IUserFacade, methodName='getUserName', paramTypes=[class java.lang.Long], args=[520]}
服務(wù)端收到消息:   DubboRequest{interfaceClass=interface com.beidao.netty.dubbo.facade.api.IUserFacade, methodName='getUserName', paramTypes=[class java.lang.Long], args=[1314]}
服務(wù)端收到消息:   DubboRequest{interfaceClass=interface com.beidao.netty.dubbo.facade.api.IUserFacade, methodName='getUserName', paramTypes=[class java.lang.Long], args=[1314520]}

源碼地址:https://github.com/MAXAmbitious/netty-study/tree/master/netty-dubbo

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
  • 序言:七十年代末屋匕,一起剝皮案震驚了整個(gè)濱河市,隨后出現(xiàn)的幾起案子借杰,更是在濱河造成了極大的恐慌过吻,老刑警劉巖,帶你破解...
    沈念sama閱讀 218,858評(píng)論 6 508
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件,死亡現(xiàn)場(chǎng)離奇詭異纤虽,居然都是意外死亡乳绕,警方通過查閱死者的電腦和手機(jī),發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 93,372評(píng)論 3 395
  • 文/潘曉璐 我一進(jìn)店門逼纸,熙熙樓的掌柜王于貴愁眉苦臉地迎上來洋措,“玉大人,你說我怎么就攤上這事杰刽〔しⅲ” “怎么了?”我有些...
    開封第一講書人閱讀 165,282評(píng)論 0 356
  • 文/不壞的土叔 我叫張陵贺嫂,是天一觀的道長滓鸠。 經(jīng)常有香客問我,道長第喳,這世上最難降的妖魔是什么糜俗? 我笑而不...
    開封第一講書人閱讀 58,842評(píng)論 1 295
  • 正文 為了忘掉前任,我火速辦了婚禮墩弯,結(jié)果婚禮上吩跋,老公的妹妹穿的比我還像新娘。我一直安慰自己渔工,他們只是感情好锌钮,可當(dāng)我...
    茶點(diǎn)故事閱讀 67,857評(píng)論 6 392
  • 文/花漫 我一把揭開白布。 她就那樣靜靜地躺著引矩,像睡著了一般梁丘。 火紅的嫁衣襯著肌膚如雪。 梳的紋絲不亂的頭發(fā)上旺韭,一...
    開封第一講書人閱讀 51,679評(píng)論 1 305
  • 那天氛谜,我揣著相機(jī)與錄音,去河邊找鬼区端。 笑死值漫,一個(gè)胖子當(dāng)著我的面吹牛,可吹牛的內(nèi)容都是我干的织盼。 我是一名探鬼主播杨何,決...
    沈念sama閱讀 40,406評(píng)論 3 418
  • 文/蒼蘭香墨 我猛地睜開眼,長吁一口氣:“原來是場(chǎng)噩夢(mèng)啊……” “哼沥邻!你這毒婦竟也來了危虱?” 一聲冷哼從身側(cè)響起,我...
    開封第一講書人閱讀 39,311評(píng)論 0 276
  • 序言:老撾萬榮一對(duì)情侶失蹤唐全,失蹤者是張志新(化名)和其女友劉穎埃跷,沒想到半個(gè)月后,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體,經(jīng)...
    沈念sama閱讀 45,767評(píng)論 1 315
  • 正文 獨(dú)居荒郊野嶺守林人離奇死亡弥雹,尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 37,945評(píng)論 3 336
  • 正文 我和宋清朗相戀三年垃帅,在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了。 大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片缅糟。...
    茶點(diǎn)故事閱讀 40,090評(píng)論 1 350
  • 序言:一個(gè)原本活蹦亂跳的男人離奇死亡挺智,死狀恐怖祷愉,靈堂內(nèi)的尸體忽然破棺而出窗宦,到底是詐尸還是另有隱情,我是刑警寧澤二鳄,帶...
    沈念sama閱讀 35,785評(píng)論 5 346
  • 正文 年R本政府宣布赴涵,位于F島的核電站,受9級(jí)特大地震影響订讼,放射性物質(zhì)發(fā)生泄漏髓窜。R本人自食惡果不足惜,卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 41,420評(píng)論 3 331
  • 文/蒙蒙 一欺殿、第九天 我趴在偏房一處隱蔽的房頂上張望寄纵。 院中可真熱鬧,春花似錦脖苏、人聲如沸程拭。這莊子的主人今日做“春日...
    開封第一講書人閱讀 31,988評(píng)論 0 22
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽恃鞋。三九已至,卻和暖如春亦歉,著一層夾襖步出監(jiān)牢的瞬間恤浪,已是汗流浹背。 一陣腳步聲響...
    開封第一講書人閱讀 33,101評(píng)論 1 271
  • 我被黑心中介騙來泰國打工肴楷, 沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留水由,地道東北人。 一個(gè)月前我還...
    沈念sama閱讀 48,298評(píng)論 3 372
  • 正文 我出身青樓赛蔫,卻偏偏與公主長得像砂客,于是被迫代替她去往敵國和親。 傳聞我的和親對(duì)象是個(gè)殘疾皇子濒募,可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 45,033評(píng)論 2 355

推薦閱讀更多精彩內(nèi)容

  • 0 準(zhǔn)備 安裝注冊(cè)中心:Zookeeper鞭盟、Dubbox自帶的dubbo-registry-simple;安裝Du...
    七寸知架構(gòu)閱讀 13,989評(píng)論 0 88
  • Spring Cloud為開發(fā)人員提供了快速構(gòu)建分布式系統(tǒng)中一些常見模式的工具(例如配置管理瑰剃,服務(wù)發(fā)現(xiàn)齿诉,斷路器,智...
    卡卡羅2017閱讀 134,659評(píng)論 18 139
  • Cora這輩子第一次喝咖啡,是在小升初考試的前幾天粤剧,三合一速溶的那種歇竟。 考前某一晚,鄰居徐阿姨從Cora家后窗悄悄...
    小清澄閱讀 302評(píng)論 0 0
  • 塞北天已寒抵恋,長安秋雨綿焕议。伊人瀟瀟去,美酒愁未蘭弧关。
    無為_768f閱讀 147評(píng)論 0 0
  • 畢業(yè)后世囊,又一次過上校園生活别瞭,該是最后一次了,珍惜著株憾。 生活真的不是你想要怎樣蝙寨,就會(huì)怎樣的,話也真的不要說的太滿...
    小小咚咚鏘閱讀 262評(píng)論 0 0