Zookeeper实现简单的分布式RPC框架

   RPC(Remote Procedure Call) 在介绍分布是RPC前首先介绍一个下JAVA中简单的RPC实现

       服务器端，通过SocketServer，持续接收客户端的请求，并将客户端的请求分发到指定的处理器出去处理。

/**
 *
 * @author zhangwei_david
 * @version $Id: ServiceServer.java, v 0.1 2015年8月8日 上午11:40:41 zhangwei_david Exp $
 */
public class ServiceServer implements InitializingBean, Lifecycle, ApplicationContextAware {

    /**服务端口号**/
    private int                port       = 12000;

    private ServerSocket       server;

    //线程池
    @Autowired
    private Executor           executorService;

    public Map handlerMap = new ConcurrentHashMap<>();

    private void publishedService() throws Exception {

        server = new ServerSocket(port);

        // 一直服务
        for (;;) {
            try {
                // 获取socket
                final Socket socket = server.accept();
                executorService.execute(new Runnable() {

                    @Override
                    public void run() {
                        try {
                            // 获取input
                            ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                            ObjectOutputStream output = new ObjectOutputStream(socket
                                .getOutputStream());
                            try {
                                // 获取引用
                                String interfaceName = input.readUTF();
                                //获取 方法名
                                String methodName = input.readUTF();
                                //
                                Class[] parameterTypes = (Class[]) input.readObject();
                                Object[] arguments = (Object[]) input.readObject();
                                try {
                                    Object service = handlerMap.get(interfaceName);
                                    Method method = service.getClass().getMethod(methodName,
                                        parameterTypes);
                                    Object result = method.invoke(service, arguments);
                                    output.writeObject(result);
                                } catch (Throwable t) {
                                    output.writeObject(t);
                                } finally {
                                    input.close();
                                }
                            } finally {
                                socket.close();
                            }

                        } catch (Exception e) {

                        }
                    }
                });
            } catch (Exception e) {

            }
        }

    }

    public void init() {

    }

    /**
     * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        //发布服务
        publishedService();
    }

    /**
     * @see org.springframework.context.Lifecycle#start()
     */
    @Override
    public void start() {
    }

    /**
     * @see org.springframework.context.Lifecycle#stop()
     */
    @Override
    public void stop() {
        if (server != null) {
            try {
                server.close();

            } catch (IOException e) {

            }
        }
    }

    /**
     * @see org.springframework.context.Lifecycle#isRunning()
     */
    @Override
    public boolean isRunning() {
        return false;
    }

    /**
     * @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext)
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map serviceBeanMap = applicationContext.getBeansWithAnnotation(SRPC.class);
        System.out.println(serviceBeanMap);
        if (serviceBeanMap != null && !serviceBeanMap.isEmpty()) {
            for (Object serviceBean : serviceBeanMap.values()) {
                String interfaceName = serviceBean.getClass().getAnnotation(SRPC.class).interf()
                        .getName();
                handlerMap.put(interfaceName, serviceBean);
            }
        }
    }

    /**
     * Setter method for property executorService.
     *
     * @param executorService value to be assigned to property executorService
     */
    public void setExecutorService(Executor executorService) {
        this.executorService = executorService;
    }

}

/**
 *
 * @author zhangwei_david
 * @version $Id: SRPC.java, v 0.1 2015年8月8日 下午12:51:17 zhangwei_david Exp $
 */
@Documented
@Target({ ElementType.TYPE })
@Retention(RetentionPolicy.RUNTIME)
@Component
public @interface SRPC {

    public Class interf();
}

 至此就实现了服务的自动发现自动注册，当然这个仅针对单机环境下的自动注册。

/**
 *
 * @author zhangwei_david
 * @version $Id: Client.java, v 0.1 2015年8月8日 下午12:28:44 zhangwei_david Exp $
 */
public class Client {
    /**
     * 引用服务
     *
     * @param  接口泛型
     * @param interfaceClass 接口类型
     * @param host 服务器主机名
     * @param port 服务器端口
     * @return 远程服务
     * @throws Exception
     */
    @SuppressWarnings("unchecked")
    public static  T refer(final Class interfaceClass, final String host, final int port)
                                                                                               throws Exception {
        if (interfaceClass == null || !interfaceClass.isInterface()) {
            throw new IllegalArgumentException("必须指定服务接口");
        }

        if (host == null || host.length() == 0) {
            throw new IllegalArgumentException("必须指定服务器的地址和端口号");
        }

        return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),
            new Class[] { interfaceClass }, new InvocationHandler() {
                @Override
                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(interfaceClass.getName());
                            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();
                    }
                }
            });
    }
}

    上面在没有使用第三方依赖包实现了简单的RPC，优化增加 request和reponse,定义RPC协议。

