学习别人的rpc框架

https://my.oschina.net/huangyong/blog/361751

https://gitee.com/huangyong/rpc

第一个链接为作者的描述,非常清晰,第二个为代码,包含以下几个功能包

自己的理解和摘取一些我认为值得记录的点

RPC:客户端发起请求,端封装请求,网络连接到某个服务端(服务治理),通过网络协议(tcp/http),序列化请求发送到服务端,服务端收到网络请求,反序列化请求本地服务,再将返回序列化,发送到客户端,客户端反序列化给本地展示结果。

http应用层协议,tcp为传输层协议,越上层的协议可能提供了越丰富的特性,越底层数据传输越快。

JAVA本身的序列化方式不管是性能还是序列化后的字节大小都不太好,所以需要根据实际情况集成其他序列化方式

服务被部署在不同的服务器节点上,需要服务发现提供注册和客户端发现功能

如何实现?通过启动类加载xml文件,启动相关组件

    public static void main(String[] args) {
        new ClassPathXmlApplicationContext("spring.xml");
    }
<!-- lang: xml -->
<beans ...>
    <context:component-scan base-package="com.xxx.rpc.sample.server"/>

    <context:property-placeholder location="classpath:config.properties"/>

    <!-- 配置服务注册组件 -->
    <bean id="serviceRegistry" class="com.xxx.rpc.registry.ServiceRegistry">
        <constructor-arg name="registryAddress" value="${registry.address}"/>
    </bean>

    <!-- 配置 RPC 服务器 -->
    <bean id="rpcServer" class="com.xxx.rpc.server.RpcServer">
        <constructor-arg name="serverAddress" value="${server.address}"/>
        <constructor-arg name="serviceRegistry" ref="serviceRegistry"/>
    </bean>
</beans>

服务注解标签

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Component // 表明可被 Spring 扫描
public @interface RpcService {

    Class<?> value();
}

具体的服务实现类

@RpcService(HelloService.class) // 指定远程接口
public class HelloServiceImpl implements HelloService {

    @Override
    public String hello(String name) {
        return "Hello! " + name;
    }
}

服务发现相关就是利用了zookeeper的znode/watcher

    String ZK_REGISTRY_PATH = "/registry";
    String ZK_DATA_PATH = ZK_REGISTRY_PATH + "/data";
public class ServiceRegistry {

    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceRegistry.class);

    private CountDownLatch latch = new CountDownLatch(1);

    private String registryAddress;

    public ServiceRegistry(String registryAddress) {
        this.registryAddress = registryAddress;
    }

    public void register(String data) {
        if (data != null) {
            ZooKeeper zk = connectServer();
            if (zk != null) {
                createNode(zk, data);
            }
        }
    }

    private ZooKeeper connectServer() {
        ZooKeeper zk = null;
        try {
            zk = new ZooKeeper(registryAddress, Constant.ZK_SESSION_TIMEOUT, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getState() == Event.KeeperState.SyncConnected) {
                        latch.countDown();
                    }
                }
            });
            latch.await();
        } catch (IOException | InterruptedException e) {
            LOGGER.error("", e);
        }
        return zk;
    }

    private void createNode(ZooKeeper zk, String data) {
        try {
            byte[] bytes = data.getBytes();
            String path = zk.create(Constant.ZK_DATA_PATH, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            LOGGER.debug("create zookeeper node ({} => {})", path, data);
        } catch (KeeperException | InterruptedException e) {
            LOGGER.error("", e);
        }
    }
}

下面是NIO rpc的实现 implements ApplicationContextAware, InitializingBean

setApplicationContext处理类处理了注解指出的服务类

afterPropertiesSet 开启了服务并向注册中心注册

