使用Netty 模仿 Dubbo 实现简单的 远程调用
使用 java的反射 动态代理 加 Netty的远程访问 实现根据接口的RPC 远程调用
定义两个公共接口:
public interface HandlerServiceOne {
String handleOne(String mes);
}
public interface HandlerServiceTwo {
String handleTwo(String mes);
}
服务端
在服务端定义实现 一个给发客户端发来的字符串加上"$"符号 一个加上"*" 号:
public class HandlerServiceOneImpl implements HandlerServiceOne {
@Override
public String handleOne(String mes) {
return"$$$$$$$" + mes + "$$$$$$$$$$$";
}
}
public class HandlerServiceTwoImpl implements HandlerServiceTwo {
@Override
public String handleTwo(String mes) {
return "*****" + mes + "*****";
}
}
启动服务端 并将两个实现类初始化 并注册进Map 中 键为 实现接口的全类名 值为 接口的实例
public class ServerBootstrap {
public static Map<String,Object> serviceMap = new HashMap();
public static void main(String[] args) {
//两个接口的全类名
String handlerServiceTwo = HandlerServiceTwo.class.getName();
String handlerServiceOne = HandlerServiceOne.class.getName();
//注册服务
serviceMap.put(handlerServiceTwo,new HandlerServiceTwoImpl());
serviceMap.put(handlerServiceOne,new HandlerServiceOneImpl());
startServer("127.0.0.1", 7000);
}
private static void startServer(String hostname, int port) {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
io.netty.bootstrap.ServerBootstrap serverBootstrap = new io.netty.bootstrap.ServerBootstrap();
serverBootstrap.group(bossGroup,workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new StringDecoder());
pipeline.addLast(new StringEncoder());
pipeline.addLast(new NettyServerHandler()); //业务处理器
}
}
);
ChannelFuture channelFuture = serverBootstrap.bind(hostname, port).sync();
System.out.println("服务提供方开始提供服务~~");
channelFuture.channel().closeFuture().sync();
}catch (Exception e) {
e.printStackTrace();
}
finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
服务端处理请求的自定义Handler 根据自定义的规范 解析出调用的 接口全类名 方法名 和参数 使用 反射调用实例
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//获取客户端发送的消息，并调用服务
System.out.println("msg=" + msg);
//客户端在调用服务器的api 时，我们需要定义一个协议
//比如我们要求 每次发消息是都必须以某个字符串开头 "HandlerServiceTwo#hello#你好"
String[] message = msg.toString().split("#");
String serviceName = message[0];
String methodName = message[1];
String arg = message[2];
Object service = ServerBootstrap.serviceMap.get(serviceName);
Object result = Class.forName(serviceName)
.getMethod(methodName,String.class)
.invoke(service, arg);
ctx.writeAndFlush(result);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
客户端
先来看看客户端自定义的 Handler
public class NettyClientHandler extends ChannelInboundHandlerAdapter implements Callable {
private ChannelHandlerContext context;//上下文
private String result; //返回的结果
private String para; //客户端调用方法时，传入的参数
//与服务器的连接创建后，就会被调用, 这个方法是第一个被调用
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
context = ctx;
}
//收到服务器的数据后，调用方法 (4)
//
@Override
public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
result = msg.toString();
notify(); //唤醒等待的线程
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
//被代理对象调用, 发送数据给服务器，-> wait -> 等待被唤醒(channelRead) -> 返回结果 (3)-》5
@Override
public synchronized Object call() throws Exception {
context.writeAndFlush(para);
//进行wait 等待channelRead 方法获取到服务器的结果后，唤醒
wait();
//服务方返回的结果
return result;
}
void setPara(String para) {
this.para = para;
}
}
实现 Callable 的接口 重写 call 方法 , 可以将此类的实例作为一个任务线程执行 并返回 call方法返回的结果
再看看 制造客户端的Factory类 ,此类可以根据接口 返回动态代理的实现类 并持有一个 Netty的客户端 作为此动态代理的实际执行类, 每次调用 接口的代理类时,都会获取 是哪个接口的全类名 , 调用的方法和调用的参数 并拼装在一起(规则可以自定义), 作为字符串 传给Handler的属性 并执行Handler 的call方法 ,
在call方法中 将字符串发送给服务端,,但此时不能直接返回 要等到客户端发送会数据 作为返回值,所以使线程等待 并等到读取到服务端返回的数据时 再进行释放 ,再回到call方法中返回 返回值
public class ClientFactory<T> {
//创建线程池
private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
private Class<T> clazz;
private static String host = "127.0.0.1" ;
private static int port = 7000;
public ClientFactory(Class<T> clazz) {
this.clazz = clazz;
}
/**
* 使用动态代理模式，获取一个代理对象
*/
public T getBean() {
//创建 netty客户端 与服务端交互 并让动态代理类持有
final NettyClientHandler clientHandler = this.getClientHandler();
return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
new Class<?>[]{clazz}, (proxy, method, args) -> {
//设置要发给服务器端的信息 接口的全类名#方法名#参数
clientHandler.setPara(this.initParams(method,args));
return executor.submit(clientHandler).get();
});
}
private String initParams(Method method,Object[] args) {
return method.getDeclaringClass().getName()+"#"+method.getName()+"#"+ args[0];
}
/**
* 初始化客户端
*/
private NettyClientHandler getClientHandler() {
NettyClientHandler nettyClientHandler = new NettyClientHandler();
//创建EventLoopGroup
NioEventLoopGroup group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(
new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new StringDecoder());
pipeline.addLast(new StringEncoder());
pipeline.addLast(nettyClientHandler);
}
}
);
try {
bootstrap.connect(host,port).sync();
} catch (Exception e) {
e.printStackTrace();
}
return nettyClientHandler;
}
}
调用 根据工厂类 制造两个接口的代理类 并进行调用:
public class ClientBootstrap {
public static void main(String[] args) throws Exception {
//创建代理对象
HandlerServiceTwo handlerServiceTwo = new ClientFactory<>(HandlerServiceTwo.class).getBean();
HandlerServiceOne handlerServiceOne = new ClientFactory<>(HandlerServiceOne.class).getBean();
//通过代理对象调用服务提供者的方法(服务)
System.out.println(handlerServiceTwo.handleTwo("Hello"));
System.out.println(handlerServiceOne.handleOne("Hello"));
}
}
控制台打印:
调用成功,本例有很多待优化的地方 可以自己发挥想象