/**
 *
 * @author zhangwei_david
 * @version $Id: SrpcRequest.java, v 0.1 2015年8月8日 下午1:45:53 zhangwei_david Exp $
 */
public class SrpcRequest implements Serializable {

    /**  */
    private static final long serialVersionUID = 6132853628325824727L;
    // 请求Id
    private String            requestId;
    // 远程调用接口名称
    private String            interfaceName;
    //远程调用方法名称
    private String            methodName;
    // 参数类型
    private Class[]        parameterTypes;
    // 参数值
    private Object[]          parameters;

    /**
     * Getter method for property requestId.
     *
     * @return property value of requestId
     */
    public String getRequestId() {
        return requestId;
    }

    /**
     * Setter method for property requestId.
     *
     * @param requestId value to be assigned to property requestId
     */
    public void setRequestId(String requestId) {
        this.requestId = requestId;
    }

    /**
     * Getter method for property interfaceName.
     *
     * @return property value of interfaceName
     */
    public String getInterfaceName() {
        return interfaceName;
    }

    /**
     * Setter method for property interfaceName.
     *
     * @param interfaceName value to be assigned to property interfaceName
     */
    public void setInterfaceName(String interfaceName) {
        this.interfaceName = interfaceName;
    }

    /**
     * Getter method for property methodName.
     *
     * @return property value of methodName
     */
    public String getMethodName() {
        return methodName;
    }

    /**
     * Setter method for property methodName.
     *
     * @param methodName value to be assigned to property methodName
     */
    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }

    /**
     * Getter method for property parameterTypes.
     *
     * @return property value of parameterTypes
     */
    public Class[] getParameterTypes() {
        return parameterTypes;
    }

    /**
     * Setter method for property parameterTypes.
     *
     * @param parameterTypes value to be assigned to property parameterTypes
     */
    public void setParameterTypes(Class[] parameterTypes) {
        this.parameterTypes = parameterTypes;
    }

    /**
     * Getter method for property parameters.
     *
     * @return property value of parameters
     */
    public Object[] getParameters() {
        return parameters;
    }

    /**
     * Setter method for property parameters.
     *
     * @param parameters value to be assigned to property parameters
     */
    public void setParameters(Object[] parameters) {
        this.parameters = parameters;
    }

}

/**
 *
 * @author zhangwei_david
 * @version $Id: SrpcResponse.java, v 0.1 2015年8月8日 下午1:47:46 zhangwei_david Exp $
 */
public class SrpcResponse implements Serializable {
    /**  */
    private static final long serialVersionUID = -5934073769679010930L;
    // 请求的Id
    private String            requestId;
    // 异常
    private Throwable         error;
    // 响应
    private Object            result;

    /**
     * Getter method for property requestId.
     *
     * @return property value of requestId
     */
    public String getRequestId() {
        return requestId;
    }

    /**
     * Setter method for property requestId.
     *
     * @param requestId value to be assigned to property requestId
     */
    public void setRequestId(String requestId) {
        this.requestId = requestId;
    }

    /**
     * Getter method for property error.
     *
     * @return property value of error
     */
    public Throwable getError() {
        return error;
    }

    /**
     * Setter method for property error.
     *
     * @param error value to be assigned to property error
     */
    public void setError(Throwable error) {
        this.error = error;
    }

    /**
     * Getter method for property result.
     *
     * @return property value of result
     */
    public Object getResult() {
        return result;
    }

    /**
     * Setter method for property result.
     *
     * @param result value to be assigned to property result
     */
    public void setResult(Object result) {
        this.result = result;
    }

}

/**
 *
 * @author zhangwei_david
 * @version $Id: ServiceServer.java, v 0.1 2015年8月8日 上午11:40:41 zhangwei_david Exp $
 */
public class ServiceServer implements InitializingBean, Lifecycle, ApplicationContextAware {

    /**服务端口号**/
    private int                port       = 12000;

    private ServerSocket       server;

    //线程池
    @Autowired
    private Executor           executorService;

    public Map handlerMap = new ConcurrentHashMap<>();

