先上一张图

使用netty 5.* 实现类似dubbo的服务调用

公共类

api

实际开发时 提供者需要实现该包下面所有接口, 而服务调用者则是通过该包内提供的接口调用方法

package apis;/*** Created by xuchonggao on 2017/6/26.*/
public interface UserService {public String findUser(String userName);String findUserList(int pageNum, int pageSize);
}

请求POJO

注意: 这个类必须是可序列化的,包括里面的属性

package dub;import java.io.Serializable;
import java.util.Arrays;/*** 调用请求类* Created by xuchonggao on 2017/6/27.*/
public class DubReq implements Serializable {private static final long serialVersionUID = 422805234202183587L;private Class interfaceClass;private String methonName;private Class[] paramTypes;private Object[] args;public DubReq() {}public Class getInterfaceClass() {return interfaceClass;}public DubReq setInterfaceClass(Class interfaceClass) {this.interfaceClass = interfaceClass;return this;}public String getMethonName() {return methonName;}public DubReq setMethonName(String methonName) {this.methonName = methonName;return this;}public Class[] getParamTypes() {return paramTypes;}public DubReq setParamTypes(Class[] paramTypes) {this.paramTypes = paramTypes;return this;}public Object[] getArgs() {return args;}public DubReq setArgs(Object[] args) {this.args = args;return this;}public DubReq(Class interfaceClass, String methonName, Class[] paramTypes, Object[] args) {this.interfaceClass = interfaceClass;this.methonName = methonName;this.paramTypes = paramTypes;this.args = args;}@Overridepublic String toString() {return "DubReq{" +"interfaceClass=" + interfaceClass +", methonName='" + methonName + '\'' +", paramTypes=" + Arrays.toString(paramTypes) +", args=" + Arrays.toString(args) +'}';}
}

服务端

package dub.server;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;/*** Created by xuchonggao on 2017/6/26.*/
public class DubServer {private int port;public DubServer(int port) {this.port = port;}public void run() throws Exception {EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap(); // (2)b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) // (3).childHandler(new ChannelInitializer<SocketChannel>() { // (4)@Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new ObjectDecoder(1024*1024, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));ch.pipeline().addLast(new ObjectEncoder());ch.pipeline().addLast(new DuBHandler());}}).option(ChannelOption.SO_BACKLOG, 128)          // (5).childOption(ChannelOption.SO_KEEPALIVE, true); // (6)// Bind and start to accept incoming connections.ChannelFuture f = b.bind(port).sync(); // (7)// Wait until the dub.server socket is closed.// In this example, this does not happen, but you can do that to gracefully// shut down your dub.server.f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {new DubServer(8080).run();}}

package dub.server;import apis.UserService;
import dub.DubReq;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import provider.UserServiceImpl;import java.lang.reflect.Method;/*** Created by xuchonggao on 2017/6/26.*/
public class DuBHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println("服务端收到消息:   " + msg);DubReq req = (DubReq) msg;// 1. 根据类名返回对象Object target = this.getInstenceByInterfaceClass(req.getInterfaceClass());// 2. 获取方法名String methodName = req.getMethonName();// 3. 获取方法参数类型// 4. 获取方法Method method = target.getClass().getMethod(methodName, req.getParamTypes());// 5. 获取参数值//调用方法 获取返回值Object res = method.invoke(new UserServiceImpl(), req.getArgs());// 写回给调用端ctx.writeAndFlush(res);}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}private Object getInstenceByInterfaceClass(Class clazz) {//根据接口返回对应的实例if (UserService.class.equals(clazz)) {return new UserServiceImpl();}return null;}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {super.exceptionCaught(ctx, cause);}
}

调用方

package dub.client;import apis.UserService;/*** Created by xuchonggao on 2017/6/26.*/
public class DubClient {public static void main(String[] args) throws Exception {// 实际上返回的是一个代理类  通过代理类发送网络请求调用服务提供者UserService userService  = (UserService) DubServiceProxy.remote(UserService.class);System.out.println(userService.findUser("aaa"));System.out.println(userService.findUser("0001"));System.out.println(userService.findUserList(1,10));}
}

package dub.client;import dub.DubReq;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;/*** 消费者handler* Created by xuchonggao on 2017/6/26.*/
public class DubConsumeHandler implements InvocationHandler {private Object res;public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {//配置客户端线程组EventLoopGroup group = new NioEventLoopGroup();try {//配置客户端启动辅助类Bootstrap b = new Bootstrap();b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new ObjectDecoder(1024, ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));ch.pipeline().addLast(new ObjectEncoder());ch.pipeline().addLast(new ConsumeHandler(proxy, method, args));}});//从注册中心获取ip和端口ChannelFuture f = b.connect("127.0.0.1", 8080).sync();f.channel().closeFuture().sync();} finally {group.shutdownGracefully();}return res;}private class ConsumeHandler extends ChannelInboundHandlerAdapter {private Object proxy;private Method method;private Object[] args;public ConsumeHandler(Object proxy, Method method, Object[] args) {this.proxy = proxy;this.method = method;this.args = args;}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 传输的对象必须实现序列化接口 包括其中的属性DubReq req = new DubReq(proxy.getClass().getInterfaces()[0], method.getName(), method.getParameterTypes(), args);ctx.write(req);ctx.flush();}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println("调用成功");res = msg;ctx.flush();//收到响应后断开连接ctx.close();}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}}}

package dub.client;import java.lang.reflect.Proxy;/*** 生成接口代理类* Created by xuchonggao on 2017/6/26.*/
public class DubServiceProxy {public static Object remote(Class clazz) {return Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, new DubConsumeHandler());}
}

结果显示

Connected to the target VM, address: '127.0.0.1:64393', transport: 'socket'
调用成功
userName:aaa
调用成功
userName:0001
调用成功
查询第1页,10条数据
Disconnected from the target VM, address: '127.0.0.1:64393', transport: 'socket'Process finished with exit code 0

码云地址

点我