Java領域要說讓我最服氣的RPC框架當屬Dubbo,原因有許多,但是最吸引我的還是它把遠程調用這個事情設計得很有藝術。
Java領域要說讓我最服氣的RPC框架當屬Dubbo,原因有許多,但是最吸引我的還是它把遠程調用這個事情設計得很有藝術。
業(yè)內對于微服務之間調用的框架選擇較多,主流是Spring Cloud的Rest方式 和 Dubbo方式,我使用Dubbo方式居多。Dubbo工業(yè)級可用,穩(wěn)定又高效,深受各大公司研發(fā)同學的喜愛。
Dubbo的優(yōu)點較多,比如:
但是,Dubbo最吸引人的,
半支煙
覺得反而倒是它的RPC調用。Dubbo的定位是一個RPC框架,這是它的核心和立足之地,所以Dubbo將RPC的調用過程透明化,使得開發(fā)者可以專注于業(yè)務邏輯,而不用關注底層通信問題。
一個RPC框架只有聚焦于先做好它的RPC調用過程這個模塊,才會有人關注,其余的優(yōu)點都是在這之后,慢慢迭代而來。
作者將RPC調用的這個過程,抽象成 一種協(xié)議消息的傳輸機制 ,再通過 控制好線程的等待和喚醒 ,來實現(xiàn)遠程方法調用。這一設計思路真是美妙,充分體驗了作者的智慧。
學Dubbo,首先就是要學習作者這種設計理念和思路。基于此,來實現(xiàn)一個簡易的遠程方法調用,將Dubbo的RPC過程簡易化。
簡易的RPC過程步驟如下,大致分5步,依舊使用Netty作用Socket通訊工具。
public class BProcessServer {
private final int port;
public BProcessServer(int port) {
this.port = port;
}
public void start() throws InterruptedException {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new BProcessServerHandler());
}
});
ChannelFuture future = bootstrap.bind(port).sync();
System.out.println("B啟動了服務,端口號: " + port);
future.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
new BProcessServer(8088).start();
}
}
public class BProcessServerHandler extends SimpleChannelInboundHandler {
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
String reqData = msg.toString(CharsetUtil.UTF_8);
System.out.println("B進程接受到了請求數(shù)據(jù): " + reqData);
executeMethod(ctx);
}
/**
* 執(zhí)行方法
*
* @param ctx
* @throws InterruptedException
*/
private void executeMethod(ChannelHandlerContext ctx) throws InterruptedException {
// TODO 將請求消息按照某種規(guī)則解析成方法名、方法參數(shù)等,其實就是反序列化的過程。
System.out.println("對接受的數(shù)據(jù)做反序列化,然后開始執(zhí)行 消息體里指定的方法...");
// 模擬方法執(zhí)行
Thread.sleep(2000);
System.out.println("執(zhí)行完畢,返回結果...");
// 將結果 通知給 A 進程
ByteBuf dataByteBuf = ctx.alloc().buffer().writeBytes("Task completed".getBytes(CharsetUtil.UTF_8));
ctx.writeAndFlush(dataByteBuf);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
public class AProcessClient {
private final String host;
private final int port;
private final Object lock = new Object(); // 監(jiān)視器對象
public AProcessClient(String host, int port) {
this.host = host;
this.port = port;
}
public void start() throws InterruptedException {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new AProcessClientHandler(lock));
}
});
ChannelFuture future = bootstrap.connect(host, port).sync();
System.out.println("A進程與B進程建立了通信連接");
Channel channel = future.channel();
// 發(fā)起遠程調用
callRemoteMethod(channel);
channel.closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
/**
* 執(zhí)行方法
*
* @param channel
* @throws InterruptedException
*/
private void callRemoteMethod(Channel channel) throws InterruptedException {
//TODO 此處需要將調用的方法和參數(shù),按照協(xié)議進行序列化。這次暫且省去此過程。
System.out.println("A進程將 請求的方法和參數(shù) 進行序列化,然后向B進程發(fā)起網(wǎng)絡調用...");
ByteBuf dataByteBuf = channel.alloc().buffer().writeBytes("Start call method".getBytes(CharsetUtil.UTF_8));
channel.writeAndFlush(dataByteBuf);
// 使用wait等待B進程通知
synchronized (lock) {
System.out.println("A進程等待B進程的響應...");
lock.wait(); // 等待通知
}
System.out.println("A進程收到了B進程的響應通知,繼續(xù)往下...");
}
public static void main(String[] args) throws InterruptedException {
new AProcessClient("localhost", 8088).start();
}
}
lock.wait()
以后的代碼得以繼續(xù)執(zhí)行。
public class AProcessClientHandler extends SimpleChannelInboundHandler {
private final Object lock;
public AProcessClientHandler(Object lock) {
this.lock = lock;
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
String resData = msg.toString(CharsetUtil.UTF_8);
System.out.println("A進程接受到了響應數(shù)據(jù): " + resData);
// B 進程任務完成,使用 notify 喚醒等待的線程
synchronized (lock) {
lock.notify(); // 喚醒 A 進程
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
Dubbo的優(yōu)秀設計思路有許多,我只鐘情其一,那就是RPC的調用過程。以上是一個簡易的RPC遠程調用的示例,用于理解Dubbo的原理和源碼,希望對你有幫助!
本篇完結!歡迎 關注、加V(yclxiao)交流、全網(wǎng)可搜(程序員半支煙)
原文鏈接: https://mp.weixin.qq.com/s/J0fzDH-iqGnnnjqaXMLs-A
count(*)、count(1)哪個更快?面試必問:通宵整理的十道經(jīng)典MySQL必問面試題
閱讀強化學習筆記之【ACE:Off-PolicyActor-CriticwithCausality-AwareEntropyRegularization】
閱讀使用MailKit在.NET Core中收發(fā)郵件的完整示例
閱讀OpenAI官方開源多智能體框架Swarm,社區(qū)反響熱烈
閱讀Vue-Vben-Admin:功能強大的Vue3后臺管理系統(tǒng)模板
閱讀llama.cpp:一個適用于中小型研發(fā)企業(yè)的高性能CPU/GPU大語言模型推理框架
閱讀本站所有軟件,都由網(wǎng)友上傳,如有侵犯你的版權,請發(fā)郵件[email protected]
湘ICP備2022002427號-10 湘公網(wǎng)安備:43070202000427號© 2013~2024 haote.com 好特網(wǎng)