简单RPC实现之Netty实现

所谓RPC就是远程方法调用(Remote  Process Call ),简单的来说就是通过MQ,TCP,HTTP或者自己写的网络协议来传输我要调用对方的什么接口,对方处理之后再把结果返回给我.就这么简单的一个过程。

运行时,一次客户机对服务器的RPC调用,其内部操作大致有如下十步:
1、调用客户端句柄;执行传送参数
2、调用本地系统内核发送网络消息
3、消息传送到远程主机
4、服务器句柄得到消息并取得参数
5、执行远程过程
6、执行的过程将结果返回服务器句柄
7、服务器句柄返回结果,调用远程系统内核
8、消息传回本地主机
9、客户句柄由内核接收消息
10、客户接收句柄返回的数据

之前一篇文章简单RPC之Socket实现我们通过socket通信实现了简单的RPC调用,接下来我们基于Netty来实现一个简单的RPC调用过程,当然还有很多不完善的地方,只供参考学习RPC使用。

一、首先定义消息传递的实体类

public class ClassInfo implements Serializable {

	private static final long serialVersionUID = -8970942815543515064L;

	private String className;//类名
	private String methodName;//函数名称
	private Class<?>[] types;//参数类型
	private Object[] objects;//参数列表
	public String getClassName() {
		return className;
	}
	public void setClassName(String className) {
		this.className = className;
	}
	public String getMethodName() {
		return methodName;
	}
	public void setMethodName(String methodName) {
		this.methodName = methodName;
	}
	public Class<?>[] getTypes() {
		return types;
	}
	public void setTypes(Class<?>[] types) {
		this.types = types;
	}
	public Object[] getObjects() {
		return objects;
	}
	public void setObjects(Object[] objects) {
		this.objects = objects;
	}
}

二、创建Netty操作的服务端,以及具体操作

(1)服务端

public class RPCServer {
	private int port;
	public RPCServer(int port){
		this.port = port;
	}
	public void start(){
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup workerGroup = new NioEventLoopGroup();

		try {
			ServerBootstrap serverBootstrap = new ServerBootstrap().group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
					.localAddress(port).childHandler(new ChannelInitializer<SocketChannel>() {

						@Override
						protected void initChannel(SocketChannel ch) throws Exception {
		                    ChannelPipeline pipeline = ch.pipeline();
		                     pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
		                        pipeline.addLast(new LengthFieldPrepender(4));
		                        pipeline.addLast("encoder", new ObjectEncoder());
		                        pipeline.addLast("decoder", new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
		                        pipeline.addLast(new InvokerHandler());
						}
					}).option(ChannelOption.SO_BACKLOG, 128)
	                .childOption(ChannelOption.SO_KEEPALIVE, true);
			ChannelFuture future = serverBootstrap.bind(port).sync();
	        System.out.println("Server start listen at " + port );
	        future.channel().closeFuture().sync();
		} catch (Exception e) {
			 bossGroup.shutdownGracefully();
	         workerGroup.shutdownGracefully();
		}
	}
	public static void main(String[] args) throws Exception {
        int port;
        if (args.length > 0) {
            port = Integer.parseInt(args[0]);
        } else {
            port = 8080;
        }
        new RPCServer(port).start();
    }
}

(2)服务端操作,由服务端我们看到具体的数据传输操作是进行序列化的,具体的操作还是比较简单的,就是获取发送过来的信息,这样就可以通过反射获得类名,根据函数名和参数值,执行具体的操作,将执行结果发送给客户端。

public class InvokerHandler extends ChannelInboundHandlerAdapter {
	public static ConcurrentHashMap<String, Object> classMap = new ConcurrentHashMap<String,Object>();
	@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ClassInfo classInfo = (ClassInfo)msg;
        Object claszz = null;
		if(!classMap.containsKey(classInfo.getClassName())){
			try {
				claszz = Class.forName(classInfo.getClassName()).newInstance();
				classMap.put(classInfo.getClassName(), claszz);
			} catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
				e.printStackTrace();
			}
		}else {
			claszz = classMap.get(classInfo.getClassName());
		}
		Method method = claszz.getClass().getMethod(classInfo.getMethodName(), classInfo.getTypes());
        Object result = method.invoke(claszz, classInfo.getObjects());
        ctx.write(result);
        ctx.flush();
        ctx.close();
    }
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
	     cause.printStackTrace();
	     ctx.close();
	}  

}