public class RpcServer implements ApplicationContextAware, InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class);

    private String serverAddress;
    private ServiceRegistry serviceRegistry;

    private Map<String, Object> handlerMap = new HashMap<>(); // 存放接口名与服务对象之间的映射关系

    public RpcServer(String serverAddress) {
        this.serverAddress = serverAddress;
    }

    public RpcServer(String serverAddress, ServiceRegistry serviceRegistry) {
        this.serverAddress = serverAddress;
        this.serviceRegistry = serviceRegistry;
    }

    @Override
    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        Map<String, Object> serviceBeanMap = ctx.getBeansWithAnnotation(RpcService.class); // 获取所有带有 RpcService 注解的 Spring Bean
        if (MapUtils.isNotEmpty(serviceBeanMap)) {
            for (Object serviceBean : serviceBeanMap.values()) {
                String interfaceName = serviceBean.getClass().getAnnotation(RpcService.class).value().getName();
                handlerMap.put(interfaceName, serviceBean);
            }
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel channel) throws Exception {
                        channel.pipeline()
                            .addLast(new RpcDecoder(RpcRequest.class)) // 将 RPC 请求进行解码(为了处理请求)
                            .addLast(new RpcEncoder(RpcResponse.class)) // 将 RPC 响应进行编码(为了返回响应)
                            .addLast(new RpcHandler(handlerMap)); // 处理 RPC 请求
                    }
                })
                .option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true);

            String[] array = serverAddress.split(":");
            String host = array[0];
            int port = Integer.parseInt(array[1]);

            ChannelFuture future = bootstrap.bind(host, port).sync();
            LOGGER.debug("server started on port {}", port);

            if (serviceRegistry != null) {
                serviceRegistry.register(serverAddress); // 注册服务地址
            }

            future.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}
public class RpcEncoder extends MessageToByteEncoder {

    private Class<?> genericClass;

    public RpcEncoder(Class<?> genericClass) {
        this.genericClass = genericClass;
    }

    @Override
    public void encode(ChannelHandlerContext ctx, Object in, ByteBuf out) throws Exception {
        if (genericClass.isInstance(in)) {
            byte[] data = SerializationUtil.serialize(in);
            out.writeInt(data.length);
            out.writeBytes(data);
        }
    }
}

请求

public class RpcRequest {

    private String requestId;
    private String className;
    private String methodName;
    private Class<?>[] parameterTypes;
    private Object[] parameters;

    // getter/setter...
}

响应

public class RpcResponse {

    private String requestId;
    private Throwable error;
    private Object result;

    // getter/setter...
}

这里将序列化和反序列化方法集中在序列化工具中,可以自由替换成其他

public class SerializationUtil {

    private static Map<Class<?>, Schema<?>> cachedSchema = new ConcurrentHashMap<>();

    private static Objenesis objenesis = new ObjenesisStd(true);

    private SerializationUtil() {
    }

    @SuppressWarnings("unchecked")
    private static <T> Schema<T> getSchema(Class<T> cls) {
        Schema<T> schema = (Schema<T>) cachedSchema.get(cls);
        if (schema == null) {
            schema = RuntimeSchema.createFrom(cls);
            if (schema != null) {
                cachedSchema.put(cls, schema);
            }
        }
        return schema;
    }

    @SuppressWarnings("unchecked")
    public static <T> byte[] serialize(T obj) {
        Class<T> cls = (Class<T>) obj.getClass();
        LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
        try {
            Schema<T> schema = getSchema(cls);
            return ProtostuffIOUtil.toByteArray(obj, schema, buffer);
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        } finally {
            buffer.clear();
        }
    }

