文档章节

JAVA NIO 编程入门(三)

木木匠
 木木匠
发布于 2018/09/16 16:24
字数 1346
阅读 54
收藏 12

一、前期回顾

上一篇文章JAVA NIO 编程入门(二)学习了NIO的聚集和分散,以及选择器的使用,并通过一个小demo进行实战,演示了如何进行分散和聚集,以及其主要使用场景,本文将是NIO编程入门最后一篇,进行一个RPC简单小demo框架的实现,对前面的知识进行总结性的实战,由于只是演示性质的demo,所以RPC功能并没有考虑很完善,也不涉及到性能等问题考虑。

二、什么是RPC

RPC 英文全称 Remote Procedure Calls,翻译过来就是远程过程调用,是分布式系统中不同节点间流行的通信方式。举例:假设有A服务和B服务分别位于不同的服务器,A服务想调用B服务像调用本地方法一样,这个时候就需要借助RPC方式进行调用。

三、RPC实现

RPC由三个主要部分组成,服务提供者,服务消费者,服务注册中心,服务注册中心提供服务提供者注册服务。客户端和服务端的交互协议采用json的形式,方便演示,同时考虑到复杂性,本次RPC不利用聚集和分散进行协议设计。

  • RPC调用过程解析

  • 请求实体类源码

@Data
public class RpcRquest {
    /**请求id*/
    private String requestId;
    /**请求接口名*/
    private String interfaceName;
    /**服务版本**/
    private String serviceVersion;
    /**方法名*/
    private String methodName;
   /**参数类型*/
    private Class<?>[] parameterTypes;
    /**参数*/
    private Object[] parameters;
}
  • 返回实体类源码

@Data
public class RpcResponse {
    /**请求流水号*/
    private String requestId;
    /**异常*/
    private Exception exception;
    /**返回结果**/
    private Object result;
}
  • 服务发现源码

public class RpcRegister {
    /**存储注册的服务提供实现类*/
    private HashMap<String, Object> registMap = new HashMap<>();
    private  static  RpcRegister register=new RpcRegister();
    public static  RpcRegister buildRegist(){
        return register;
    }
    public RpcRegister regist(String interfaceName,Object obj){
        registMap.put(interfaceName,obj);
        return this;
    }
    public Object findServier(String interfaceName){
        return  registMap.get(interfaceName);
    }
}

这里利用一个map存提供服务的实例,后续再在服务端只需要通过接口就可以查找到对应的实现类。

  • 服务提供者源码

public class ProviderServer implements Runnable {
    /**
     * 服务提供端口
     */
    private int port;


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

