原创

rpc服务实现原理的简单示例

平时开发中我们在做分布式项目时经常会用到rpc框架,实现微服务架构。现在公司用到比较多的类似dubbo等框架,这些开源框架是集成了很多功能较复杂的框架。此篇文章实现一个最简单的rpc服务示例,来简单说明rpc实现的最基本思想。

源码已放到github: https://github.com/readreamer23/RpcFrameDemo

一、 核心框架类
封装暴露服务和引用服务方法,供服务提供者和消费者调用
类名: RpcFrame.java


package com.rpc;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.ServerSocket;
import java.net.Socket;

/**
 * rpc简单框架实现
 * 定义provider暴露服务方法和consumer引用服务方法
 *
 */
public class RpcFrame {

    /**
     * 暴露服务--服务提供者用
     *
     * @param service 服务实现
     * @param port    服务端口
     * @throws Exception
     */
    public static void export(final Object service, int port) throws Exception {
        if (service == null)
            throw new IllegalArgumentException("service instance == null");
        if (port <= 0 || port > 65535)
            throw new IllegalArgumentException("无效端口:" + port);
        System.out.println("provider服务暴露service= " + service.getClass().getName() + ", 端口 " + port);
        ServerSocket server = new ServerSocket(port);
        for (; ; ) {
            try {
                final Socket socket = server.accept();
                new Thread(new Runnable() {

                    //@Override
                    public void run() {
                        try {
                            try {
                                ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                                try {
                                    String methodName = input.readUTF();
                                    Class<?>[] parameterTypes = (Class<?>[]) input.readObject();
                                    Object[] arguments = (Object[]) input.readObject();
                                    ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
                                    try {
                                        Method method = service.getClass().getMethod(methodName, parameterTypes);
                                        Object result = method.invoke(service, arguments);
                                        output.writeObject(result);
                                    } catch (Throwable t) {
                                        output.writeObject(t);
                                    } finally {
                                        output.close();
                                    }
                                } finally {
                                    input.close();
                                }
                            } finally {
                                socket.close();
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }



    /**
     * 引用服务--消费者用
     *
     * @param <T> 接口泛型
     * @param interfaceClass 接口类型
     * @param host 服务器主机名
     * @param port 服务器端口
     * @return 远程服务
     * @throws Exception
     */
    @SuppressWarnings("unchecked")
    public static <T> T refer(final Class<T> interfaceClass, final String host, final int port) throws Exception {
        if (interfaceClass == null)
            throw new IllegalArgumentException("Interface class == null");
        if (! interfaceClass.isInterface())
            throw new IllegalArgumentException("The " + interfaceClass.getName() + " 必须是接口类!");
        if (host == null || host.length() == 0)
            throw new IllegalArgumentException("Host == null!");
        if (port <= 0 || port > 65535)
            throw new IllegalArgumentException("Invalid port " + port);
        System.out.println("Get remote service " + interfaceClass.getName() + " from server " + host + ":" + port);
        return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class<?>[] {interfaceClass}, new InvocationHandler() {
            public Object invoke(Object proxy, Method method, Object[] arguments) throws Throwable {
                Socket socket = new Socket(host, port);
                try {
                    ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
                    try {
                        output.writeUTF(method.getName());
                        output.writeObject(method.getParameterTypes());
                        output.writeObject(arguments);
                        ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                        try {
                            Object result = input.readObject();
                            if (result instanceof Throwable) {
                                throw (Throwable) result;
                            }
                            return result;
                        } finally {
                            input.close();
                        }
                    } finally {
                        output.close();
                    }
                } finally {
                    socket.close();
                }
            }
        });
    }


}

二、 定义服务接口
定义一层接口,对外暴露服务用接口,而不是直接暴露实现类
类名: TestApiService.java

package com.rpc.service;

/**
 * 测试业务服务接口.
 */
public interface TestApiService {

    /**
     * 测试方法
     * @param name
     * @return
     */
    String getName(String name);
}

三、 实现服务--接口实现类
类名: TestApiServiceImpl.java

package com.rpc.service.impl;

import com.rpc.service.TestApiService;

/**
 * 测试业务服务接口实现
 */
public class TestApiServiceImpl implements TestApiService {

    public String getName(String name) {

        return "name-" + name;
    }

}

四、 暴露服务---服务提供者类
类名: Provider.java

package com.rpc;

import com.rpc.service.TestApiService;
import com.rpc.service.impl.TestApiServiceImpl;

/**
 * 服务提供者--暴露服务
 */
public class Provider {


    public static void main(String[] args) throws Exception {
        System.out.println("......服务提供者provider:");
        TestApiService service = new TestApiServiceImpl();
        RpcFrame.export(service, ConfigUtil.PORT);
        System.out.println("provider服务已经启动! ");
    }

}

五、 引用服务---消费者类
类名: Consumer.java

package com.rpc;
import com.rpc.service.TestApiService;
/**
 * 消费者--引用服务
 */
public class Consumer {
    public static void main(String[] args) throws Exception {
        System.out.println("......消费者" );
        TestApiService service = RpcFrame.refer(TestApiService.class,ConfigUtil.HOST,ConfigUtil.PORT);
        System.out.println("consumer开始消费!" );
        for (int i = 0; i < 10; i ++) {
            String hello = service.getName("第" + i+"个");
            System.out.println(hello);
        }
    }
}

六、 运行测试
1.先运行服务提供者 Provider.java
运行结果打印日志截图如下

2.再运行消费者 Consumer.java
运行结果打印日志截图如下:

源码链接: https://github.com/readreamer23/RpcFrameDemo

这个例子是使用阻塞的socket IO流来进行server和client的通信,也就是rpc应用中服务提供方和服务消费方。并且是端对端的,用端口号来直接进行通信。方法的远程调用使用的是jdk的动态代理,参数的序列化也是使用的最简单的objectStream。真实的rpc框架会对上面的实现方式进行替换,采用更快更稳定,更高可用易扩展,更适宜分布式场景的中间件,技术来替换,而且会用zookeeper等组件做服务注册与发现。还得有相应的监控系统等辅助平台应用于企业中。

正文到此结束
Loading...