    public static <T> T deserialize(byte[] data, Class<T> cls) {
        try {
            T message = (T) objenesis.newInstance(cls);
            Schema<T> schema = getSchema(cls);
            ProtostuffIOUtil.mergeFrom(data, message, schema);
            return message;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }
}

请求处理类handler,为了避免使用 Java 反射带来的性能问题,我们可以使用 CGLib 提供的反射 API,如下面用到的FastClassFastMethod

意思都一样,就是根据请求拿出请求相关信息,然后反射调用拿到结果构造返回

public class RpcHandler extends SimpleChannelInboundHandler<RpcRequest> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcHandler.class);

    private final Map<String, Object> handlerMap;

    public RpcHandler(Map<String, Object> handlerMap) {
        this.handlerMap = handlerMap;
    }

    @Override
    public void channelRead0(final ChannelHandlerContext ctx, RpcRequest request) throws Exception {
        RpcResponse response = new RpcResponse();
        response.setRequestId(request.getRequestId());
        try {
            Object result = handle(request);
            response.setResult(result);
        } catch (Throwable t) {
            response.setError(t);
        }
        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
    }

    private Object handle(RpcRequest request) throws Throwable {
        String className = request.getClassName();
        Object serviceBean = handlerMap.get(className);

        Class<?> serviceClass = serviceBean.getClass();
        String methodName = request.getMethodName();
        Class<?>[] parameterTypes = request.getParameterTypes();
        Object[] parameters = request.getParameters();

        /*Method method = serviceClass.getMethod(methodName, parameterTypes);
        method.setAccessible(true);
        return method.invoke(serviceBean, parameters);*/

        FastClass serviceFastClass = FastClass.create(serviceClass);
        FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, parameterTypes);
        return serviceFastMethod.invoke(serviceBean, parameters);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOGGER.error("server caught exception", cause);
        ctx.close();
    }
}

客户端配置

<!-- lang: java -->
<beans ...>
    <context:property-placeholder location="classpath:config.properties"/>

    <!-- 配置服务发现组件 -->
    <bean id="serviceDiscovery" class="com.xxx.rpc.registry.ServiceDiscovery">
        <constructor-arg name="registryAddress" value="${registry.address}"/>
    </bean>

    <!-- 配置 RPC 代理 -->
    <bean id="rpcProxy" class="com.xxx.rpc.client.RpcProxy">
        <constructor-arg name="serviceDiscovery" ref="serviceDiscovery"/>
    </bean>
</beans>

服务发现

public class ServiceDiscovery {

    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceDiscovery.class);

    private CountDownLatch latch = new CountDownLatch(1);

    private volatile List<String> dataList = new ArrayList<>();

    private String registryAddress;

    public ServiceDiscovery(String registryAddress) {
        this.registryAddress = registryAddress;

        ZooKeeper zk = connectServer();
        if (zk != null) {
            watchNode(zk);
        }
    }

    public String discover() {
        String data = null;
        int size = dataList.size();
        if (size > 0) {
            if (size == 1) {
                data = dataList.get(0);
                LOGGER.debug("using only data: {}", data);
            } else {
                data = dataList.get(ThreadLocalRandom.current().nextInt(size));
                LOGGER.debug("using random data: {}", data);
            }
        }
        return data;
    }

    private ZooKeeper connectServer() {
        ZooKeeper zk = null;
        try {
            zk = new ZooKeeper(registryAddress, Constant.ZK_SESSION_TIMEOUT, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getState() == Event.KeeperState.SyncConnected) {
                        latch.countDown();
                    }
                }
            });
            latch.await();
        } catch (IOException | InterruptedException e) {
            LOGGER.error("", e);
        }
        return zk;
    }

    private void watchNode(final ZooKeeper zk) {
        try {
            List<String> nodeList = zk.getChildren(Constant.ZK_REGISTRY_PATH, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == Event.EventType.NodeChildrenChanged) {
                        watchNode(zk);
                    }
                }
            });
            List<String> dataList = new ArrayList<>();
            for (String node : nodeList) {
                byte[] bytes = zk.getData(Constant.ZK_REGISTRY_PATH + "/" + node, false, null);
                dataList.add(new String(bytes));
            }
            LOGGER.debug("node data: {}", dataList);
            this.dataList = dataList;
        } catch (KeeperException | InterruptedException e) {
            LOGGER.error("", e);
        }
    }
}

代理实现,构建服务请求信息,然后注册的服务里找一个,客户端netty将请求发送到服务端

public class RpcProxy {

    private String serverAddress;
    private ServiceDiscovery serviceDiscovery;

    public RpcProxy(String serverAddress) {
        this.serverAddress = serverAddress;
    }

    public RpcProxy(ServiceDiscovery serviceDiscovery) {
        this.serviceDiscovery = serviceDiscovery;
    }

