用netty实现rpc调用

netty是一个nio的框架，这个与原生的java的nio的不同之处在于原生的nio比较难以使用，通过netty可以不用了解底层的编写，快速实现网络编程，更快实现我们需要的业务//虽然我推荐多了解底层
netty的使用比较广泛，比如游戏的数据包传输，还有rpc的调用，比如dubbo。因为个人倾向了解一样东西可以先从demo入手，因此决定写一个简单的rpc调用来了解netty，show code！
先新建一个maven项目，在里面创建三个moudle，分别是start_interface，start_rpc_consumer和start_rpc_provider。在interface的pom.xml加依赖

  <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.16.Final</version>
        </dependency>

然后新建RPCClient作为客户端的使用

package com.example.start_interface;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
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;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.lang.reflect.Proxy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RPCClient {
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static RPCClientHandler client;


    /**
     * 创建一个代理对象
     */
    public Object createProxy(final Class<?> serviceClass, final String providerName
            , final RPCClientHandler rpcClientHandler, String ip, int port) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class<?>[]{serviceClass}, (proxy, method, args) -> {
                    if (client == null) {
                        initClient(rpcClientHandler, ip, port);
                    }
                    

                    String param = providerName;
                    String[] params = param.split("#");

                    String className = params[0];
                    String methodName = params[1];
                    RPCParam rpcParam =new RPCParam();
                    rpcParam.setClazzName(className);
                    rpcParam.setMethodName(methodName);
                    rpcParam.initParams(args.length);
                    for (int i = 0; i < args.length; i++) {
                        rpcParam.getParams()[i]=args[i];
                        rpcParam.getParamTypes()[i] = args[i].getClass();
                    }
                    // 设置参数
                    client.setRpcParam(rpcParam);
                    return executor.submit(client).get();
                });
    }

    /**
     * 初始化客户端
     */
    private static void initClient(RPCClientHandler rpcClientHandler, String ip, int port) {
        client = rpcClientHandler;
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new ObjectEncoder());
                        p.addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
                        p.addLast(client);
                    }
                });
        try {
            b.connect(ip, port).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

以及他的Handler

package com.example.start_interface;

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

import java.util.concurrent.Callable;

public class RPCClientHandler extends ChannelInboundHandlerAdapter implements Callable {
    private ChannelHandlerContext context;
    //结果
    private Object result;
    //参数
    private RPCParam rpcParam;

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        context = ctx;
    }

    /**
     * 收到服务端数据，唤醒等待线程
     */
    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) {
        result = msg;
        notify();
    }

    /**
     * 写出数据，开始等待唤醒
     */
    @Override
    public synchronized Object call() throws InterruptedException {
   
        context.writeAndFlush(rpcParam);
        //context.writeAndFlush("as");
        wait();
        return result;
    }

    public RPCParam getRpcParam() {
        return rpcParam;
    }

    public void setRpcParam(RPCParam rpcParam) {
        this.rpcParam = rpcParam;
    }
}

以及RPCServer

package com.example.start_interface;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
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;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

public class RPCServer {

    public static void startServer(String hostName, int port) {
        try {

            ServerBootstrap bootstrap = new ServerBootstrap();
            NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
            bootstrap.group(eventLoopGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(new ObjectEncoder());
                            p.addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
                            p.addLast(new RPCServerHandler());
                        }
                    });
            bootstrap.bind(hostName, port).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

以及RPCServerHandler

package com.example.start_interface;

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

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

public class RPCServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
       
        System.out.println(msg.toString());
        RPCParam rpcParam = (RPCParam) msg;
        String className =  rpcParam.getClazzName();
        String methodName = rpcParam.getMethodName();

        try {
            Class clazz = Class.forName(className);
            Object object = clazz.newInstance();

            Method method = object.getClass().getDeclaredMethod(methodName,rpcParam.getParamTypes());

            Object result = method.invoke(object,rpcParam.getParams());
            ctx.writeAndFlush(result);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InstantiationException e) {
            e.printStackTrace();
        } catch (NoSuchMethodException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        }

    }
}

