Dubbo
Dubbo最早的定位是rpc框架掸鹅，即遠(yuǎn)程服務(wù)調(diào)用被辑，解決的是跨服務(wù)之間的方法調(diào)用問(wèn)題制恍，本文還是在這個(gè)定位基礎(chǔ)上嘗試手寫(xiě)一個(gè)簡(jiǎn)單的Dubbo
需求
首先要搭建測(cè)試的項(xiàng)目結(jié)構(gòu)雹熬，兩個(gè)服務(wù)consumer和provider宽菜，分別代表調(diào)用方和提供方，二者功能依賴(lài)于interface橄唬，其中暴露接口
interface包中定義一個(gè)接口
// interface
public interface HelloService {
String sayHello(String name);
}
provider實(shí)現(xiàn)
// provider
public class HelloServiceImpl implements HelloService {
public String sayHello(String name) {
return "hello "+name;
}
}
consumer調(diào)用
// consumer
public class Consumer {
public static void main(String[] args) {
// todo 獲取不到HelloService的實(shí)現(xiàn)
HelloService helloService = null;
System.out.println(helloService.sayHello("pq"));
}
}
當(dāng)前的需求即consumer服務(wù)調(diào)用provider服務(wù)里sayHello方法的實(shí)現(xiàn)赋焕，顯然當(dāng)前無(wú)法實(shí)現(xiàn)，這是一種遠(yuǎn)程發(fā)放調(diào)用仰楚，我們?cè)谛陆ㄒ粋€(gè)Module命名為dubbo隆判，意圖通過(guò)依賴(lài)它來(lái)實(shí)現(xiàn)遠(yuǎn)程方法的調(diào)用
網(wǎng)絡(luò)
由于跨服務(wù)了，所以遠(yuǎn)程調(diào)用必然是要走網(wǎng)絡(luò)的僧界，dubbo使用了netty侨嘀，我們也用netty來(lái)實(shí)現(xiàn)通訊
首先定義網(wǎng)絡(luò)請(qǐng)求的數(shù)據(jù)，遠(yuǎn)程調(diào)用需要的信息：哪個(gè)類(lèi)捂襟，哪個(gè)方法咬腕，什么參數(shù)，我們把這些信息封裝一下
// dubbo
@Data
@AllArgsConstructor
public class Invocation implements Serializable {
private String className;
private String methodName;
private Class<?>[] paramTypes;
private Object[] args;
}
服務(wù)端
provider作為服務(wù)的提供方葬荷，需要依靠netty搭建一個(gè)服務(wù)器涨共，當(dāng)接受到請(qǐng)求(Invocation對(duì)象)時(shí)，可以根據(jù)className宠漩，methodName等信息找到對(duì)應(yīng)的本地方法進(jìn)行調(diào)用
所以provider首先要維護(hù)一個(gè)map存儲(chǔ)className和class的對(duì)應(yīng)關(guān)系举反，這樣在收到請(qǐng)求時(shí)可以通過(guò)className找到對(duì)應(yīng)的類(lèi)，再通過(guò)反射獲取對(duì)應(yīng)的方法進(jìn)行調(diào)用
在我們的dubbo框架中封裝這么一個(gè)map結(jié)構(gòu)供provider使用
// dubbo
public class LocalRegister {
private static Map<String, Object> map = new HashMap<String, Object>();
public static void register(String className, Object impl) {
map.put(className, impl);
}
public static Object get(String className) {
return map.get(className);
}
}
然后再做一個(gè)處理請(qǐng)求netty服務(wù)供provider使用
// dubbo
public class NettyServer {
public void start(Integer port) {
try {
final ServerBootstrap bootstrap = new ServerBootstrap();
EventLoopGroup bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("bossGroup", true));
EventLoopGroup workerGroup = new NioEventLoopGroup(10, new DefaultThreadFactory("workerGroup", true));
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast("decoder", new ObjectDecoder(ClassResolvers
.weakCachingConcurrentResolver(this.getClass()
.getClassLoader())));
channel.pipeline().addLast("encoder", new ObjectEncoder());
channel.pipeline().addLast("handler", new RequestHandler());
}
});
ChannelFuture cf = bootstrap.bind(port).sync();
cf.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
對(duì)應(yīng)的handler如下
// dubbo
public class RequestHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
Invocation invocation = (Invocation) msg;
// 根據(jù)className獲取寄存的服務(wù)對(duì)象
Object serviceImpl = LocalRegister.get(invocation.getClassName());
// 通過(guò)methodName等信息獲取對(duì)應(yīng)的方法
Method method = serviceImpl.getClass().getMethod(invocation.getMethodName(), invocation.getParamTypes());
// 調(diào)用方法
Object result = method.invoke(serviceImpl, invocation.getArgs());
// 返回服務(wù)結(jié)果
ctx.writeAndFlush(result);
}
}
provider啟動(dòng)類(lèi)Starter
// provider
public class Starter {
public static void main(String[] args) {
// 存儲(chǔ)服務(wù)于名字映射關(guān)系
HelloServiceImpl helloService = new HelloServiceImpl();
String className = HelloService.class.getName();
LocalRegister.register(className, helloService);
// 開(kāi)啟netty服務(wù)
NettyServer nettyServer = new NettyServer();
System.out.println("provider 端口號(hào)9001");
nettyServer.start(9001);
}
}
代理
consumer只能拿到到HelloService接口扒吁，那么實(shí)例化的方法可以采用jdk動(dòng)態(tài)代理生成代理實(shí)現(xiàn)火鼻，而代理的實(shí)際執(zhí)行方式是通過(guò)netty網(wǎng)絡(luò)發(fā)送請(qǐng)求給provider，
首先還是在dubbo框架中封裝一個(gè)netty的客戶端供consumer發(fā)起請(qǐng)求
// dubbo
@Setter
public class NettyClient {
/**
* 管道上下文
*/
private volatile ChannelHandlerContext channelHandlerContext;
/**
* 返回消息暫存
*/
private Object message;
public void start(String hostName, Integer port) {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast("decoder", new ObjectDecoder(ClassResolvers
.weakCachingConcurrentResolver(this.getClass()
.getClassLoader())));
channel.pipeline().addLast("encoder", new ObjectEncoder());
channel.pipeline().addLast(new ResponseHandler(NettyClient.this));
}
});
bootstrap.connect(hostName, port).sync();
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 發(fā)送遠(yuǎn)程調(diào)用
* @param hostName
* @param port
* @param invocation
* @return
*/
public synchronized String send(String hostName, Integer port, Invocation invocation) {
start(hostName, port);
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
// 發(fā)送數(shù)據(jù)
channelHandlerContext.writeAndFlush(invocation);
// 等待
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
// 返回?cái)?shù)據(jù)
return message.toString();
}
}
其中的ResponseHandler入下
// dubbo
public class ResponseHandler extends ChannelInboundHandlerAdapter {
private final NettyClient client;
public ResponseHandler(NettyClient client) {
this.client = client;
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
synchronized (client) {
client.notify();
}
client.setChannelHandlerContext(ctx);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
client.setMessage(msg);
synchronized (client) {
client.notify();
}
}
}
然后在我們的dubbo框架中實(shí)現(xiàn)創(chuàng)建代理
// dubbo
public class ProxyFactory {
/**
* 根據(jù)接口創(chuàng)建代理 jdk動(dòng)態(tài)代理
* @param interfaceClass
* @param <T>
* @return
*/
public static <T> T getProxy(final Class<T> interfaceClass) {
return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class[]{interfaceClass}, new InvocationHandler() {
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
// 請(qǐng)求封裝成對(duì)象
Invocation invocation = new Invocation(interfaceClass.getName(), method.getName(), method.getParameterTypes(), args);
NettyClient nettyClient = new NettyClient();
// 發(fā)起網(wǎng)絡(luò)請(qǐng)求
String response = nettyClient.send("127.0.0.1", 9001, invocation);
return response;
}
});
}
}
最后回到consumer添加啟動(dòng)類(lèi)，通過(guò)代理創(chuàng)建HelloService的實(shí)現(xiàn)魁索，嘗試調(diào)用provider的sayHello方法
// consumer
public class Consumer {
public static void main(String[] args) {
HelloService helloService = ProxyFactory.getProxy(HelloService.class);
System.out.println(helloService.sayHello("pq"));
}
}
測(cè)試
- 啟動(dòng)
provider融撞，輸出如下
- 啟動(dòng)
consumer，輸出如下
證明已實(shí)現(xiàn)跨遠(yuǎn)程方法調(diào)用~
