当前位置：首页 > 编程语言 > Java+Nettty自定义RPC框架

Java+Nettty自定义RPC框架

2021-04-12 12:27

阅读：715

标签：ISE   ali   print   rup   utf-8   trace   equal   conf   png   

本次利用Java+netty实现自定义rpc框架，共分为三个工程,公共模块+服务提供者+服务消费者：
 
技术图片
rpc-common工程
 
pom.xml
 

    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    4.0.0com.rpc.common
    rpc-common
    0.0.1-SNAPSHOTUTF-81.81.8io.netty
            netty-all
            4.1.16.Finalcom.alibaba
 
            fastjson
 
            1.2.41org.apache.commons
            commons-lang3
            3.0org.apache.maven.plugins
                maven-compiler-plugin
                3.11.81.8utf-8
 
 
RpcDecoder.java
 
 
package com.rpc.decoder;
 
import java.util.List;
 
import com.rpc.util.SerializationUtil;
 
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
/**
 *
 * @author linxu
 *
 */
public class RpcDecoder extends ByteToMessageDecoder {
    private Class> genericClass;
 
    public RpcDecoder(Class> genericClass) {
        this.genericClass = genericClass;
    }
 
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception {
        if (in.readableBytes() ) {
            return;
        }
        in.markReaderIndex();
        int dataLength = in.readInt();
        if (dataLength ) {
            ctx.close();
        }
        if (in.readableBytes()  dataLength) {
            in.resetReaderIndex();
        }
        byte[] data = new byte[dataLength];
        in.readBytes(data);
        Object obj = SerializationUtil.toClass(genericClass, data);
        out.add(obj);
    }
 
}
 
 
RpcEncoder.java
 
package com.rpc.decoder;
 
import com.rpc.util.SerializationUtil;
 
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
/**
 *
 * @author linxu
 *
 */
@SuppressWarnings("rawtypes")
public class RpcEncoder extends MessageToByteEncoder {
     private Class> genericClass; 
       
        public RpcEncoder(Class> genericClass) { 
            this.genericClass = genericClass; 
        } 
       
        @Override 
        public void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception { 
            if (genericClass.isInstance(msg)) { 
                byte[] data = SerializationUtil.toByte(msg); 
                out.writeInt(data.length); 
                out.writeBytes(data); 
            } 
        } 
     
}
 
SrpcRequest.java
 
package com.rpc.message;
 
import java.io.Serializable;
import java.util.Arrays;
/**
 *
 * @author linxu
 *
 */
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; 
   
    public String getRequestId() { 
        return requestId; 
    } 
   
    public void setRequestId(String requestId) { 
        this.requestId = requestId; 
    } 
   
    public String getInterfaceName() { 
        return interfaceName; 
    } 
   
    public void setInterfaceName(String interfaceName) { 
        this.interfaceName = interfaceName; 
    } 
   
    public String getMethodName() { 
        return methodName; 
    } 
   
    public void setMethodName(String methodName) { 
        this.methodName = methodName; 
    } 
   
    public Class>[] getParameterTypes() { 
        return parameterTypes; 
    } 
   
    public void setParameterTypes(Class>[] parameterTypes) { 
        this.parameterTypes = parameterTypes; 
    } 
   
    public Object[] getParameters() { 
        return parameters; 
    } 
   
    public void setParameters(Object[] parameters) { 
        this.parameters = parameters; 
    } 
   
    @Override 
    public String toString() { 
        return "SrpcRequest [requestId=" + requestId + ", interfaceName=" + interfaceName 
                + ", methodName=" + methodName + ", parameterTypes=" 
                + Arrays.toString(parameterTypes) + ", parameters=" + Arrays.toString(parameters) 
                + "]"; 
    } 
   
} 
 
 
 
SrpcResponse.java
 
package com.rpc.message;
 
import java.io.Serializable;
 
/**
 *
 * @author linxu
 *
 */
public class SrpcResponse implements Serializable{
    private static final long serialVersionUID = -5934073769679010930L; 
    // 请求的Id 
    private String            requestId; 
    // 异常 
    private Throwable         error; 
    // 响应 
    private Object            result; 
   
    public String getRequestId() { 
        return requestId; 
    } 
   
    public void setRequestId(String requestId) { 
        this.requestId = requestId; 
    } 
    public Throwable getError() { 
        return error; 
    } 
   
    public void setError(Throwable error) { 
        this.error = error; 
    } 
   
    public Object getResult() { 
        return result; 
    } 
   
    public void setResult(Object result) { 
        this.result = result; 
    } 
   
    @Override 
    public String toString() { 
        return "SrpcResponse [requestId=" + requestId + ", error=" + error + ", result=" + result 
                + "]"; 
    } 
   
} 
 
 
 
SerializationUtil.java
 
package com.rpc.util;
 
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
/**
 *
 * @author 86136
 *
 */