    private void publishedService() throws Exception {

        server = new ServerSocket(port);

        // 一直服务
        for (;;) {
            try {
                // 获取socket
                final Socket socket = server.accept();
                executorService.execute(new Runnable() {

                    @Override
                    public void run() {
                        try {
                            // 获取input
                            ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                            try {
                                // 获取RPC请求
                                SrpcRequest request = (SrpcRequest) input.readObject();
                                ObjectOutputStream output = new ObjectOutputStream(socket
                                    .getOutputStream());
                                try {
                                    SrpcResponse response = doHandle(request);
                                    output.writeObject(response);
                                } finally {
                                    input.close();
                                }
                            } finally {
                                socket.close();
                            }

                        } catch (Exception e) {

                        }
                    }
                });
            } catch (Exception e) {

            }
        }

    }

    private SrpcResponse doHandle(SrpcRequest request) {
        SrpcResponse response = new SrpcResponse();
        response.setRequestId(request.getRequestId());
        try {
            Object service = handlerMap.get(request.getInterfaceName());
            Method method = service.getClass().getMethod(request.getMethodName(),
                request.getParameterTypes());
            response.setResult(method.invoke(service, request.getParameters()));

        } catch (Exception e) {
            response.setError(e);
        }
        return response;
    }

    public void init() {

    }

    /**
     * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        //发布
        publishedService();
    }

    /**
     * @see org.springframework.context.Lifecycle#start()
     */
    @Override
    public void start() {
    }

    /**
     * @see org.springframework.context.Lifecycle#stop()
     */
    @Override
    public void stop() {
        if (server != null) {
            try {
                server.close();

            } catch (IOException e) {

            }
        }
    }

    /**
     * @see org.springframework.context.Lifecycle#isRunning()
     */
    @Override
    public boolean isRunning() {
        return false;
    }

    /**
     * @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext)
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map serviceBeanMap = applicationContext.getBeansWithAnnotation(SRPC.class);
        System.out.println(serviceBeanMap);
        if (serviceBeanMap != null && !serviceBeanMap.isEmpty()) {
            for (Object serviceBean : serviceBeanMap.values()) {
                String interfaceName = serviceBean.getClass().getAnnotation(SRPC.class).interf()
                        .getName();
                handlerMap.put(interfaceName, serviceBean);
            }
        }
    }

    /**
     * Setter method for property executorService.
     *
     * @param executorService value to be assigned to property executorService
     */
    public void setExecutorService(Executor executorService) {
        this.executorService = executorService;
    }

}

/**
 *
 * @author zhangwei_david
 * @version $Id: Client.java, v 0.1 2015年8月8日 下午12:28:44 zhangwei_david Exp $
 */
public class Client {
    /**
     * 引用服务
     *
     * @param  接口泛型
     * @param interfaceClass 接口类型
     * @param host 服务器主机名
     * @param port 服务器端口
     * @return 远程服务
     * @throws Exception
     */
    @SuppressWarnings("unchecked")
    public static  T refer(final Class interfaceClass, final String host, final int port)
            throws Exception {
        if (interfaceClass == null || !interfaceClass.isInterface()) {
            throw new IllegalArgumentException("必须指定服务接口");
        }

        if (host == null || host.length() == 0) {
            throw new IllegalArgumentException("必须指定服务器的地址和端口号");
        }

        return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),
            new Class[] { interfaceClass }, new InvocationHandler() {
            @Override
            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 {
                        SrpcRequest request = new SrpcRequest();
                        request.setRequestId(UUID.randomUUID().toString());
                        request.setInterfaceName(interfaceClass.getName());
                        request.setMethodName(method.getName());
                        request.setParameterTypes(method.getParameterTypes());
                        request.setParameters(arguments);
                        output.writeObject(request);
                        ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                        try {
                            SrpcResponse response = (SrpcResponse) input.readObject();
                            if (response.getError() != null
                                    && response.getError() instanceof Throwable) {
                                throw response.getError();
                            }
                            return response.getResult();
                        } finally {
                            input.close();
                        }
                    } finally {
                        output.close();
                    }
                } finally {
                    socket.close();
                }
            }
        });
    }
}

     后续继续优化序列化和NIO优化

/**
 *
 * @author zhangwei_david
 * @version $Id: ServiceServer.java, v 0.1 2015年8月8日 上午11:40:41 zhangwei_david Exp $
 */
public class ServiceServer implements InitializingBean, Lifecycle, ApplicationContextAware {

    /**服务端口号**/
    private int                 port       = 12000;

    private Selector            selector;

    private ServerSocketChannel serverSocketChannel;

    public Map  handlerMap = new ConcurrentHashMap<>();