    @SuppressWarnings("unchecked")
    public <T> T create(Class<?> interfaceClass) {
        return (T) Proxy.newProxyInstance(
            interfaceClass.getClassLoader(),
            new Class<?>[]{interfaceClass},
            new InvocationHandler() {
                @Override
                public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                    RpcRequest request = new RpcRequest(); // 创建并初始化 RPC 请求
                    request.setRequestId(UUID.randomUUID().toString());
                    request.setClassName(method.getDeclaringClass().getName());
                    request.setMethodName(method.getName());
                    request.setParameterTypes(method.getParameterTypes());
                    request.setParameters(args);

                    if (serviceDiscovery != null) {
                        serverAddress = serviceDiscovery.discover(); // 发现服务
                    }

                    String[] array = serverAddress.split(":");
                    String host = array[0];
                    int port = Integer.parseInt(array[1]);

                    RpcClient client = new RpcClient(host, port); // 初始化 RPC 客户端
                    RpcResponse response = client.send(request); // 通过 RPC 客户端发送 RPC 请求并获取 RPC 响应

                    if (response.isError()) {
                        throw response.getError();
                    } else {
                        return response.getResult();
                    }
                }
            }
        );
    }
}

客户端简单实现

public class RpcClient extends SimpleChannelInboundHandler<RpcResponse> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcClient.class);

    private String host;
    private int port;

    private RpcResponse response;

    private final Object obj = new Object();

    public RpcClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    @Override
    public void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
        this.response = response;

        synchronized (obj) {
            obj.notifyAll(); // 收到响应,唤醒线程
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        LOGGER.error("client caught exception", cause);
        ctx.close();
    }

    public RpcResponse send(RpcRequest request) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel channel) throws Exception {
                        channel.pipeline()
                            .addLast(new RpcEncoder(RpcRequest.class)) // 将 RPC 请求进行编码(为了发送请求)
                            .addLast(new RpcDecoder(RpcResponse.class)) // 将 RPC 响应进行解码(为了处理响应)
                            .addLast(RpcClient.this); // 使用 RpcClient 发送 RPC 请求
                    }
                })
                .option(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture future = bootstrap.connect(host, port).sync();
            future.channel().writeAndFlush(request).sync();

            synchronized (obj) {
                obj.wait(); // 未收到响应,使线程等待
            }

            if (response != null) {
                future.channel().closeFuture().sync();
            }
            return response;
        } finally {
            group.shutdownGracefully();
        }
    }
}

客户端发送测试

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring.xml")
public class HelloServiceTest {

    @Autowired
    private RpcProxy rpcProxy;

    @Test
    public void helloTest() {
        HelloService helloService = rpcProxy.create(HelloService.class);
        String result = helloService.hello("World");
        Assert.assertEquals("Hello! World", result);
    }
}

原文地址:https://www.cnblogs.com/it-worker365/p/9990180.html

时间: 2024-11-10 17:34:40

学习别人的rpc框架的相关文章

大数据学习——hadoop的RPC框架

项目结构 服务端代码 test-hadoop-rpc pom.xml <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="ht

一个入门rpc框架的学习

一个入门rpc框架的学习 参考 huangyong-rpc 轻量级分布式RPC框架 该程序是一个短连接的rpc实现 简介 RPC,即 Remote Procedure Call(远程过程调用),说得通俗一点就是:调用远程计算机上的服务,就像调用本地服务一样. RPC 可基于 HTTP 或 TCP 协议,Web Service 就是基于 HTTP 协议的 RPC, 它具有良好的跨平台性,但其性能却不如基于 TCP 协议的 RPC.会两方面会直接影响 RPC 的性能,一是传输方式,二是序列化. 众所

rpc框架dubbo学习入门及环境搭建(spring boot+Kotlin)

学习之前,确保有以下知识基础: Java网络编程 Socket传输数据 IO流 rpc简介及实现 rpc是remote procedure call的简写,意思为远程过程调用. rpc应用较多的情景是分布式开发,那什么是分布式开发呢? 原本我也是想自己解释的,奈何网上大佬解释得很清楚了,这里就不献丑了,建议阅读完下面推荐的几篇再继续往下 [转]分布式架构的演进(JavaWeb) 如何给老婆解释什么是RPC 如何实现一个简单的RPC 刚开始的时候,服务和调用都是在同一机器,这叫本地过程调用 之后,