public class SerializationUtil {
    /**
     * 序列化
     *
     * @param t
     * @return
     */
    public static byte[] toByte(T t) {
        ByteArrayOutputStream b = new ByteArrayOutputStream();
        ObjectOutputStream o = null;
        try {
            o = new ObjectOutputStream(b);
            o.writeObject(t);
            return b.toByteArray();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } finally {
            if (b != null) {
                try {
                    b.close();
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                if (o != null) {
                    try {
                        o.close();
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        }
        return null;
 
    }
 
    /**
     * 反序列
     *
     * @param clazz
     * @param buffer
     * @return
     * @throws Exception
     */
    @SuppressWarnings("unchecked")
    public static  T toClass(Class clazz, byte[] buffer) throws Exception {
        ByteArrayInputStream i = new ByteArrayInputStream(buffer);
        ObjectInputStream o = null;
        try {
            o = new ObjectInputStream(i);
            return (T) o.readObject();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } finally {
            if (i != null) {
                i.close();
            }
            if (o != null) {
                o.close();
            }
        }
        return null;
 
    }
}
 
 
DeptService.java
 
package com.user.service;
 
public interface DeptService {
    public String selectDept(String d);
 
}
 
 
UserService.java
 
package com.user.service;
 
public interface UserService {
    String sayHello(String word);
}
 
 
 
rpc-consumer工程
 
ClientBootstrap.java
 
package com.rpc;
 
import com.user.service.DeptService;
import com.user.service.UserService;
 
/**
 * 调用测试
 *
 * @author linxu
 *
 */
public class ClientBootstrap {
 
    public static void main(String[] args) throws InterruptedException {
        test1();
        test2();
    }
 
    public static void test1() {
        RpcConsumer consumer = new RpcConsumer();
        UserService service = (UserService) consumer.createProxy(UserService.class);
        System.out.println(service.sayHello("are you ok 001 ?"));
    }
 
    public static void test2() {
        RpcConsumer consumer = new RpcConsumer();
        DeptService service = (DeptService) consumer.createProxy(DeptService.class);
        System.out.println(service.selectDept("are you ok 002 ?"));
    }
}
 
ClientHandler.java
 
package com.rpc;
 
import java.util.concurrent.Callable;
 
import com.rpc.message.SrpcRequest;
import com.rpc.message.SrpcResponse;
 
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
 
public class ClientHandler extends ChannelInboundHandlerAdapter implements Callable {
    private ChannelHandlerContext context;
    private SrpcResponse result;
    private SrpcRequest  para;
 
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        context = ctx;
    }
 
    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) {
        result = (SrpcResponse)msg;
        notify();
    }
 
    @Override
    public synchronized Object call() throws InterruptedException {
        context.writeAndFlush(para);
        wait();
        return result;
    }
 
    void setPara(SrpcRequest  para) {
        this.para = para;
    }
}
 
 
RpcConsumer.java
 
package com.rpc;
 
import java.lang.reflect.Proxy;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
import org.apache.commons.lang3.StringUtils;
 
import com.rpc.decoder.RpcDecoder;
import com.rpc.decoder.RpcEncoder;
import com.rpc.message.SrpcRequest;
import com.rpc.message.SrpcResponse;
 
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
 
public class RpcConsumer {
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    private static ClientHandler client;
 
    /**
     * 代理对象去执行了一个socket连接请求，
     */
    public Object createProxy(final Class> interfaceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class>[] { interfaceClass },
                (proxy, method, arguments) -> {
                    if (client == null) {
                        initClient();
                    }
                    //请求封装
                    SrpcRequest request = new SrpcRequest();
                    request.setRequestId(UUID.randomUUID().toString());
                    request.setInterfaceName(interfaceClass.getName());
                    request.setMethodName(method.getName());
                    request.setParameterTypes(method.getParameterTypes());
                    request.setParameters(arguments);
                    client.setPara(request);
                    //请求结果
                    SrpcResponse response = (SrpcResponse) executor.submit(client).get();
                     
                    if (response == null || !StringUtils.equals(request.getRequestId(), response.getRequestId())) {
                        return null;
                    }
                    if (response.getError() != null) {
                        throw response.getError();
                    }
                    return response.getResult();
                });
    }
 
    private static void initClient() {
        client = new ClientHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new RpcEncoder(SrpcRequest.class));
                        p.addLast(new RpcDecoder(SrpcResponse.class));
                        p.addLast(client);
                    }
                });
        try {
            b.connect("localhost", 8888).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
 
pom.xml
 

    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    4.0.0com.rpc.consumer
    rpc-consumer
    0.0.1-SNAPSHOTUTF-81.81.8com.rpc.common
            rpc-common
            0.0.1-SNAPSHOTorg.apache.maven.plugins
                maven-compiler-plugin
                3.11.81.8utf-8
 
 
 
rpc-provider工程
 
Bootstrap.java
 
package com.rpc.bootstrap;
 
import java.util.HashMap;
import java.util.Map;
 
import com.rpc.decoder.RpcDecoder;
import com.rpc.decoder.RpcEncoder;
import com.rpc.message.SrpcRequest;
import com.rpc.message.SrpcResponse;
 
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
 
@SuppressWarnings({ })
public class Bootstrap {
    // 服务器接口容器
    @SuppressWarnings({ })
    public static final Map serviceRegistry = new HashMap();
 
    public static void startServer(String hostName, int port) {
        try {
            NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
            ServerBootstrap bootstrap = new ServerBootstrap();
 
            bootstrap.group(eventLoopGroup).channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(new RpcDecoder(SrpcRequest.class));
                            p.addLast(new RpcEncoder(SrpcResponse.class));
                            p.addLast(new RpcServerHandler());
                        }
                    });
            bootstrap.bind(hostName, port).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
 
    public static void main(String[] args) {
        //服务注册
        registryService();
        //netty启动
        startServer("localhost", 8888);
    }
 
    public static void registryService() {
        final Map serviceInterfaceConfiguration = ServiceRegistry.registry();
        if (serviceInterfaceConfiguration != null && !serviceInterfaceConfiguration.isEmpty()) {
            serviceInterfaceConfiguration.forEach((k, v) -> {
                try {
                    @SuppressWarnings("deprecation")
                    Object object = Class.forName(v).newInstance();
                    serviceRegistry.put(k, object);
                } catch (ClassNotFoundException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } catch (InstantiationException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } catch (IllegalAccessException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
 
            });
        }
 
    }
}
 
RpcServerHandler.java
 
package com.rpc.bootstrap;
 
import java.lang.reflect.Method;
 
import com.rpc.message.SrpcRequest;
import com.rpc.message.SrpcResponse;
 
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
 
public class RpcServerHandler extends SimpleChannelInboundHandler {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, SrpcRequest msg) throws Exception {
        SrpcResponse response = new SrpcResponse();
        response.setRequestId(msg.getRequestId());
        try {
            response.setResult(handle(msg));
        } catch (Exception e) {
            response.setError(e);
            e.printStackTrace();
        }
        ctx.writeAndFlush(response);
    }
 
    /**
     * 执行服务接口方法
     *
     * @param request
     * @return
     * @throws Exception
     */
    private Object handle(SrpcRequest request) throws Exception {
        Object service = Bootstrap.serviceRegistry.get(request.getInterfaceName());
        Method method = service.getClass().getMethod(request.getMethodName(), request.getParameterTypes());
        return method.invoke(service, request.getParameters());
    }
}
 
 
ServiceRegistry.java
 