三、客户端,通过代理机制来触发远程调用

(1)客户端,当执行具体的函数时会调用远程操作,将具体操作的类、函数及参数信息发送到服务端

public class RPCProxy {

	@SuppressWarnings("unchecked")
	public static <T> T create(Object target){

		return (T) Proxy.newProxyInstance(target.getClass().getClassLoader(),target.getClass().getInterfaces(), new InvocationHandler(){

			@Override
			public Object invoke(Object proxy, Method method, Object[] args)
						throws Throwable {

				ClassInfo classInfo = new ClassInfo();
				classInfo.setClassName(target.getClass().getName());
				classInfo.setMethodName(method.getName());
				classInfo.setObjects(args);
				classInfo.setTypes(method.getParameterTypes());

				ResultHandler resultHandler = new ResultHandler();
		        EventLoopGroup group = new NioEventLoopGroup();
		        try {
		            Bootstrap b = new Bootstrap();
		            b.group(group)
		             .channel(NioSocketChannel.class)
		             .option(ChannelOption.TCP_NODELAY, true)
		             .handler(new ChannelInitializer<SocketChannel>() {
		                 @Override
		                 public void initChannel(SocketChannel ch) throws Exception {
		                	 ChannelPipeline pipeline = ch.pipeline();
                             pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                             pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                             pipeline.addLast("encoder", new ObjectEncoder());
                             pipeline.addLast("decoder", new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
                             pipeline.addLast("handler",resultHandler);
		                 }
		             });  

		            ChannelFuture future = b.connect("localhost", 8080).sync();
		            future.channel().writeAndFlush(classInfo).sync();
		            future.channel().closeFuture().sync();
		        } finally {
		            group.shutdownGracefully();
		        }
		        return resultHandler.getResponse();
			}
		});
	}
}

(2)获取远程调用返回的结果值

public class ResultHandler extends ChannelInboundHandlerAdapter {

	private Object response;  

    public Object getResponse() {
    return response;
}  

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        response=msg;
        System.out.println("client接收到服务器返回的消息:" + msg);
    }  

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("client exception is general");
    }
}

四、接口、实现类及Main操作

接口:

public interface HelloRpc {
	String hello(String name);
}

实现类:

public class HelloRpcImpl implements HelloRpc {

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

}

Main操作:

public class Main {
	public static void main(String [] args){
		HelloRpc helloRpc = new HelloRpcImpl();
		helloRpc = RPCProxy.create(helloRpc);
		System.err.println(helloRpc.hello("rpc"));
	}
}

完整代码地址github

时间: 2024-10-07 04:02:11

简单RPC实现之Netty实现的相关文章

PHP实现简单RPC

1.什么是rpc RPC全称为Remote Procedure Call,翻译过来为“远程过程调用”.目前,主流的平台中都支持各种远程调用技术,以满足分布式系统架构中不同的系统之间的远程通信和相互调用.远程调用的应用场景极其广泛,实现的方式也各式各样. 2.从通信协议的层面 基于HTTP协议的(例如基于文本的SOAP(XML).Rest(JSON),基于二进制Hessian(Binary)) 基于TCP协议的(通常会借助Mina.Netty等高性能网络框架) 3.从不同的开发语言和平台层面 单种

网络编程 -- RPC实现原理 -- Netty -- 迭代版本V4 -- 粘包拆包

