Netty实现远程调用RPC功能

添加依赖

<dependency>
    <groupid>io.netty</groupid>
    <artifactid>netty-all</artifactid>
    <version>4.1.2.final</version>
</dependency>

<dependency>
    <groupid>org.reflections</groupid>
    <artifactid>reflections</artifactid>
    <version>0.9.10</version>
</dependency>

组织架构

服务端

封装类信息

public class classinfo implements serializable {

    private static final long serialversionuid = 1l;

    private string classname;  //类名
    private string methodname;//方法名
    private class<?>[] types; //参数类型
    private object[] objects;//参数列表

    public string getclassname() {
        return classname;
    }

    public void setclassname(string classname) {
        this.classname = classname;
    }

    public string getmethodname() {
        return methodname;
    }

    public void setmethodname(string methodname) {
        this.methodname = methodname;
    }

    public class<?>[] gettypes() {
        return types;
    }

    public void settypes(class<?>[] types) {
        this.types = types;
    }

    public object[] getobjects() {
        return objects;
    }

    public void setobjects(object[] objects) {
        this.objects = objects;
    }
}

服务端网络处理服务器

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

    public void start() {
        eventloopgroup bossgroup = new nioeventloopgroup();
        eventloopgroup workergroup = new nioeventloopgroup();
        try {
            serverbootstrap serverbootstrap = new serverbootstrap();
            serverbootstrap.group(bossgroup, workergroup)
                    .channel(nioserversocketchannel.class)
                    .option(channeloption.so_backlog, 128)
                    .childoption(channeloption.so_keepalive, true)
                    .localaddress(port).childhandler(
                            new channelinitializer<socketchannel>() {
                                @override
                                protected void initchannel(socketchannel ch) throws exception {
                                    channelpipeline pipeline = ch.pipeline();
                                    //编码器
                                    pipeline.addlast("encoder", new objectencoder());
                                    //解码器
                                    pipeline.addlast("decoder", new objectdecoder(integer.max_value, classresolvers.cachedisabled(null)));
                                    //服务器端业务处理类
                                    pipeline.addlast(new invokehandler());
                                }
                            });
            channelfuture future = serverbootstrap.bind(port).sync();
            system.out.println("......server is ready......");
            future.channel().closefuture().sync();
        } catch (exception e) {
            bossgroup.shutdowngracefully();
            workergroup.shutdowngracefully();
        }
    }

    public static void main(string[] args) throws exception {
        new nettyrpcserver(9999).start();
    }
}

服务器端业务处理类

public class invokehandler extends channelinboundhandleradapter {
    //得到某接口下某个实现类的名字
    private string getimplclassname(classinfo classinfo) throws exception{
        //服务方接口和实现类所在的包路径
        string interfacepath="com.lyz.server";
        int lastdot = classinfo.getclassname().lastindexof(".");
        string interfacename=classinfo.getclassname().substring(lastdot);
        class superclass=class.forname(interfacepath+interfacename);
        reflections reflections = new reflections(interfacepath);
        //得到某接口下的所有实现类
        set<class> implclassset=reflections.getsubtypesof(superclass);
        if(implclassset.size()==0){
            system.out.println("未找到实现类");
            return null;
        }else if(implclassset.size()>1){
            system.out.println("找到多个实现类，未明确使用哪一个");
            return null;
        }else {
            //把集合转换为数组
            class[] classes=implclassset.toarray(new class[0]);
            return classes[0].getname(); //得到实现类的名字
        }
    }

    @override  //读取客户端发来的数据并通过反射调用实现类的方法
    public void channelread(channelhandlercontext ctx, object msg) throws exception {
        classinfo classinfo = (classinfo) msg;
        system.out.println(classinfo);
        object clazz = class.forname(getimplclassname(classinfo)).newinstance();
        method method = clazz.getclass().getmethod(classinfo.getmethodname(), classinfo.gettypes());
        //通过反射调用实现类的方法
        object result = method.invoke(clazz, classinfo.getobjects());
        ctx.writeandflush(result);
    }
}

服务端接口及实现类

// 无参接口
public interface hellonetty {
    string hello();
}

// 实现类
public class hellonettyimpl implements hellonetty {
    @override
    public string hello() {
        return "hello,netty";
    }
}

// 带参接口
public interface hellorpc {
    string hello(string name);
}

// 实现类
public class hellorpcimpl implements hellorpc {
    @override
    public string hello(string name) {
        return "hello," + name;
    }
}

客户端

代理类

public class nettyrpcproxy {
    //根据接口创建代理对象
    public static object create(class target) {
        return proxy.newproxyinstance(target.getclassloader(), new class[]{target}, new invocationhandler() {
            @override
            public object invoke(object proxy, method method, object[] args)
                    throws throwable {
                //封装classinfo
                classinfo classinfo = new classinfo();
                classinfo.setclassname(target.getname());
                classinfo.setmethodname(method.getname());
                classinfo.setobjects(args);
                classinfo.settypes(method.getparametertypes());

                //开始用netty发送数据
                eventloopgroup group = new nioeventloopgroup();
                resulthandler resulthandler = new resulthandler();
                try {
                    bootstrap b = new bootstrap();
                    b.group(group)
                            .channel(niosocketchannel.class)
                            .handler(new channelinitializer<socketchannel>() {
                                @override
                                public void initchannel(socketchannel ch) throws exception {
                                    channelpipeline pipeline = ch.pipeline();
                                    //编码器
                                    pipeline.addlast("encoder", new objectencoder());
                                    //解码器  构造方法第一个参数设置二进制数据的最大字节数  第二个参数设置具体使用哪个类解析器
                                    pipeline.addlast("decoder", new objectdecoder(integer.max_value, classresolvers.cachedisabled(null)));
                                    //客户端业务处理类
                                    pipeline.addlast("handler", resulthandler);
                                }
                            });
                    channelfuture future = b.connect("127.0.0.1", 9999).sync();
                    future.channel().writeandflush(classinfo).sync();
                    future.channel().closefuture().sync();
                } finally {
                    group.shutdowngracefully();
                }
                return resulthandler.getresponse();
            }
        });
    }
}

客户端业务处理类

public class resulthandler extends channelinboundhandleradapter {

    private object response;
    public object getresponse() {
        return response;
    }

    @override //读取服务器端返回的数据(远程调用的结果)
    public void channelread(channelhandlercontext ctx, object msg) throws exception {
        response = msg;
        ctx.close();
    }
}

客户端接口

// 无参接口
public interface hellonetty {
    string hello();
}

// 带参接口
public interface hellorpc {
    string hello(string name);
}

测试类 服务调用方

public class testnettyrpc {
    public static void main(string [] args){

        //第1次远程调用
        hellonetty hellonetty=(hellonetty) nettyrpcproxy.create(hellonetty.class);
        system.out.println(hellonetty.hello());

        //第2次远程调用
        hellorpc hellorpc =  (hellorpc) nettyrpcproxy.create(hellorpc.class);
        system.out.println(hellorpc.hello("rpc"));

    }
}

输出结果

服务端

......server is ready......
com.lyz.serverstub.classinfo@2b894733
com.lyz.serverstub.classinfo@167bfa9

客户端

hello,netty
hello,rpc

下一篇通过netty实现线上聊天功能