netty是由JBOSS提供的一个java开源框架，现为 Github上的独立项目。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。
也就是说，Netty 是一个基于NIO的客户、服务器端的编程框架，使用Netty 可以确保你快速和简单的开发出一个网络应用，例如实现了某种协议的客户、服务端应用。Netty相当于简化和流线化了网络应用的编程开发过程，例如：基于TCP和UDP的socket服务开发。
1 构建工程 目录
api
public interface IRpcHelloService {
String hello(String name);
}
public interface IRpcService {
int add(int a, int b);
int sub(int a, int b);
int mult(int a, int b);
int div(int a, int b);
}
2 protocol
@Data
public class InvokerProtocol implements Serializable {
private String className;//服务名
private String methodName;//方法名
private Class[] prames;//形参列表
private Object[] values;//实例参数
}
3 provider
public class RpcHelloServiceImpl implements IRpcHelloService {
public String hello(String name) {
return "Hello " + name +"!";
}
}
public class RpcServiceImpl implements IRpcService {
public int add(int a, int b) {
return a + b;
}
public int sub(int a, int b) {
return a - b;
}
public int mult(int a, int b) {
return a * b;
}
public int div(int a, int b) {
return a / b;
}
3 proxy
public class RpcProxy {
public static T create(Class clazz){
MethodProxy methodProxy = new MethodProxy(clazz);
return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, methodProxy);
}
private static class MethodProxy implements InvocationHandler {
private Class clazz;
public MethodProxy(Class clazz) {
this.clazz = clazz;
}
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
if (Object.class.equals(method.getDeclaringClass()))
return method.invoke(this, args);
else
return rpcInvoker(proxy, method, args);
}
private Object rpcInvoker(Object proxy, Method method, Object[] args) {
//先构造一个消息协议的内容
//发起网络请求
EventLoopGroup worker = new NioEventLoopGroup();
try {
InvokerProtocol msg = new InvokerProtocol();
msg.setClassName(this.clazz.getName());
msg.setMethodName(method.getName());
msg.setPrames(method.getParameterTypes());
msg.setValues(args);
final RpcProxyHandler proxyHandler =new RpcProxyHandler();
Bootstrap client = new Bootstrap();
client.group(worker)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer(){
protected void initChannel(SocketChannel socketChannel) throws Exception {
//在netty中 把所有的业务逻辑中全部归到一个队列中 这个队列中包含了各种各样的逻辑
//pipleline
ChannelPipeline channelPipeline = socketChannel.pipeline();
//就是对我们处理逻辑的封装
//对自定义协议 进行编码 解码
channelPipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4, 0,4));
//定义编码器
channelPipeline.addLast(new LengthFieldPrepender(4));
//参数的处理 反序列化java识别的对象
channelPipeline. ("encoder", new ObjectEncoder());
channelPipeline.addLast("decoder", new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
//最后一步 完成对数据的解析 执行属于自己的逻辑
//1 注册 2 服务位置做登记
channelPipeline.addLast(proxyHandler);
}
});
ChannelFuture future = client.connect("localhost", 8080).sync();
future.channel().writeAndFlush(msg).sync();
future.channel().closeFuture().sync();
return proxyHandler.getResponse();
} catch (Exception e) {
e.printStackTrace();
} finally {
worker.shutdownGracefully();
}
return null;
}
}
}
public class RpcProxyHandler extends ChannelInboundHandlerAdapter {
private Object response;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {
response = msg;
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {
cause.printStackTrace();
System.out.println("client is exception");
}
public Object getResponse() {
return response;
}
}
4 register
public class RegisterHandler extends ChannelInboundHandlerAdapter {
private List<String> className = new ArrayList();
private Map <String, Object> registerMap = new ConcurrentHashMap();
//有客户端连上的时候 回调
public RegisterHandler() {
//根据包名 将class文件扫描
scanClass("rpcnetty.provider");
//根据包名 将class文件扫描到容器中
doRegister();
}
private void doRegister() {
if (className.isEmpty()) return;
for (String className : className) {
try {
Class clazz = Class.forName(className);
Class i = clazz.getInterfaces()[0];
String serviceName = i.getName();
//本来存的是网络路径 调用的时候解析
registerMap.put(serviceName, clazz.newInstance());
} catch (Exception e) {
e.printStackTrace();
}
}
}
private void scanClass(String name) {
URL url = this.getClass().getClassLoader().getResource(name.replaceAll("\\.", "/"));
File classpath = new File(url.getFile());
for (File file : classpath.listFiles()) {
if (file.isDirectory()) {
scanClass(name +"." + file.getName());
} else {
className.add(name +"." + file.getName().replaceAll(".class", ""));
}
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {
Object result = new Object();
InvokerProtocol request = (InvokerProtocol)msg;
//去容器中找到符合的服务对象
if (registerMap.containsKey(request.getClassName())) {
Object service = registerMap.get(request.getClassName());
Method method = service.getClass().getMethod(request.getMethodName(), request.getPrames());
result = method.invoke(service, request.getValues());
}
ctx.write(result);
ctx.flush();
ctx.close();
}
//发生异常 回调
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {
}
}
public class RegisterHandler extends ChannelInboundHandlerAdapter {
private List<String> className = new ArrayList();
private Map<String, Object> registerMap = new ConcurrentHashMap();
//有客户端连上的时候 回调
public RegisterHandler() {
//根据包名 将class文件扫描
scanClass("rpcnetty.provider");
//根据包名 将class文件扫描到容器中
doRegister();
}
private void doRegister() {
if (className.isEmpty()) return;
for (String className : className) {
try {
Class clazz = Class.forName(className);
Class i = clazz.getInterfaces()[0];
String serviceName = i.getName();
//本来存的是网络路径 调用的时候解析
registerMap.put(serviceName, clazz.newInstance());
} catch (Exception e) {
e.printStackTrace();
}
}
}
private void scanClass(String name) {
URL url = this.getClass().getClassLoader().getResource(name.replaceAll("\\.", "/"));
File classpath = new File(url.getFile());
for (File file : classpath.listFiles()) {
if (file.isDirectory()) {
scanClass(name +"." + file.getName());
} else {
className.add(name +"." + file.getName().replaceAll(".class", " "));
}
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
Object result = new Object();
InvokerProtocol request = (InvokerProtocol)msg;
//去容器中找到符合的服务对象
if (registerMap.containsKey(request.getClassName())) {
Object service = registerMap.get(request.getClassName());
Method method = service.getClass().getMethod(request.getMethodName(), request.getPrames());
result = method.invoke(service, request.getValues());
}
ctx.write(result);
ctx.flush();
ctx.close();
}
//发生异常 回调
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {
}
}
public class RpcRegister {
private int port =8080;
public RpcRegister(int port) {
this.port = port;
}
public void start() {
try {
ServerBootstrap server = new ServerBootstrap();
//基于 nio 来实现 selector主线程 worker 线程
//主线程池
EventLoopGroup bossGroup = new NioEventLoopGroup();
//子线程池
EventLoopGroup workGroup = new NioEventLoopGroup();
server.group(bossGroup, workGroup).channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer() {
protected void initChannel(SocketChannel ch)throws Exception {
/ /在netty中 把所有的业务逻辑中全部归到一个队列中 这个队列中包含了各种各样的逻辑
//pipleline
ChannelPipeline channelPipeline = ch.pipeline();
//就是对我们处理逻辑的封装
//对自定义协议 进行编码 解码
channelPipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4, 0,4));
//定义编码器
channelPipeline.addLast(new LengthFieldPrepender(4));
//参数的处理 反序列化java识别的对象
channelPipeline.addLast("encoder", new ObjectEncoder());
channelPipeline.addLast("decoder", new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
//最后一步 完成对数据的解析 执行属于自己的逻辑
//1 注册 2 服务位置做登记
channelPipeline.addLast(new RegisterHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
//正式启动服务相当于死循环轮询
ChannelFuture future = server.bind(this.port).sync();
System.out.println("GP RPC Register start listen at : " + this.port);
future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
new RpcRegister(8080).start();
}
}
consumer
public class RpcConsumer {
public static void main(String[] args) {
IRpcService service = RpcProxy.create(IRpcService.class);
System.out.println(" 8 + 2 = " + service.add(8, 2));
System.out.println(" 8 - 2 = " + service.sub(8, 2));
System.out.println(" 8 * 2 = " + service.mult(8, 2));
System.out.println(" 8 / 2 = " + service.div(8, 2));
}
测试 启动 注册中心
运行结果：