    @Override
    public void run() {
        try {
            Selector selector = Selector.open();
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.socket().bind(new InetSocketAddress(port));
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            while (true) {
                int readyChannels = selector.selectNow();
                if (readyChannels == 0) continue;
                Set selectedKeys = selector.selectedKeys();
                Iterator keyIterator = selectedKeys.iterator();
                while (keyIterator.hasNext()) {
                    SelectionKey key = (SelectionKey) keyIterator.next();
                    if (key.isAcceptable()) {
                        ServerSocketChannel serverSocketChannel1 = (ServerSocketChannel) key.channel();
                        SocketChannel socketChannel = serverSocketChannel1.accept();
                        ByteBuffer buf1 = ByteBuffer.allocate(2048);
                        socketChannel.read(buf1);
                        buf1.flip();
                        String reciveStr = new String(buf1.array());
                        if (buf1.hasRemaining()) {
                            System.out.println(">>>服务端收到数据:" + reciveStr);
                            //判断接受的内容是否有结束符,如果有,说明是一个请求结束。
                            if (reciveStr.contains(RpcConstant.PROTOCOL_END)) {
                                RpcRquest req = JSONObject.parseObject(reciveStr.replace(RpcConstant.PROTOCOL_END, ""), RpcRquest.class);
                                RpcResponse res = new RpcResponse();
                                res.setRequestId(req.getRequestId());
                                System.out.println(req.toString());
                                Class<?> remoteInterface = Class.forName(req.getInterfaceName());
                                Method method = remoteInterface.getMethod(req.getMethodName(), req.getParameterTypes());
                                if (null != method) {
                                    Object obj = method.invoke(RpcRegister.buildRegist().findServier(req.getInterfaceName()), req.getParameters());
                                    res.setException(null);
                                    res.setResult(obj);
                                }
                                buf1.clear();
                                buf1.put(JSONObject.toJSON(res).toString().getBytes());
                                buf1.flip();
                                socketChannel.write(buf1);
                            }
                        }
                        socketChannel.close();
                    } else if (key.isConnectable()) {
                    } else if (key.isReadable()) {
                    } else if (key.isWritable()) {

                    }
                    keyIterator.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        } catch (NoSuchMethodException e) {
            e.printStackTrace();
        }

    }

}

这部分源码在《JAVA NIO 编程入门(二)》的基础上增加了反射的部分内容,主要根据接口调用协议,生成客户端需要调用的方法,进行调用,然后将结果返回。

  • 初始化工厂类
public class RpcInitFactory {
    /**
     * 客户端连接远程ip地址
     **/
    private String ip;
    /***远程端口*/
    private int port;


    public RpcInitFactory(String ip, int port) {
        this.ip = ip;
        this.port = port;
    }
}
  • 通用客户端

@Data
public  class CommonClient {
    private  RpcInitFactory factory;

    public CommonClient(RpcInitFactory factory) {
        this.factory = factory;
    }

    public <T> T invoke(RpcRquest req) {
        RpcResponse response = null;
        req.setRequestId(UUID.randomUUID().toString());
        try {
            SocketChannel socketChannel = SocketChannel.open();
            socketChannel.connect(new InetSocketAddress(factory.getIp(), factory.getPort()));
            ByteBuffer buf1 = ByteBuffer.allocate(2048);
            buf1.put(JSONObject.toJSON(req).toString().getBytes());
            buf1.put(RpcConstant.PROTOCOL_END.getBytes());
            buf1.flip();
            if (buf1.hasRemaining())
                socketChannel.write(buf1);
            buf1.clear();

            ByteBuffer body = ByteBuffer.allocate(2048);
            socketChannel.read(body);
            body.flip();
            if (body.hasRemaining()) {
                response = JSONObject.parseObject(new String(body.array()), RpcResponse.class);
            }
            body.clear();
            socketChannel.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return (T) response;
    }
  • 调用客户端接口
public interface Idemo {

    /***加法**/
    public Integer add(Integer i,Integer j);
}
  • 客户端接口实现类
public class DemoRemoteImpl implements Idemo {


    private  CommonClient client;

    public DemoRemoteImpl(CommonClient client) {
        this.client = client;
    }

    @Override
    public Integer add(Integer i, Integer j) {
        //构造rpc请求实体类
        RpcRquest rpcRquest=new RpcRquest();
        //设置版本号
        rpcRquest.setServiceVersion("123");
        //设置调用的接口名称
        rpcRquest.setInterfaceName(Idemo.class.getName());
        //设置调用方法名称
        rpcRquest.setMethodName("add");
        //设置参数
        rpcRquest.setParameters(new Integer[] {i,j});
        //设置参数类型
        rpcRquest.setParameterTypes(new Class[] {Integer.class,Integer.class});
        //进行远程调用
        RpcResponse response=  client.invoke(rpcRquest);
        if (null!=response){
            return Integer.parseInt(response.getResult().toString());
        }
        return null;
    }

}
  • 服务端接口实现类

public class DemoImp implements Idemo{
    @Override
    public Integer add(Integer i, Integer j) {
        return i+j;
    }
}

测试

  • 启动服务端
public static void main(String[] args) {
        ProviderServer server = new ProviderServer(8090);
        RpcRegister.buildRegist().regist(Idemo.class.getName(), new DemoImp());
        new Thread(server).start();
    }
  • 启动客户端
 public static void main(String[] args) {
        RpcInitFactory initFactory=  new RpcInitFactory("127.0.0.1",8090);
        Idemo demo = new DemoRemoteImpl(new CommonClient(initFactory));
        System.out.println(demo.add(2, 1));
    }
  • 结果:

四、总结

到这里RPC的小demo功能实现完毕,实际上的RPC框架要比这个复杂的多,真正的RPC框架要考虑性能,高可用,半包,粘包等问题,这里只是给出了一个RPC框架的实现原理,便于理解RPC框架的实现,并不能真正用于生产环境。

推荐阅读

Java锁之ReentrantLock(一)

Java锁之ReentrantLock(二)

Java锁之ReentrantReadWriteLock

JAVA NIO编程入门(一)

JAVA NIO 编程入门(二)

© 著作权归作者所有

木木匠

木木匠

粉丝 152
博文 32
码字总数 71470
作品 0
广州
高级程序员
私信 提问
JAVA NIO编程入门(一)

JAVA NIO编程入门(一) 一、前言 笔者之前接触的NIO编程比较少,所以对这一块的基础也比较弱,NIO作为java编程中一个重要的模块,不能很好的掌握它,感觉自己在java方面就掌握的不够,所以,...

木木匠
2018/09/01
160
0
少啰嗦!一分钟带你读懂Java的NIO和经典IO的区别

本文引用了“架构师社区”公众号的《史上讲的最好的Java NIO与IO的区别与应用》一文部分内容,感谢原作者的技术分享。 1、引言 很多初涉网络编程的程序员,在研究Java NIO(即异步IO)和经典...

JackJiang2011
06/25
0
0
JAVA NIO编程入门(二)

一、回顾 上一篇文章 JAVA NIO编程入门(一)我们学习了NIO编程的基础知识,并通过一个小demo实战帮助了解NIO编程的channel,buffer等概念。本文会继续学习JAVA NIO编程,并通过一个小示例来...

木木匠
2018/09/09
72
0
Java NIO系列教程(十六) Java NIO Files

原文地址 译者:章筱虎 java NIO Files类(java.nio.file.Files) 提供了操作文件的相关方法。本篇文章将会覆盖大多数常用的方法。Files类包含了很多方法,如果你需要的功能在文中没有提及,需...

章 筱虎
2018/01/23
0
0
Java NIO AsynchronousFileChannel

原文链接 , 原文作者:Jakob Jenkov, 翻译:Neil Hao 在Java 7,AsynchronousFileChannel 被添加到了Java NIO中。使用AsynchronousFileChannel可以实现异步地读取和写入文件数据。 创建一个A...

Neil_Hao
2018/01/20
0
0

没有更多内容

加载失败,请刷新页面

加载更多

 介绍一款优秀的通用管理权限快速开发框架

这是一套以权限管理为主的轻量化快速开发框架,配置有流程、专业表单、权限、app、企业微信等基础功能模块,在开发通用软件的效率上很有优势。 软件平台常用研发需求分析 《那些年我们一起做...

我想造火箭
19分钟前
6
0
ElasticDL:蚂蚁金服开源基于 TensorFlow 的弹性分布式深度学习系统

9 月 11 日,蚂蚁金服在2019谷歌开发者大会上海站上开源了 ElasticDL 项目,这是业界首个基于 TensorFlow 实现弹性深度学习的开源系统。 开源地址为:https://github.com/sql-machine-learni...

SOFAStack
26分钟前
4
0
CSS--渐变

一、什么是渐变 多种颜色平缓变化的一种显示效果 二、渐变的主要元素 色标:一种颜色,以及出现的位置,一个渐变至少两种色标 三、渐变的分类 1、线性渐变 以直线的方式来填充渐变色 2、径向...

wytao1995
41分钟前
6
0
Java通过模板生成PDF再转换为图片

1、添加maven依赖 <dependency> <groupId>com.itextpdf</groupId> <artifactId>itextpdf</artifactId> <version>5.5.13.1</version></dependency><dependency> <g......

醉美閑聖
48分钟前
4
0
SpringBoot-MVC RequestBody中LocalDateTime的自适应配置

请求的json报文中可能会出现 一下几种: ['2019-01-01','2019-01-01 12:03:34','20190101120334'] 但是接收的Request实体类日期字段是LocalDateTime类型 LocalDateTime applyDate; 希望的情况......

汉堡OSC
今天
15
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部