基于Netty的异步Rpc调用的小框架

基于netty写的一个异步Rpc调用小框架，欢迎拍砖，新手。

客户端与服务端通信的类

package cc.ymsoft.Framework;

import java.io.Serializable;

@SuppressWarnings("serial")
public class MethodAndArgs implements Serializable{
private String methodName;//调用的方法名称
private Class<?>[] types;//参数类型
private Object[] objects;//参数列表
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;
}
public MethodAndArgs() {
super();
// TODO Auto-generated constructor stub
}
public MethodAndArgs(String methodName, Class<?>[] types, Object[] objects) {

this.methodName = methodName;
this.types = types;
this.objects = objects;
}

}

 框架类，有两个静态方法，regist（在服务器上注册服务）和getobjt（获得接口的代理类）

/**
* @author xulang
*/
package cc.ymsoft.Framework;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
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.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
/**
* 服务端处理
* @author hadoop
*
*/
class TcpServerHandler extends ChannelInboundHandlerAdapter {

private Object obj;
private Object response;

public TcpServerHandler(Object obj) {
super();
this.obj = obj;

}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
// TODO Auto-generated method stub
MethodAndArgs methodAndArgs=(MethodAndArgs) msg;
Method method=obj.getClass().getMethod(methodAndArgs.getMethodName(), methodAndArgs.getTypes());
ctx.writeAndFlush(method.invoke(obj, methodAndArgs.getObjects()));
ctx.close();
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
// TODO Auto-generated method stub
System.out.println("client die");
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// TODO Auto-generated method stub
System.out.println("channelActive>>>>>>>>");
ctx.writeAndFlush("调用异常");
ctx.close();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("服务器异常");
}
}
/**
* 客户端处理
* @author hadoop
*
*/
class TcpClientHander 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");
}
}

public class RpcFramework {

/**
* 服务注册
* @param obj 需要注册的服务对象
* @param port 端口
* @param ip 地址
* @throws InterruptedException
*/
public static void regist(final Object obj,int port,String ip) throws InterruptedException {
int BIZGROUPSIZE = Runtime.getRuntime().availableProcessors()*2;

int BIZTHREADSIZE = 100;
EventLoopGroup bossGroup = new NioEventLoopGroup(BIZGROUPSIZE);
EventLoopGroup workerGroup = new NioEventLoopGroup(BIZTHREADSIZE);
if (obj == null)
throw new IllegalArgumentException("对象不能为null");
if (port <= 0 || port > 65535)
throw new IllegalArgumentException("错误的端口" + port);
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup);
bootstrap.channel(NioServerSocketChannel.class);
bootstrap.childHandler(new ChannelInitializer<Channel>() {

@Override
protected void initChannel(Channel ch) throws Exception {
// TODO Auto-generated method stub
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 StringDecoder(CharsetUtil.UTF_8));
// pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
pipeline.addLast(new TcpServerHandler(obj));
}
});
ChannelFuture f = bootstrap.bind(ip, port).sync();
f.channel().closeFuture().sync();
System.out.println("TCP服务器已启动");
}
@SuppressWarnings("unchecked")
public static <T>T getObj(Class<T> interfaceClass,final String host,final int port) {
if (interfaceClass == null)
throw new IllegalArgumentException("接口类型不能为空");
if (!interfaceClass.isInterface())
throw new IllegalArgumentException("类名" + interfaceClass.getName() + "必须是接口");
if (host == null || host.length() == 0)
throw new IllegalArgumentException("目标主机不能为空");
if (port <= 0 || port > 65535)
throw new IllegalArgumentException("端口错误：" + port);

return (T)	Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class<?>[]{interfaceClass}, new InvocationHandler() {

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

MethodAndArgs mArgs=new MethodAndArgs(method.getName(), method.getParameterTypes(), args);
final TcpClientHander tcpClientHander=new TcpClientHander();
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group);
//     b.channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true);
b.channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true);

b.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected 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",tcpClientHander);
}
});

ChannelFuture f = b.connect(host, port).sync();

f.channel().writeAndFlush(mArgs).sync();
f.channel().closeFuture().sync();

} catch (Exception e) {

} finally {

group.shutdownGracefully();
}
return tcpClientHander.getResponse();

}
});
}
}

  测试

接口

package cc.ymsoft.test;

interface HelloService {
String SayHello(String name);
}

 接口实现类

package cc.ymsoft.test;

public class HelloImp implements HelloService {

@Override
public String SayHello(String name) {
// TODO Auto-generated method stub

return "你好:"+name;
}

}

 客户端

package cc.ymsoft.test;

import cc.ymsoft.Framework.RpcFramework;

public class HelloInvoke {

public static void main(String[] args) throws Exception {
final HelloService helloService = RpcFramework.getObj(HelloService.class, "127.0.0.1", 1717);

System.out.println(helloService.SayHello("XL"));

}

}

 服务端

package cc.ymsoft.test;

import cc.ymsoft.Framework.RpcFramework;

public class HelloPro {
public static void main(String[] args) throws Exception {
HelloService hello=new HelloImp();
RpcFramework.regist(hello, 1717, "127.0.0.1");
}

}

完整代码在github https://github.com/xulang/NettyRpc