其中为了获取参数，我们需要自己写一个param，来方便传输得到方法和参数

package com.example.start_interface;

import java.io.Serializable;

public class RPCParam implements Serializable {
    private String clazzName;
    private String methodName;
    private Object[] params;
    private Class<?>[] paramTypes;

    public String getClazzName() {
        return clazzName;
    }

    public void setClazzName(String clazzName) {
        this.clazzName = clazzName;
    }

    public String getMethodName() {
        return methodName;
    }

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

    public Object[] getParams() {
        return params;
    }

    public void setParams(Object[] params) {
        this.params = params;
    }

    public void initParams(int size){
        params = new Object[size];
        paramTypes = new Class[size];
    }

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

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

其实返回结果感觉也是应该自己写一个了，但是因为是一个练手demo，就算了,然后再加一个interface

package com.example.start_interface;

public interface UserService {
    /**
     * helloWorld
     * @param word
     * @return
     */
    String sayHello(String word);

    String test(Integer i);
}

这时的整体框架算是搭建好了，这时候开始测试一下
这时候在consume包里实现调用

package com.example.start_rpc_consumer;

import com.example.start_interface.RPCClient;
import com.example.start_interface.RPCClientHandler;
import com.example.start_interface.UserService;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class StartRpcConsumerApplication {

    public static final String providerName = "com.example.start_rpc_provider.UserServiceImpl#test#";


    public static void main(String[] args) throws InterruptedException {
        SpringApplication.run(StartRpcConsumerApplication.class, args);

        RPCClient rpcClient = new RPCClient();
        UserService userService = (UserService) rpcClient.createProxy(UserService.class,providerName
                ,new RPCClientHandler(),"localhost", 8888);
        for (;;) {
            System.out.println("ok???");
            Thread.sleep(1000);
            System.out.println(userService.test(12));
            System.out.println("ok");
        }
    }

}

在provider包实现impl的类以及开启rpc服务

package com.example.start_rpc_provider;

import com.example.start_interface.UserService;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

public class UserServiceImpl implements UserService {
    @Override
    public String sayHello(String word) {
        System.out.println("调用成功--参数：" + word);
        return "调用成功--参数：" + word;
    }

    @Override
    public String test(Integer i){
        return "canshu"+i;
    }

   
}

package com.example.start_rpc_provider;

import com.example.start_interface.RPCServer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class StartRpcProviderApplication {

    public static void main(String[] args) {
        SpringApplication.run(StartRpcProviderApplication.class, args);
        //UserServiceImpl.startServer("localhost", 8990);
        RPCServer.startServer("localhost", 8888);
    }

}

以下是输出结果

image.png

其实在这个demo里面学到了不少知识，除了粘包拆包以及netty的调用，还有就是动态代理以及反射机制，其中比较有意思的是，这个其实并不支持基本数据类型，只支持object类，我在网上看到的比较有意思的解决方法是，1.默认以后的参数都使用object类不使用基本数据类型。2写一个工具把基本数据类型进行装箱，这个就以后有机会再玩玩啦。
这个是根据之前的一篇博客进行改写的，但是忘了那个网址了，希望有知道的童鞋可以告诉我然后我再加上出处

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 212,657评论 6赞 492
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 90,662评论 3赞 385
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 158,143评论 0赞 348
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 56,732评论 1赞 284
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 65,837评论 6赞 386
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 50,036评论 1赞 291
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 39,126评论 3赞 410
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 37,868评论 0赞 268
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 44,315评论 1赞 303
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 36,641评论 2赞 327
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,773评论 1赞 341
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 34,470评论 4赞 333
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 40,126评论 3赞 317
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,859评论 0赞 21
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 32,095评论 1赞 267
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 46,584评论 2赞 362
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 43,676评论 2赞 351

用netty实现rpc调用

推荐阅读更多精彩内容