网络编程 -- RPC实现原理 -- 目录
啦啦啦
V3——RPC -- 远程方法调用 及 null的传输 + Spring
服务提供商：
1. 配置 rpc03_server.xml 注入 服务提供商 rpcServiceProvider并指定初始化方法 及 服务实例 IUserService
2. 读取 服务消费者 请求的 MethodStaics ，通过反射获取服务端实例方法的返回值。返回值为null值，则映射为NullWritable实例返回。不为null，则不加以约束。
服务代理商：
1. 配置 rpc03_client.xml 注入 服务代理商 RPCObjectProxy并指定 目标对象 RPCClient 及 代理的接口 lime.pri.limeNio.netty.rpc03.service.IUserService
2. List<User> users = userService.queryAll(10, 4); ： 调用 目标对象的 Object invokeMethod(MethodStaics methodStaics); 方法，通过TCP/IP将MethodStaics实例发送至服务提供商。
3. 读取 服务提供商返回值。返回值为NullWritable实例，则映射为null值。其他实例，则不加以约束。
服务提供商：
XML ： rpc03_server.xml
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"xmlns:p="http://www.springframework.org/schema/p"xsi:schemaLocation=" http://www.springframework.org/schema/beans classpath:/org/springframework/beans/factory/xml/spring-beans-4.1.xsd http://www.springframework.org/schema/context classpath:/org/springframework/context/config/spring-context-4.1.xsd http://mybatis.org/schema/mybatis-spring http://mybatis.org/schema/mybatis-spring.xsd http://www.springframework.org/schema/aop classpath:org/springframework/aop/config/spring-aop-4.1.xsd" default-lazy-init="false"><bean id="rpcServiceProvider" class="lime.pri.limeNio.netty.rpc03.core.server.RPCServiceProvider"init-method="start" destroy-method="close"><constructor-arg index="0" value="9999" /></bean><bean id="IUserService" class="lime.pri.limeNio.netty.rpc03.service.impl.UserService" /></beans>
Class ： RPCServiceProvider 实现ApplicationContextAware 获取通过容器的getBean()方法获取 服务实例
package lime.pri.limeNio.netty.rpc03.core.server;import java.lang.reflect.Method; import java.util.List;import org.springframework.beans.BeansException; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware;import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature;import io.netty.bootstrap.ServerBootstrap; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; 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.NioServerSocketChannel; import io.netty.handler.codec.LengthFieldBasedFrameDecoder; import io.netty.handler.codec.LengthFieldPrepender; import io.netty.handler.codec.MessageToMessageCodec; import io.netty.util.CharsetUtil; import lime.pri.limeNio.netty.rpc03.core.assist.MethodStaics; import lime.pri.limeNio.netty.rpc03.core.assist.NullWritable;public class RPCServiceProvider implements ApplicationContextAware {private ServerBootstrap serverBootstrap;private EventLoopGroup boss;private EventLoopGroup worker;private int port;private ApplicationContext act;public RPCServiceProvider() {super();}public RPCServiceProvider(int port) {this.serverBootstrap = new ServerBootstrap();this.boss = new NioEventLoopGroup();this.worker = new NioEventLoopGroup();this.serverBootstrap.group(boss, worker);this.serverBootstrap.channel(NioServerSocketChannel.class);this.port = port;}public void start() {serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new LengthFieldPrepender(2)).addLast(new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 2)).addLast(new MessageToMessageCodec<ByteBuf, Object>() {@Overrideprotected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out)throws Exception {out.add(Unpooled.buffer().writeBytes(JSON.toJSONString(msg, SerializerFeature.WriteClassName).getBytes(CharsetUtil.UTF_8)));}@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out)throws Exception {out.add(JSON.parse(msg.toString(CharsetUtil.UTF_8)));}}).addLast(new ChannelHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {MethodStaics methodStaics = (MethodStaics) msg;Object bean = act.getBean(methodStaics.getTargetInterface().getSimpleName());Method method = bean.getClass().getDeclaredMethod(methodStaics.getMethod(),methodStaics.getParameterTypes());Object invoke = method.invoke(bean, methodStaics.getArgs());// 如果返回值为空，则返回NullWritable实例代替传输invoke = null == invoke ? new NullWritable() : invoke;ChannelFuture channelFuture = ctx.writeAndFlush(invoke);channelFuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);channelFuture.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);channelFuture.addListener(ChannelFutureListener.CLOSE);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.err.println(cause);}});}});/*** 绑定监听端口并启动服务 注意 ： 启动的服务是阻塞的，防止阻塞Spring工厂需要采用异步启动*/new Thread() {public void run() {try {System.out.println("服务启动@" + port + " ...");ChannelFuture channelFuture = serverBootstrap.bind(port).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {System.out.println(e);} finally {}};}.start();}public void close() {boss.shutdownGracefully();worker.shutdownGracefully();}public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.act = applicationContext;} }
Class ： IUserService
package lime.pri.limeNio.netty.rpc03.service;import java.util.List;import lime.pri.limeNio.netty.rpc03.entity.User;public interface IUserService {User queryById(Integer id);List<User> queryByName(String name);List<User> queryAll(Integer pageSize, Integer pageNum); }
Class ： UserService
package lime.pri.limeNio.netty.rpc03.service.impl;import java.util.ArrayList; import java.util.Date; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap;import lime.pri.limeNio.netty.rpc03.entity.User; import lime.pri.limeNio.netty.rpc03.service.IUserService;public class UserService implements IUserService {private static Map<Integer, User> userMap = new ConcurrentHashMap<Integer, User>();static {for (int i = 1; i <= 100; i++) {userMap.put(i, new User(i, "lime_" + i, new Date()));}}public User queryById(Integer id) {return userMap.get(id);}public List<User> queryAll(Integer pageSize, Integer pageNum) {int stNum = (pageNum - 1) * pageSize + 1;int enNum = pageNum * pageSize;List<User> result = new ArrayList<User>();for (int i = stNum; i <= enNum; i++) {result.add(userMap.get(i));}return result;}public List<User> queryByName(String name) {List<User> result = null;Iterator<User> iterator = userMap.values().iterator();while (iterator.hasNext()) {User user = iterator.next();if (user.getName().equals(name)) {if (null == result)result = new ArrayList<User>();result.add(user);}}return result;}}
服务代理商：
Class ：
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"xmlns:p="http://www.springframework.org/schema/p"xsi:schemaLocation=" http://www.springframework.org/schema/beans classpath:/org/springframework/beans/factory/xml/spring-beans-4.1.xsd http://www.springframework.org/schema/context classpath:/org/springframework/context/config/spring-context-4.1.xsd http://mybatis.org/schema/mybatis-spring http://mybatis.org/schema/mybatis-spring.xsd http://www.springframework.org/schema/aop classpath:org/springframework/aop/config/spring-aop-4.1.xsd" default-lazy-init="false"><bean id="userService" class="lime.pri.limeNio.netty.rpc03.core.client.proxy.RPCObjectProxy"><constructor-arg index="0" ref="rpcClient" /><constructor-arg index="1" value="lime.pri.limeNio.netty.rpc03.service.IUserService" /></bean><bean id="rpcClient" class="lime.pri.limeNio.netty.rpc03.core.client.impl.RemoteRPCClient"destroy-method="close"><constructor-arg index="0" ref="hostAndPort" /></bean><bean id="hostAndPort" class="lime.pri.limeNio.netty.rpc03.core.client.assist.HostAndPort"><constructor-arg index="0" value="127.0.0.1" /><constructor-arg index="1" value="9999" /></bean></beans>
Class ： RPCObjectProxy
package lime.pri.limeNio.netty.rpc03.core.client.proxy;import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; import java.lang.reflect.Proxy;import org.springframework.beans.factory.FactoryBean;import lime.pri.limeNio.netty.rpc03.core.assist.MethodStaics; import lime.pri.limeNio.netty.rpc03.core.client.rpcClient.RPCClient;/*** 通过接口动态创建代理对象* * @author lime* @param <T>** 实现FactoryBean接口，与Spring整合* */ public class RPCObjectProxy implements InvocationHandler, FactoryBean<Object> {private RPCClient rpcClient;private Class<?> targetInterface;public RPCObjectProxy() {super();}public RPCObjectProxy(RPCClient rpcClient, Class<?> targetInterface) {super();this.rpcClient = rpcClient;this.targetInterface = targetInterface;}public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {return rpcClient.invokeMethod(new MethodStaics(targetInterface, method.getName(), args, method.getParameterTypes()));}// 产生代理对象public Object getObject() throws Exception {return Proxy.newProxyInstance(RPCObjectProxy.class.getClassLoader(), new Class[] { targetInterface }, this);}public Class<?> getObjectType() {return targetInterface;}public boolean isSingleton() {return true;} }
Class ： RPCClient
package lime.pri.limeNio.netty.rpc03.core.client.rpcClient;import lime.pri.limeNio.netty.rpc03.core.assist.MethodStaics;/*** 通过RPCClient实现对远程方法的调用* * @author lime**/ public interface RPCClient {Object invokeMethod(MethodStaics methodStaics);void close(); }
Class ： RemoteRPCClient
package lime.pri.limeNio.netty.rpc03.core.client.rpcClient.impl;import java.net.InetSocketAddress; import java.util.List;import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature;import io.netty.bootstrap.Bootstrap; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; 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; import io.netty.handler.codec.LengthFieldBasedFrameDecoder; import io.netty.handler.codec.LengthFieldPrepender; import io.netty.handler.codec.MessageToMessageCodec; import io.netty.util.CharsetUtil; import lime.pri.limeNio.netty.rpc03.core.assist.MethodStaics; import lime.pri.limeNio.netty.rpc03.core.assist.NullWritable; import lime.pri.limeNio.netty.rpc03.core.client.assist.HostAndPort; import lime.pri.limeNio.netty.rpc03.core.client.rpcClient.RPCClient;/*** 通过TCP/IP协议实现远程方法调用* * @author lime**/ public class RemoteRPCClient implements RPCClient {private Bootstrap bootstrap;private EventLoopGroup worker;private HostAndPort hostAndPort;public RemoteRPCClient() {super();}public RemoteRPCClient(HostAndPort hostAndPost) {this.hostAndPort = hostAndPost;// 初始化数据this.bootstrap = new Bootstrap();this.worker = new NioEventLoopGroup();this.bootstrap.group(this.worker);this.bootstrap.channel(NioSocketChannel.class);}public Object invokeMethod(final MethodStaics methodStaics) {bootstrap.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 2)).addLast(new LengthFieldPrepender(2)).addLast(new MessageToMessageCodec<ByteBuf, Object>() {@Overrideprotected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out)throws Exception {out.add(Unpooled.buffer().writeBytes(JSON.toJSONString(msg, SerializerFeature.WriteClassName).getBytes(CharsetUtil.UTF_8)));}@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out)throws Exception {out.add(JSON.parse(msg.toString(CharsetUtil.UTF_8)));}}).addLast(new ChannelHandlerAdapter() {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {ChannelFuture channelFuture = ctx.writeAndFlush(methodStaics);channelFuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);channelFuture.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {methodStaics.setResult(msg);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.err.println(cause);}});}});ChannelFuture channelFuture;try {channelFuture = bootstrap.connect(new InetSocketAddress(hostAndPort.getHost(), hostAndPort.getPort())).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {// TODO Auto-generated catch block e.printStackTrace();}// 服务端返回值为null，处理方式return methodStaics.getResult() instanceof NullWritable ? null : methodStaics.getResult();}public void close() {worker.shutdownGracefully();}}
Class ： HostAndPort
package lime.pri.limeNio.netty.rpc03.core.client.assist;import java.io.Serializable;public class HostAndPort implements Serializable{/*** */private static final long serialVersionUID = 1L;private String host;private int port;public HostAndPort() {super();// TODO Auto-generated constructor stub }public HostAndPort(String host, int port) {super();this.host = host;this.port = port;}public String getHost() {return host;}public void setHost(String host) {this.host = host;}public int getPort() {return port;}public void setPort(int port) {this.port = port;}@Overridepublic String toString() {return "HostAndPort [host=" + host + ", port=" + port + "]";}}
辅助类：
Class ： MethodStaics
package lime.pri.limeNio.netty.rpc03.core.assist;import java.io.Serializable; import java.util.Arrays;/*** @author lime**/ public class MethodStaics implements Serializable{/*** */private static final long serialVersionUID = 1L;private Class<?> targetInterface;private String method;private Object[] args;private Class[] parameterTypes;private Object result;public MethodStaics() {super();// TODO Auto-generated constructor stub }public MethodStaics(Class<?> targetInterface, String method, Object[] args, Class[] parameterTypes) {super();this.targetInterface = targetInterface;this.method = method;this.args = args;this.parameterTypes = parameterTypes;}@Overridepublic String toString() {return "MethodStaics [targetInterface=" + targetInterface + ", method=" + method + ", args="+ Arrays.toString(args) + ", parameterTypes=" + Arrays.toString(parameterTypes) + "]";}public Class<?> getTargetInterface() {return targetInterface;}public void setTargetInterface(Class<?> targetInterface) {this.targetInterface = targetInterface;}public String getMethod() {return method;}public void setMethod(String method) {this.method = method;}public Object[] getArgs() {return args;}public void setArgs(Object[] args) {this.args = args;}public Class[] getParameterTypes() {return parameterTypes;}public void setParameterTypes(Class[] parameterTypes) {this.parameterTypes = parameterTypes;}public Object getResult() {return result;}public void setResult(Object result) {this.result = result;}}
Class ： NullWritable
package lime.pri.limeNio.netty.rpc03.core.assist;import java.io.Serializable;public class NullWritable implements Serializable{/*** */private static final long serialVersionUID = 1L;}
Class ： User
package lime.pri.limeNio.netty.rpc03.entity;import java.io.Serializable; import java.util.Date;public class User implements Serializable {/*** */private static final long serialVersionUID = 1L;private int id;private String name;private Date birth;public User() {super();// TODO Auto-generated constructor stub }public User(int id, String name, Date birth) {super();this.id = id;this.name = name;this.birth = birth;}@Overridepublic String toString() {return "User [id=" + id + ", name=" + name + ", birth=" + birth + "]";}public int getId() {return id;}public void setId(int id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public Date getBirth() {return birth;}public void setBirth(Date birth) {this.birth = birth;}}
测试类：
Class ： TtServer
package lime.pri.limeNio.netty.rpc03.tT;import org.springframework.context.support.ClassPathXmlApplicationContext;public class TtServer {public static void main(String[] args) throws Exception {new ClassPathXmlApplicationContext("classpath:spring/rpc03_server.xml");} }
Class ： TtClient
package lime.pri.limeNio.netty.rpc03.tT;import java.util.List;import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext;import lime.pri.limeNio.netty.rpc03.entity.User; import lime.pri.limeNio.netty.rpc03.service.IUserService;public class TtClient {public static void main(String[] args) throws Exception {ApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:spring/rpc03_client.xml");IUserService userService = (IUserService) ctx.getBean("userService");System.out.println("queryByName");List<User> usersWithName = userService.queryByName("lime");System.out.println(usersWithName);System.out.println("queryAll");List<User> users = userService.queryAll(10, 3);for (User user : users) {System.out.println(user);}System.out.println("queryById");User user = userService.queryById(23);System.out.println(user);} }
Console ： Server
六月 25, 2017 2:08:04 下午 org.springframework.context.support.ClassPathXmlApplicationContext prepareRefresh 信息: Refreshing org.springframework.context.support.ClassPathXmlApplicationContext@17f052a3: startup date [Sun Jun 25 14:08:04 CST 2017]; root of context hierarchy 六月 25, 2017 2:08:04 下午 org.springframework.beans.factory.xml.XmlBeanDefinitionReader loadBeanDefinitions 信息: Loading XML bean definitions from class path resource [spring/rpc03_server.xml] 服务启动@9999 ...
Console ： Client
六月 25, 2017 2:08:18 下午 org.springframework.context.support.ClassPathXmlApplicationContext prepareRefresh 信息: Refreshing org.springframework.context.support.ClassPathXmlApplicationContext@17f052a3: startup date [Sun Jun 25 14:08:18 CST 2017]; root of context hierarchy 六月 25, 2017 2:08:18 下午 org.springframework.beans.factory.xml.XmlBeanDefinitionReader loadBeanDefinitions 信息: Loading XML bean definitions from class path resource [spring/rpc03_client.xml] queryByName null queryAll User [id=21, name=lime_21, birth=Sun Jun 25 14:08:04 CST 2017] User [id=22, name=lime_22, birth=Sun Jun 25 14:08:04 CST 2017] User [id=23, name=lime_23, birth=Sun Jun 25 14:08:04 CST 2017] User [id=24, name=lime_24, birth=Sun Jun 25 14:08:04 CST 2017] User [id=25, name=lime_25, birth=Sun Jun 25 14:08:04 CST 2017] User [id=26, name=lime_26, birth=Sun Jun 25 14:08:04 CST 2017] User [id=27, name=lime_27, birth=Sun Jun 25 14:08:04 CST 2017] User [id=28, name=lime_28, birth=Sun Jun 25 14:08:04 CST 2017] User [id=29, name=lime_29, birth=Sun Jun 25 14:08:04 CST 2017] User [id=30, name=lime_30, birth=Sun Jun 25 14:08:04 CST 2017] queryById User [id=23, name=lime_23, birth=Sun Jun 25 14:08:04 CST 2017]
啦啦啦
