前言
Dubbo是基于Netty搭建的RPC框架，為了更好地理解Netty在Dubbo中的應(yīng)用沫浆，仿照Dubbo搭建了一個(gè)簡易版的RPC框架眠冈。
概述
整個(gè)調(diào)用邏輯如下：
1旋膳、生產(chǎn)者服務(wù)端啟動(dòng)Netty服務(wù)端。
2熊户、消費(fèi)者客戶端通過JDK動(dòng)態(tài)代理啟動(dòng)Netty客戶端萍膛，通過注冊(cè)中心地址連接生產(chǎn)者服務(wù)端，同時(shí)將接口調(diào)用信息（接口嚷堡、方法蝗罗、參數(shù)等）先序列化再發(fā)送給生產(chǎn)者服務(wù)端。
3蝌戒、生產(chǎn)者服務(wù)端接收消息并通過反射調(diào)用相應(yīng)方法绿饵，然后返回調(diào)用結(jié)果給消費(fèi)者。
4瓶颠、消費(fèi)者接收生產(chǎn)者傳來的調(diào)用結(jié)果拟赊。
實(shí)現(xiàn)
新建DubboRequest類(相當(dāng)于POJO)，作為消息載體
package com.beidao.netty.dubbo.facade.api;
import java.io.Serializable;
import java.util.Arrays;
/**
* dubbo請(qǐng)求類
* @author 0200759
*
*/
public class DubboRequest implements Serializable{
private static final long serialVersionUID = 422805234202183587L;
private Class<?> interfaceClass;
private String methodName;
private Class<?>[] paramTypes;
private Object[] args;
public DubboRequest(Class<?> interfaceClass, String methodName, Class<?>[] paramTypes, Object[] args) {
this.interfaceClass = interfaceClass;
this.methodName = methodName;
this.paramTypes = paramTypes;
this.args = args;
}
public Class<?> getInterfaceClass() {
return interfaceClass;
}
public void setInterfaceClass(Class<?> interfaceClass) {
this.interfaceClass = interfaceClass;
}
public String getMethodName() {
return methodName;
}
public void setMethodName(String methodName) {
this.methodName = methodName;
}
public Class<?>[] getParamTypes() {
return paramTypes;
}
public void setParamTypes(Class<?>[] paramTypes) {
this.paramTypes = paramTypes;
}
public Object[] getArgs() {
return args;
}
public void setArgs(Object[] args) {
this.args = args;
}
@Override
public String toString() {
return "DubboRequest{" +
"interfaceClass=" + interfaceClass +
", methodName='" + methodName + '\'' +
", paramTypes=" + Arrays.toString(paramTypes) +
", args=" + Arrays.toString(args) +
'}';
}
}
新建Dubbo消費(fèi)者調(diào)用接口IUserFacade
package com.beidao.netty.dubbo.facade.api;
/**
* dubbo api接口
* @author 0200759
*
*/
public interface IUserFacade {
/**
* 返回用戶名接口
* @param string
* @return
*/
public String getUserName(Long id);
}
新建Dubbo消費(fèi)者攔截器
package com.beidao.netty.dubbo.client;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import com.beidao.netty.dubbo.facade.api.DubboRequest;
import io.netty.bootstrap.Bootstrap;
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.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.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
/**
* dubbo消費(fèi)者攔截器
* @author 0200759
*
*/
public class DubboConsumerHandler 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 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 {
ch.pipeline().addLast(new ObjectDecoder(1024, ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
ch.pipeline().addLast(new ObjectEncoder());
ch.pipeline().addLast(new ConsumerHandler(proxy, method, args));
}
});
//從注冊(cè)中心獲取服務(wù)端ip和端口
ChannelFuture f = bootstrap.connect("127.0.0.1", 8080).sync();
f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
return res;
}
/**
*
* netty-dubbo消費(fèi)者攔截器
* @author 0200759
*
*/
private class ConsumerHandler extends ChannelInboundHandlerAdapter{
private Object proxy;
private Method method;
private Object[] args;
public ConsumerHandler(Object proxy, Method method, Object[] args) {
this.proxy = proxy;
this.args = args;
this.method = method;
}
public void channelActive(ChannelHandlerContext ctx) {
// 傳輸?shù)膶?duì)象必須實(shí)現(xiàn)序列化接口 包括其中的屬性
DubboRequest req = new DubboRequest(proxy.getClass().getInterfaces()[0], method.getName(), method.getParameterTypes(), args);
ctx.write(req);
ctx.flush();
}
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("調(diào)用成功");
res = msg;
ctx.flush();
//收到響應(yīng)后斷開連接
ctx.close();
}
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
}
}
新建Dubbo服務(wù)代理類
package com.beidao.netty.dubbo.client;
import java.lang.reflect.Proxy;
/**
* Dubbo代理類
* @author 0200759
*
*/
public class DubboProxy {
public static Object getProxyInstance(Class<?> clazz) {
return Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, new DubboConsumerHandler());
}
}
新建Dubbo消費(fèi)者
package com.beidao.netty.dubbo.client;
import com.beidao.netty.dubbo.facade.api.IUserFacade;
/**
* dubbo客戶端(消費(fèi)者)
* @author 0200759
*
*/
public class DubboClient {
public static void main(String[] args){
IUserFacade userFacade = (IUserFacade) DubboProxy.getProxyInstance(IUserFacade.class);
System.out.println(userFacade.getUserName(520L));
System.out.println(userFacade.getUserName(1314L));
System.out.println(userFacade.getUserName(1314520L));
}
}
新建 dubbo服務(wù)端實(shí)現(xiàn)類
package com.beidao.netty.dubbo.facade.impl;
import com.beidao.netty.dubbo.facade.api.IUserFacade;
/**
* dubbo服務(wù)端實(shí)現(xiàn)類
* @author 0200759
*
*/
public class UserFacade implements IUserFacade {
public String getUserName(Long id) {
return "I love you, "+id;
}
}
dubbo生產(chǎn)者攔截器
package com.beidao.netty.dubbo.sever;
import java.lang.reflect.Method;
import com.beidao.netty.dubbo.facade.api.DubboRequest;
import com.beidao.netty.dubbo.facade.api.IUserFacade;
import com.beidao.netty.dubbo.facade.impl.UserFacade;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
/**
* netty-dubbo服務(wù)端攔截器
* @author 0200759
*
*/
public class DubboServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("服務(wù)端收到消息: " + msg);
DubboRequest req = (DubboRequest) msg;
// 1. 根據(jù)類名返回對(duì)象
Object target = this.getInstenceByInterfaceClass(req.getInterfaceClass());
// 2. 獲取方法名
String methodName = req.getMethodName();
// 3. 獲取方法參數(shù)類型
// 4. 獲取方法
Method method = target.getClass().getMethod(methodName, req.getParamTypes());
// 5. 獲取參數(shù)值
//調(diào)用方法 獲取返回值
Object res = method.invoke(target, req.getArgs());
// 寫回給調(diào)用端
ctx.writeAndFlush(res);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
/**
* 根據(jù)接口返回對(duì)應(yīng)的實(shí)例
* @param clazz
* @return
*/
private Object getInstenceByInterfaceClass(Class<?> clazz) {
if (IUserFacade.class.equals(clazz)) {
return new UserFacade();
}
return null;
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
}
}
Dubbo生產(chǎn)者服務(wù)端
package com.beidao.netty.dubbo.sever;
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;
/**
* @author 0200759
*
*/
public class DubboServer {
private int port;
public DubboServer(int port) {
this.port = port;
}
public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public 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 DubboServerHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(port).sync(); // (7)
f.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
new DubboServer(8080).run();
}
}
驗(yàn)證
1粹淋、啟動(dòng)生產(chǎn)者服務(wù)端DubboServer吸祟。
2、啟動(dòng)消費(fèi)者調(diào)用端DubboClient桃移。
消費(fèi)者客戶端控制臺(tái)顯示如下：
調(diào)用成功
I love you, 520
調(diào)用成功
I love you, 1314
調(diào)用成功
I love you, 1314520
生產(chǎn)者服務(wù)端控制臺(tái)顯示如下：
服務(wù)端收到消息: DubboRequest{interfaceClass=interface com.beidao.netty.dubbo.facade.api.IUserFacade, methodName='getUserName', paramTypes=[class java.lang.Long], args=[520]}
服務(wù)端收到消息: DubboRequest{interfaceClass=interface com.beidao.netty.dubbo.facade.api.IUserFacade, methodName='getUserName', paramTypes=[class java.lang.Long], args=[1314]}
服務(wù)端收到消息: DubboRequest{interfaceClass=interface com.beidao.netty.dubbo.facade.api.IUserFacade, methodName='getUserName', paramTypes=[class java.lang.Long], args=[1314520]}
源碼地址：https://github.com/MAXAmbitious/netty-study/tree/master/netty-dubbo
