网络编程 -- RPC实现原理 -- RPC -- 迭代版本V3 -- 远程方法调用 整合 Spring

<?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>

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;
}
}

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);
}

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;
}}

<?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>

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;
}
}

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();
}

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();
}}

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 + "]";
}}

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;
}}

package lime.pri.limeNio.netty.rpc03.core.assist;import java.io.Serializable;public class NullWritable implements Serializable{ /**
*
*/
private static final long serialVersionUID = 1L;}

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;
}}

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");
}
}

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);
}
}

六月 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 ...

六月 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]