RPC框架研究(二)Hadoop源代码-1

报名了阿里中间件性能大赛,我来说是一个全新的挑战.一切从空白学起,比赛的过程也是学习的过程 是的.想让自己学好.给自己报一个比赛吧~ 就像当初学围棋,也是报了围棋比赛,为了不至于输的太慘.一个星期里学了好多东西 第二天 Hadoop源代码-1 小雨 天真的以为学了Java回调机制后就能够把原来的RPC框架改为异步调用了,结果对着代码一下午都没想出要怎么去改,怎么入手. 于是决定研究一下Hadoop的源代码,看看别人是怎么实现RPC的,这也是我第一次研究源代码,曾经都是仅仅管用.无论怎样实现. 使

带你手写基于 Spring 的可插拔式 RPC 框架(一)介绍

概述 首先这篇文章是要带大家来实现一个框架,听到框架大家可能会觉得非常高大上,其实这和我们平时写业务员代码没什么区别,但是框架是要给别人使用的,所以我们要换位思考,怎么才能让别人用着舒服,怎么样才能让我们的框架性能优异.通过自己写一个框架,我们能学到的有很多,能让我们脱离 CURD,在更高的层面上去思考. 目的 写这个框架最主要的目的是要让大家了解整个框架的设计思想和用到的技术,并不是让大家关注代码,当然我实现的代码一定不是完美的,还有很多需要改进的地方,希望大家不吝赐教,一起进步. 提前准备

RPC框架实现

转载RPC框架实现 RPC(Remote Procedure Call,远程过程调用)框架是分布式服务的基石,实现RPC框架需要考虑方方面面.其对业务隐藏了底层通信过程(TCP/UDP.打包/解包.序列化/反序列化),使上层专注于功能实现:框架层面,提供各类可选架构(多进程/多线程/协程):应对设备故障(高负载/死机).网络故障(拥塞/网络分化),提供相应容灾措施. 监控是分布式服务中相当重要的一部分,其有助于我们了解业务发展状况,出灾时提供第一手分析资讯.分布式系统中的每一个模块,都需要一些指

RPC框架研究(一)Java回调机制

报名了阿里中间件性能大赛,我来说是一个全新的挑战,一切从空白学起,比赛的过程也是学习的过程 是的,想让自己学好,给自己报一个比赛吧~ 就像当初学围棋,也是报了围棋比赛,为了不至于输的太惨,一个星期里学了好多东西 第一天 Java回调机制 晴 首先还是来看看赛事介绍 比赛总共分为两道题:"RPC"与"MOM",两题都需完成.我们会对"RPC"成绩(qps)进行排名,排名前100位的队伍有资格进入MOM的比赛环节,若未进入前100名直接淘汰,最终名次

一个轻量级分布式RPC框架--NettyRpc

1.背景 最近在搜索Netty和Zookeeper方面的文章时,看到了这篇文章<轻量级分布式 RPC 框架>,作者用Zookeeper.Netty和Spring写了一个轻量级的分布式RPC框架.花了一些时间看了下他的代码,写的干净简单,写的RPC框架可以算是一个简易版的dubbo.这个RPC框架虽小,但是麻雀虽小,五脏俱全,有兴趣的可以学习一下. 项目地址:https://github.com/luxiaoxun/NettyRpc 自己花了点时间整理了下代码,并修改一些问题,以下是自己学习的一

RPC框架分析

RPC框架分析 RPC 是什么? RPC 的全称是 Remote Procedure Call 是一种进程间通信方式.它允许程序调用另一个地址空间(通常是共享网络的另一台机器上)的过程或函数,而不用程序员显式编码这个远程调用的细节.即程序员无论是调用本地的还是远程的,本质上编写的调用代码基本相同. RPC 起源 RPC 这个概念术语在上世纪 80 年代由 Bruce Jay Nelson 提出.这里我们追溯下当初开发 RPC 的原动机是什么?在 Nelson 的论文 "Implementing