package com.rpc.bootstrap;
 
import java.util.HashMap;
import java.util.Map;
 
public class ServiceRegistry {
 
    public static MapserviceInterfaceConfiguration=new HashMap();
     
     
    public static void put(String k,String v) {
        serviceInterfaceConfiguration.put(k, v);
         
    }
    public static Map registry() {
        //服务接口注册
        put("com.user.service.UserService", "com.user.serviceimp.UserServiceImpl");
        put("com.user.service.DeptService", "com.user.serviceimp.DeptServiceImpl");
        return serviceInterfaceConfiguration;
         
    }
     
}
 
 
DeptServiceImpl.java
 
package com.user.serviceimp;
 
import com.user.service.DeptService;
 
public class DeptServiceImpl implements DeptService{
 
    @Override
    public String selectDept(String d) {
         
        return d+":successful";
    }
 
}
 
UserServiceImpl.java
 
package com.user.serviceimp;
 
import com.user.service.UserService;
 
public class UserServiceImpl implements UserService{
 
    @Override
    public String sayHello(String word) {
        // TODO Auto-generated method stub
        System.err.println("调用成功"+word);
        return "调用成功"+word;
    }
 
}
 
pom,xml
 
4.0.0com.rpc.provider
  rpc-provider
  0.0.1-SNAPSHOTUTF-81.81.8com.rpc.common
            rpc-common
            0.0.1-SNAPSHOTorg.apache.maven.plugins
                maven-compiler-plugin
                3.11.81.8utf-8

 

Java+Nettty自定义RPC框架

标签：ISE   ali   print   rup   utf-8   trace   equal   conf   png   

原文地址：https://www.cnblogs.com/mature1021/p/13354230.html

上一篇：Java游戏服（简单事件通知实现）

下一篇：Python + Flask 项目开发实践系列《二》

文章来自：搜素材网的编程语言模块，转载请注明文章出处。
文章标题：Java+Nettty自定义RPC框架
文章链接：http://soscw.com/index.php/essay/74714.html

评论

亲，登录后才可以留言！

热门文章

推荐文章

最新文章

置顶文章

关于我们 | 版权声明 | 常见问题 | 素材投稿 | 联系我们 | 网站地图 |
搜素材网素材除本站原创外均由用户分享，若发现权利被侵害，请联系及时联系我们，我们会在第一时间进行处理。
特别说明：本站所有资源除本站原创外仅供学习与参考，请勿用于商业用途,如有侵犯您的版权请联系客服服务QQ：点击这里给我发消息
Copyright © 2025 soscw.com 搜素材网素材网版权所有 蜀ICP备18015633号-1