网络编程 -- RPC实现原理 -- RPC -- 迭代版本V3 -- 远程方法调用 整合 Spring
Posted limeOracle
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了网络编程 -- RPC实现原理 -- RPC -- 迭代版本V3 -- 远程方法调用 整合 Spring相关的知识，希望对你有一定的参考价值。
啦啦啦
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>() { @Override protected 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>() { @Override protected 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))); } @Override protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception { out.add(JSON.parse(msg.toString(CharsetUtil.UTF_8))); } }).addLast(new ChannelHandlerAdapter() { @Override public 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); } @Override public 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>() { @Override protected 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>() { @Override protected 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))); } @Override protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception { out.add(JSON.parse(msg.toString(CharsetUtil.UTF_8))); } }).addLast(new ChannelHandlerAdapter() { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { ChannelFuture channelFuture = ctx.writeAndFlush(methodStaics); channelFuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); channelFuture.addListener(ChannelFutureListener.CLOSE_ON_FAILURE); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { methodStaics.setResult(msg); } @Override public 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; } @Override public 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; } @Override public 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; } @Override public 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]
啦啦啦
以上是关于网络编程 -- RPC实现原理 -- RPC -- 迭代版本V3 -- 远程方法调用 整合 Spring的主要内容，如果未能解决你的问题，请参考以下文章
网络编程 -- RPC实现原理 -- RPC -- 迭代版本V3 -- 远程方法调用 整合 Spring
网络编程 -- RPC实现原理 -- NIO多线程 -- 迭代版本V2