网络编程 -- RPC实现原理 -- 目录 啦啦啦 V2--Netty -- new LengthFieldPrepender(2) : 设置数据包 2 字节的特征码 new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 2) :  65535 :数据包长度.0:分隔符偏移值.2:分隔符长度.0:数据包偏移值.2:数据包长度. Class : Server package lime.pri.limeNio.netty.netty04; import j

一个简单RPC框架是如何炼成的(V)——引入传输层

开局篇我们说了,RPC框架的四个核心内容 RPC数据的传输. RPC消息 协议 RPC服务注册 RPC消息处理    接下来处理数据传输.实际应用场景一般都是基于socket.socket代码比较多,使用起来也比较麻烦.而且具体的传输通道使用socket或者其他的方式,如更上层的http,或者android里的binder,都是可替换的,只是具体的一种实现而已.所以,这里我就偷个懒,只是引入一个很简单的Connection类,用来描述一下如何将数据传输 这一层给独立出来. 首先简单列出Conne

一个简单RPC框架是如何炼成的(VI)——引入服务注册机制

开局篇我们说了,RPC框架的四个核心内容 RPC数据的传输. RPC消息 协议 RPC服务注册 RPC消息处理 接下来处理RPC服务的注册机制.所谓注册机制,就是Server需要声明支持哪些rpc方法,然后当客户端发送调用某个声明的rpc方法之后,服务端能自动找到执行该请求的具体方法.以实际的例子为例,这是现在server端处理RPC请求的代码 def procRequest(self): # 循环读取并处理收到的客户端请求 while True: req = self.conn.recv()

网络编程 -- RPC实现原理 -- Netty -- 迭代版本V3 -- 编码解码

网络编程 -- RPC实现原理 -- 目录 啦啦啦 V2--Netty -- pipeline.addLast(io.netty.handler.codec.MessageToMessageCodec<INBOUND_IN, OUTBOUND_IN>) 覆写编码解码方法. pipeline相当于拦截器.在pipeline中添加MessageToMessageCodec接口的实现类,该接口的实现类中的encode()方法自动将发送的Object对象转换为ByteBuf,decode()方法自动将

一个简单RPC框架是如何炼成的(II)——制定RPC消息

开局篇我们说了,RPC框架的四个核心内容 RPC数据的传输. RPC消息 协议 RPC服务注册 RPC消息处理 下面,我们先看一个普通的过程调用 class Client(object): def __init__(self): self.remote = None ## # 内部是委托给远程remote对象来获取结果. def sayHello(self): if self.remote: return self.remote.sayHello() else : return None cla

简单RPC框架-基于Consul的服务注册与发现

*:first-child { margin-top: 0 !important; } body>*:last-child { margin-bottom: 0 !important; } /* BLOCKS =============================================================================*/ p, blockquote, ul, ol, dl, table, pre { margin: 15px 0; } /* HEAD

一个简单RPC框架是如何炼成的(III)——实现带参数的RPC调用

上一篇,我们制定了一个很简单的RPC消息 的格式,但是还遗留了两个问题 我们并没有实现相应的encode和decode方法,没有基于可以跨设备的字符串传输,而是直接的内存变量传递. 现在的RPC request不支持带参数的请求命令.如add(a, b), 如何在RPC消息中描述参数a,b . 我先来实现第二个问题,即带参数的RPC调用. 其实,也没什么太大不同.既然是要带参数,那只能扩展原来的Request消息了,加个parameter成员,用于表示参数,具体的格式采用字典方式,{ 'arg1

RPC、基于netty的长连接和websocket

1 RPC RPC也采用C/S的编程模式,以模块调用的简单性忽略通讯的具体细节,以便程序员不用关心C/S之间的通讯协议,集中精力对付实现过程.这就决定了 RPC生成的通讯包不可能对每种应用都有最恰当的处理办法,与Socket方法相比,传输相同的有效数据,RPC占用更多的网络带宽. RPC实在socket的基础上实现的,但是它比socket需要更多的网络和资源系统. 2 基于netty的长连接 异步.高性能 Boss线程(一个服务器端口对于一个)---接收到客户端连接---生成Channel---