    private void publishedService() throws Exception {

        // 一直服务
        for (;;) {
            try {
                //超时1s
                selector.select(1000);
                Set selectionKeys = selector.selectedKeys();
                Iterator it = selectionKeys.iterator();
                SelectionKey key = null;
                while (it.hasNext()) {
                    key = it.next();
                    it.remove();
                    try {
                        handleInput(key);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }

    private void handleInput(SelectionKey key) throws IOException {
        if (key.isValid()) {
            if (key.isAcceptable()) {
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                sc.register(selector, SelectionKey.OP_READ);
            }
            if (key.isReadable()) {
                SocketChannel sc = (SocketChannel) key.channel();
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);

                int readBytes = sc.read(readBuffer);
                if (readBytes > 0) {
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    SrpcRequest request = SerializationUtil.deserializer(bytes, SrpcRequest.class);
                    SrpcResponse response = doHandle(request);
                    doWriteResponse(sc, response);
                } else if (readBytes < 0) {
                    key.cancel();
                    sc.close();
                }
            }
        }
    }

    private void doWriteResponse(SocketChannel channel, SrpcResponse response) throws IOException {
        if (response == null) {
            return;
        }
        byte[] bytes = SerializationUtil.serializer(response);
        ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
        writeBuffer.put(bytes);
        writeBuffer.flip();
        channel.write(writeBuffer);
    }

    /**
     *
     * @throws IOException
     * @throws ClosedChannelException
     */
    private void init() throws IOException, ClosedChannelException {
        // 打开socketChannel
        serverSocketChannel = ServerSocketChannel.open();
        //设置非阻塞模式
        serverSocketChannel.configureBlocking(false);
        // 绑定端口
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        // 创建selector
        selector = Selector.open();

        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    }

    private SrpcResponse doHandle(SrpcRequest request) {
        SrpcResponse response = new SrpcResponse();
        if (StringUtils.isBlank(request.getRequestId())) {
            response.setError(new IllegalArgumentException("request id must be not null"));
        }
        response.setRequestId(request.getRequestId());
        try {
            Object service = handlerMap.get(request.getInterfaceName());
            Method method = service.getClass().getMethod(request.getMethodName(),
                request.getParameterTypes());
            response.setResult(method.invoke(service, request.getParameters()));

        } catch (Exception e) {
            response.setError(e);
        }
        return response;
    }

    /**
     * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        init();
        //发布
        publishedService();
    }

    /**
     * @see org.springframework.context.Lifecycle#start()
     */
    @Override
    public void start() {
    }

    /**
     * @see org.springframework.context.Lifecycle#stop()
     */
    @Override
    public void stop() {

    }

    /**
     * @see org.springframework.context.Lifecycle#isRunning()
     */
    @Override
    public boolean isRunning() {
        return false;
    }

    /**
     * @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext)
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map serviceBeanMap = applicationContext.getBeansWithAnnotation(SRPC.class);
        System.out.println(serviceBeanMap);
        if (serviceBeanMap != null && !serviceBeanMap.isEmpty()) {
            for (Object serviceBean : serviceBeanMap.values()) {
                String interfaceName = serviceBean.getClass().getAnnotation(SRPC.class).interf()
                    .getName();
                handlerMap.put(interfaceName, serviceBean);
            }
        }
    }

}

/**
 *
 * @author zhangwei_david
 * @version $Id: Client.java, v 0.1 2015年8月8日 下午12:28:44 zhangwei_david Exp $
 */
public class Client {
    /**
     * 引用服务
     *
     * @param  接口泛型
     * @param interfaceClass 接口类型
     * @param host 服务器主机名
     * @param port 服务器端口
     * @return 远程服务
     * @throws Exception
     */
    @SuppressWarnings("unchecked")
    public static  T refer(final Class interfaceClass, final String host, final int port)
                                                                                               throws Exception {
        if (interfaceClass == null || !interfaceClass.isInterface()) {
            throw new IllegalArgumentException("必须指定服务接口");
        }

        if (host == null || host.length() == 0) {
            throw new IllegalArgumentException("必须指定服务器的地址和端口号");
        }

        return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),
            new Class[] { interfaceClass }, new InvocationHandler() {
                @Override
                public Object invoke(Object proxy, Method method, Object[] arguments)
                                                                                     throws Throwable {
                    //创建请求
                    SrpcRequest request = new SrpcRequest();
                    request.setRequestId(UUID.randomUUID().toString());
                    request.setInterfaceName(interfaceClass.getName());
                    request.setMethodName(method.getName());
                    request.setParameterTypes(method.getParameterTypes());
                    request.setParameters(arguments);
                    SrpcResponse response = sendReqeust(request, host, port);
                    if (response == null
                        || !StringUtils.equals(request.getRequestId(), response.getRequestId())) {
                        return null;
                    }
                    if (response.getError() != null) {
                        throw response.getError();
                    }
                    return response.getResult();

                }
            });

    }

    public static SrpcResponse sendReqeust(SrpcRequest request, String host, int port)
                                                                                      throws IOException {
        SocketChannel socketChannel = connect(host, port);
        byte[] requestBytes = SerializationUtil.serializer(request);
        ByteBuffer writeBuffer = ByteBuffer.allocate(requestBytes.length);
        writeBuffer.put(requestBytes);
        writeBuffer.flip();
        socketChannel.write(writeBuffer);
        return readResoponse(socketChannel);
    }

    /**
     *
     * @return
     * @throws IOException
     */
    private static SrpcResponse readResoponse(SocketChannel socketChannel) throws IOException {
        try {
            ByteBuffer readBuffer = ByteBuffer.allocate(1024);
            while (socketChannel.read(readBuffer) != -1) {
                readBuffer.flip();
                byte[] bytes = new byte[readBuffer.remaining()];
                readBuffer.get(bytes);
                return SerializationUtil.deserializer(bytes, SrpcResponse.class);
            }
            return null;
        } finally {
            socketChannel.close();
        }

    }

    private static SocketChannel connect(String host, int port) throws IOException {//连接到CSDN
        InetSocketAddress socketAddress = new InetSocketAddress(host, port);
        return SocketChannel.open(socketAddress);

    }
}

     在分布式系统中，为了提供系统的可用性和稳定性一般都会将服务部署在多台服务器上，为了实现自动注册自动发现远程服务，通过ZK,和ProtocolBuffe 以及Netty实现一个简单的分布式RPC框架。

   首先简单介绍一下Zookeeper和ProtocalBuffer

   Zookeeper 是由Apache Handoop的子项目发展而来。是知名的互联网公司Yahoo创建的。Zookeeper为分布式应用提供了高效且可靠的分布式协调服务。

  ProtocolBuffer是用于结构化数据串行化的灵活、高效、自动的方法，有如XML，不过它更小、更快、也更简单。你可以定义自己的数据结构，然后使用代码生成器生成的代码来读写这个数据结构。你甚至可以在无需重新部署程序的情况下更新数据结构。

   RPC 就是Remote Procedure Call Protocol 远程过程调用协议。

    JAVA对象要能够在网络上传输都必须序列化，使用高效的序列化框架ProtocolBuffer实现序列化。

/**
 * 序列化工具
 * @author zhangwei_david
 * @version $Id: SerializationUtil.java, v 0.1 2014年12月31日 下午5:41:35 zhangwei_david Exp $
 */
public class SerializationUtil {

    private static Map cachedSchema = new ConcurrentHashMap();

    private static Objenesis                objenesis    = new ObjenesisStd(true);

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

    /**
     * 序列化
     *
     * @param obj
     * @return
     */
    public static  byte[] serializer(T obj) {
        @SuppressWarnings("unchecked")
        Class clazz = (Class) obj.getClass();
        LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
        try {
            Schema schema = getSchema(clazz);
            return ProtostuffIOUtil.toByteArray(obj, schema, buffer);
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        } finally {
            buffer.clear();
        }
    }

    /**
     * 反序列化
     *
     * @param data
     * @param clazz
     * @return
     */
    public static  T deserializer(byte[] data, Class clazz) {
        try {
            T obj = objenesis.newInstance(clazz);
            Schema schema = getSchema(clazz);
            ProtostuffIOUtil.mergeFrom(data, obj, schema);
            return obj;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }
}

远程调用的请求对象

/**
 *Rpc 请求的主体
 * @author zhangwei_david
 * @version $Id: SrRequest.java, v 0.1 2014年12月31日 下午6:06:25 zhangwei_david Exp $
 */
public class RpcRequest {
    // 请求Id
    private String     requestId;
    // 远程调用类名称
    private String     className;
    //远程调用方法名称
    private String     methodName;
    // 参数类型
    private Class[] parameterTypes;
    // 参数值
    private Object[]   parameters;

    /**
     * Getter method for property requestId.
     *
     * @return property value of requestId
     */
    public String getRequestId() {
        return requestId;
    }

    /**
     * Setter method for property requestId.
     *
     * @param requestId value to be assigned to property requestId
     */
    public void setRequestId(String requestId) {
        this.requestId = requestId;
    }

    /**
     * Getter method for property className.
     *
     * @return property value of className
     */
    public String getClassName() {
        return className;
    }

    /**
     * Setter method for property className.
     *
     * @param className value to be assigned to property className
     */
    public void setClassName(String className) {
        this.className = className;
    }

    /**
     * Getter method for property methodName.
     *
     * @return property value of methodName
     */
    public String getMethodName() {
        return methodName;
    }

    /**
     * Setter method for property methodName.
     *
     * @param methodName value to be assigned to property methodName
     */
    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }

    /**
     * Getter method for property parameterTypes.
     *
     * @return property value of parameterTypes
     */
    public Class[] getParameterTypes() {
        return parameterTypes;
    }

    /**
     * Setter method for property parameterTypes.
     *
     * @param parameterTypes value to be assigned to property parameterTypes
     */
    public void setParameterTypes(Class[] parameterTypes) {
        this.parameterTypes = parameterTypes;
    }

    /**
     * Getter method for property parameters.
     *
     * @return property value of parameters
     */
    public Object[] getParameters() {
        return parameters;
    }

    /**
     * Setter method for property parameters.
     *
     * @param parameters value to be assigned to property parameters
     */
    public void setParameters(Object[] parameters) {
        this.parameters = parameters;
    }

    /**
     * @see java.lang.Object#toString()
     */
    @Override
    public String toString() {
        return "RpcRequest [requestId=" + requestId + ", className=" + className + ", methodName="
                + methodName + ", parameterTypes=" + Arrays.toString(parameterTypes)
                + ", parameters=" + Arrays.toString(parameters) + "]";
    }

}

 远程调用的响应对象

/**
 *Rpc 响应的主体
 * @author zhangwei_david
 * @version $Id: SrResponse.java, v 0.1 2014年12月31日 下午6:07:27 zhangwei_david Exp $
 */
public class RpcResponse {
    // 请求的Id
    private String    requestId;
    // 异常
    private Throwable error;
    // 响应
    private Object    result;

    /**
     * Getter method for property requestId.
     *
     * @return property value of requestId
     */
    public String getRequestId() {
        return requestId;
    }

    /**
     * Setter method for property requestId.
     *
     * @param requestId value to be assigned to property requestId
     */
    public void setRequestId(String requestId) {
        this.requestId = requestId;
    }

    /**
     * Getter method for property error.
     *
     * @return property value of error
     */
    public Throwable getError() {
        return error;
    }

    /**
     * Setter method for property error.
     *
     * @param error value to be assigned to property error
     */
    public void setError(Throwable error) {
        this.error = error;
    }

    /**
     * Getter method for property result.
     *
     * @return property value of result
     */
    public Object getResult() {
        return result;
    }

    /**
     * Setter method for property result.
     *
     * @param result value to be assigned to property result
     */
    public void setResult(Object result) {
        this.result = result;
    }

    /**
     *如果有异常则表示失败
     * @return
     */
    public boolean isError() {
        return error != null;
    }

    /**
     * @see java.lang.Object#toString()
     */
    @Override
    public String toString() {
        return "RpcResponse [requestId=" + requestId + ", error=" + error + ", result=" + result
                + "]";
    }

}

 RPC编码与解码

/**
 *RPC 解码
 * @author zhangwei_david
 * @version $Id: RpcDecoder.java, v 0.1 2014年12月31日 下午8:53:16 zhangwei_david Exp $
 */
public class RpcDecoder extends ByteToMessageDecoder {

    private Class genericClass;

    public RpcDecoder(Class genericClass) {
        this.genericClass = genericClass;
    }

    @Override
    public final void decode(ChannelHandlerContext ctx, ByteBuf in, List
					
						 out)
                                                                                     throws Exception {
        if (in.readableBytes() < 4) {
            return;
        }
        in.markReaderIndex();
        int dataLength = in.readInt();
        if (dataLength < 0) {
            ctx.close();
        }
        if (in.readableBytes() < dataLength) {
            in.resetReaderIndex();
        }
        byte[] data = new byte[dataLength];
        in.readBytes(data);

        Object obj = SerializationUtil.deserializer(data, genericClass);
        out.add(obj);
    }
}
 (352.7 KB)
下载次数: 65
 (1.6 KB)
下载次数: 31