从源码全面解析 dubbo 服务端服务调用的来龙去脉

  • 👏作者简介：大家好，我是爱敲代码的小黄，独角兽企业的Java开发工程师，CSDN博客专家，阿里云专家博主
  • 📕系列专栏：Java设计模式、Spring源码系列、Netty源码系列、Kafka源码系列、JUC源码系列、duubo源码系列
  • 🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
  • 🍂博主正在努力完成2023计划中：以梦为马，扬帆起航，2023追梦人
  • 📝联系方式：hls1793929520，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬👀

在这里插入图片描述

一、引言

对于 Java 开发者而言，关于 dubbo ，我们一般当做黑盒来进行使用，不需要去打开这个黑盒。

但随着目前程序员行业的发展，我们有必要打开这个黑盒，去探索其中的奥妙。

本期 dubbo 源码解析系列文章，将带你领略 dubbo 源码的奥秘

本期源码文章吸收了之前 Spring、Kakfa、JUC源码文章的教训，将不再一行一行的带大家分析源码，我们将一些不重要的部分当做黑盒处理，以便我们更快、更有效的阅读源码。

虽然现在是互联网寒冬，但乾坤未定，你我皆是黑马！

废话不多说，发车！

二、服务端调用流程

对于整个服务端调用来说，主要分为两部分：

  • 服务端启动时：封装的 Handler 和 Filter
  • 服务端调用时：经过 Handler，然后经过 Filter，最后调用目标方法

1、启动封装

我们启动的时候，会经过 Exporter<?> exporter = protocolSPI.export(invoker) ，走 Dubbo SPI 的扩展机制

1.1 过滤器的封装

首先是我们的过滤器的封装：ProtocolFilterWrapper

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {FilterChainBuilder builder = getFilterChainBuilder(invoker.getUrl());// buildInvokerChain:return protocol.export(builder.buildInvokerChain(invoker, SERVICE_FILTER_KEY, CommonConstants.PROVIDER));
}public <T> Invoker<T> buildInvokerChain(final Invoker<T> originalInvoker, String key, String group) {// 获取所有的过滤器filters = ScopeModelUtil.getExtensionLoader(Filter.class, moduleModels.get(0)).getActivateExtension(url, key, group);// 如果当前的过滤器不为空if (!CollectionUtils.isEmpty(filters)) {// 将所有的过滤器封装成链表（last）的形式for (int i = filters.size() - 1; i >= 0; i--) {final Filter filter = filters.get(i);final Invoker<T> next = last;last = new CopyOfFilterChainNode<>(originalInvoker, next, filter);}// 将过滤器链表封装成CallbackRegistrationInvoker类型return new CallbackRegistrationInvoker<>(last, filters);} 
}

到这里，我们将过滤器封装成一个链表并且将其封装成 CallbackRegistrationInvoker 形式

我们直接跳到 DubboProtocol.export 中：

public <T> Exporter<T> export(Invoker<T> invoker){// 得到当前注册Zookeeper的URLURL url = invoker.getUrl();// 根据URL得到唯一的key:cn/com.common.service.IUserService:1.0.0.test:20883// 分组(group) + 接口(intfenerce) + 版本(1.0.0.test) + 端口号(20883)String key = serviceKey(url);// DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
}public DubboExporter(Invoker<T> invoker, String key, Map<String, Exporter<?>> exporterMap) {super(invoker);this.key = key;this.exporterMap = exporterMap;// 将当前的过滤器放至exporterMap中,方便我们后面的获取exporterMap.put(key, this);// 打开服务openServer(url);optimizeSerialization(url);return exporter;
}

到这里，我们的过滤器就封装到了 exporterMap 里面，后面会用到，我们后面再聊

1.2 Hander的封装

在我们上面封装完过滤器之后，我们会进行 openServer 打开服务这个操作，该操作会进行 Handler 的封装并启动我们的 Netty 服务

这里的Handler 一共封装成下面的流程：

NettyServerhandler -> NettyServer -> MultiMessageHandler--->HeartbeatHandler---->AllChannelHandler -> DecodeHandler  -> HeaderExchangeHandler -> ExchangeHandlerAdapter

最终会走到 NettyServer 的 doOpen 方法：这里对 Netty 不太清楚的，可以看博主的 Netty 源码文章：【Netty 从成神到升仙系列 大结局】全网一图流死磕解析 Netty 源码

// 典型的Netty启动的流程
protected void doOpen() throws Throwable {bootstrap = new ServerBootstrap();bossGroup = createBossGroup();workerGroup = createWorkerGroup();final NettyServerHandler nettyServerHandler = createNettyServerHandler();channels = nettyServerHandler.getChannels();// 初始化我们的服务端启动器initServerBootstrap(nettyServerHandler);ChannelFuture channelFuture = bootstrap.bind(getBindAddress());channelFuture.syncUninterruptibly();channel = channelFuture.channel();
}protected void initServerBootstrap(NettyServerHandler nettyServerHandler) {boolean keepalive = getUrl().getParameter(KEEP_ALIVE_KEY, Boolean.FALSE);bootstrap.group(bossGroup, workerGroup).channel(NettyEventLoopFactory.serverSocketChannelClass()).option(ChannelOption.SO_REUSEADDR, Boolean.TRUE).childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE).childOption(ChannelOption.SO_KEEPALIVE, keepalive).childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {int idleTimeout = UrlUtils.getIdleTimeout(getUrl());NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);if (getUrl().getParameter(SSL_ENABLED_KEY, false)) {ch.pipeline().addLast("negotiation", new SslServerTlsHandler(getUrl()));}// 这里添加Netty的Handler责任链ch.pipeline()// 解码器.addLast("decoder", adapter.getDecoder())// 编码器.addLast("encoder", adapter.getEncoder()).addLast("server-idle-handler", new IdleStateHandler(0, 0, idleTimeout, MILLISECONDS))// 把上面封装的Handler放到Netty中,便于我们的调用执行.addLast("handler", nettyServerHandler);}});
}

这里如果对 Netty 责任链不熟悉的可参考：【Netty 从成神到升仙系列 五】Netty 的责任链真有这么神奇吗？

2、服务调用

我们上篇文章剖析了 消费端 是如何进行的服务调用：从源码全面解析 dubbo 消费端服务调用的来龙去脉

这篇我们来看下服务端是如何进行服务调用的

2.1 Handler调用

我们直接跳到 NettyServerHandler 的 channelRead 的方法

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);// 主要看这个Handler做的事情handler.received(channel, msg);ctx.fireChannelRead(msg);
}

image-20230627221354346

我们从上图看到，重点是这五个 Handler ：

  • MultiMessageHandler：处理多个数据包发送的消息，也称为“多包消息”
  • HeartbeatHandler：定期发送“心跳”消息，以确保连接仍然存在并响应正常
  • AllChannelHandler：管理所有通道的打开和关闭
  • DecodeHandler：将二进制数据解码为消息对象。
  • HeaderExchangeHandler：负责协议头部的交换和处理

我们挨个去看看他们实际的作用

2.1 MultiMessageHandler
  • 如果当前的请求是多个的话，需要进行切割成单个请求往下传递
  • 如果是单个请求的话，直接向下传递即可
public void received(Channel channel, Object message) throws RemotingException {if (message instanceof MultiMessage) {MultiMessage list = (MultiMessage) message;for (Object obj : list) {try {handler.received(channel, obj);}}} else {handler.received(channel, message);}
}
2.2 HeartbeatHandler
  • 服务端：判断当前的请求是不是心跳请求，如果是心跳请求的话，发送心跳请求
  • 消费端：判断当前的请求是不是心跳请求，处理心跳请求
public void received(Channel channel, Object message) throws RemotingException {// 记录最近读取的时间setReadTimestamp(channel);// 判断当前的请求是不是心跳请求if (isHeartbeatRequest(message)) {Request req = (Request) message;if (req.isTwoWay()) {// 如果当前是同一个心跳检测则返回同一个响应Response res = new Response(req.getId(), req.getVersion());res.setEvent(HEARTBEAT_EVENT);channel.send(res);if (logger.isDebugEnabled()) {int heartbeat = channel.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0);}}return;}// 消费端使用:处理心跳响应if (isHeartbeatResponse(message)) {return;}handler.received(channel, message);
}
2.3 AllChannelHandler
  • 为每一个服务端的请求从线程池中分配一个线程执行
public void received(Channel channel, Object message) throws RemotingException {// 获取线程ExecutorService executor = getPreferredExecutorService(message);try {// 将当前的Handler丢进线程池里面执行executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));}
}

image-20230627224241967

2.4 DecodeHandler
  • 根据当前的响应请求进行解码操作
public void received(Channel channel, Object message) throws RemotingException {if (message instanceof Decodeable) {decode(message);}if (message instanceof Request) {decode(((Request) message).getData());}if (message instanceof Response) {decode(((Response) message).getResult());}handler.received(channel, message);
}
2.5 HeaderExchangeHandler
  • 调用我们的过滤器并等待数据的返回
  • 将数据通过 Channel 返回至客户端
public void received(Channel channel, Object message) throws RemotingException {final ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);// 服务端：当前的消息是请求if (message instanceof Request) {Request request = (Request) message;if (request.isEvent()) {handlerEvent(channel, request);} else {// 进行请求的解析if (request.isTwoWay()) {handleRequest(exchangeChannel, request);} else {handler.received(exchangeChannel, request.getData());}}} else if (message instanceof Response) {handleResponse(channel, (Response) message);} else {handler.received(exchangeChannel, message);}
}void handleRequest(final ExchangeChannel channel, Request req) throws RemotingException {Response res = new Response(req.getId(), req.getVersion());// Request [id=17, version=2.0.2, twoWay=true, event=false, broken=false, data=RpcInvocation [methodName=getUserById, parameterTypes=[class java.lang.Long]]]Object msg = req.getData();// 执行过滤器的操作CompletionStage<Object> future = handler.reply(channel, msg);// 等待数据的返回future.whenComplete((appResult, t) -> {try {if (t == null) {res.setStatus(Response.OK);res.setResult(appResult);} else {res.setStatus(Response.SERVICE_ERROR);res.setErrorMessage(StringUtils.toString(t));}// 没有问题的话，将我们的数据返回channel.send(res);} });
}

服务端解码之后的消息：

image-20230627225422895

2.2 过滤器调用

我们直接跳到 DubboProtocol 的 reply 方法

public CompletableFuture<Object> reply(ExchangeChannel channel, Object message){Invocation inv = (Invocation) message;// 得到过滤器的invokerInvoker<?> invoker = getInvoker(channel, inv);// 执行过滤器Result result = invoker.invoke(inv);// 返回结果return result.thenApply(Function.identity());
}Invoker<?> getInvoker(Channel channel, Invocation inv){// 根据组+版本号+接口确定唯一的keyString serviceKey = serviceKey(port,path,(String) inv.getObjectAttachmentWithoutConvert(VERSION_KEY),(String) inv.getObjectAttachmentWithoutConvert(GROUP_KEY));// 得到过滤器(我们在上面进行过对应的添加)DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey);// 返回过滤器return exporter.getInvoker();
}

这里的过滤器总共十个，这里不带大家细看每一个的源码了，了解意思即可

  • ContextFilter：负责将请求上下文传递给请求处理流程中的其他组件
  • ProfilerServerFilter：负责性能分析和监视
  • EchoFilter：将请求消息作为响应消息返回
  • ClassLoaderFilter：负责加载和管理类加载器
  • GenericFilter：提供一种通用的请求处理机制
  • ExceptionFilter：负责处理异常并生成错误响应消息
  • MonitorFilter：负责监视请求处理过程中的状态和信息
  • TimeoutFilter：负责处理请求处理超时
  • TraceFilter：跟踪请求处理流程中的各个阶段和信息
  • ClassLoaderCallbackFilter：提供了回调函数的机制

2.3 方法调用

最终我们会到 AbstractProxyInvoker 的 invoke 方法

public Result invoke(Invocation invocation){// 执行我们动态代理的方法Object value = doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments());// 等待返回的结果CompletableFuture<Object> future = wrapWithFuture(value, invocation);// 封装请求CompletableFuture<AppResponse> appResponseFuture = future.handle((obj, t) -> {AppResponse result = new AppResponse(invocation);if (t != null) {if (t instanceof CompletionException) {result.setException(t.getCause());} else {result.setException(t);}} else {result.setValue(obj);}// 返回数据// AppResponse [value=User(id=2, name=天涯, age=12), exception=null]return result;});return new AsyncRpcResult(appResponseFuture, invocation);
}

这里可以参考这篇文章：从源码全面解析 dubbo 服务暴露的来龙去脉

public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {try {final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);return new AbstractProxyInvoker<T>(proxy, type, url) {@Override// proxy:实现类// methodName：getUserById// parameterTypes：class java.lang.Long// arguments:2// 执行相对应的方法即可protected Object doInvoke(T proxy, String methodName,Class<?>[] parameterTypes,Object[] arguments) throws Throwable {return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);}};}
}

三、流程图

高清图片可私信博主

在这里插入图片描述

四、总结

鲁迅先生曾说：独行难，众行易，和志同道合的人一起进步。彼此毫无保留的分享经验，才是对抗互联网寒冬的最佳选择。

其实很多时候，并不是我们不够努力，很可能就是自己努力的方向不对，如果有一个人能稍微指点你一下，你真的可能会少走几年弯路。

如果你也对 后端架构和中间件源码 有兴趣，欢迎添加博主微信：hls1793929520，一起学习，一起成长

我是爱敲代码的小黄，独角兽企业的Java开发工程师，CSDN博客专家，喜欢后端架构和中间件源码。

我们下期再见。

我从清晨走过，也拥抱夜晚的星辰，人生没有捷径，你我皆平凡，你好，陌生人，一起共勉。

往期文章推荐：

  • 美团二面：聊聊ConcurrentHashMap的存储流程
  • 从源码全面解析Java 线程池的来龙去脉
  • 从源码全面解析LinkedBlockingQueue的来龙去脉
  • 从源码全面解析 ArrayBlockingQueue 的来龙去脉
  • 从源码全面解析ReentrantLock的来龙去脉
  • 阅读完synchronized和ReentrantLock的源码后，我竟发现其完全相似
  • 从源码全面解析 ThreadLocal 关键字的来龙去脉
  • 从源码全面解析 synchronized 关键字的来龙去脉
  • 阿里面试官让我讲讲volatile，我直接从HotSpot开始讲起，一套组合拳拿下面试

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/4353.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

如何删除Git仓库中的敏感文件及其历史记录

如何删除Git仓库中的敏感文件及其历史记录

本文主要介绍如何使用 git filter-branch 命令删除 Git 仓库中的敏感文件及其历史记录。在 Git 中&#xff0c;我们通常会将敏感信息(如密码、私钥等)存储在 .gitignore 文件中&#xff0c;以防止这些信息被意外提交到仓库。有时候&#xff0c;因为疏忽或私有仓库转公开仓库&am…
阅读更多...
RocketMQ --- 高级篇

RocketMQ --- 高级篇

一、高级功能 1.1、消息存储 分布式队列因为有高可靠性的要求&#xff0c;所以数据要进行持久化存储。 消息生成者发送消息MQ收到消息&#xff0c;将消息进行持久化&#xff0c;在存储中新增一条记录返回ACK给生产者MQ push 消息给对应的消费者&#xff0c;然后等待消费者返回…
阅读更多...
Python 使用 NetworkX

Python 使用 NetworkX

Python 使用 NetworkX 说明&#xff1a;本篇文章主要讲述 python 使用 networkx 绘制有向图&#xff1b; 1. 介绍&安装 NetworkX 是一个用于创建、操作和研究复杂网络的 Python 库。它提供了丰富的功能&#xff0c;可以帮助你创建、分析和可视化各种类型的网络&#xff…
阅读更多...
基于OpenCV的人脸对齐步骤详解及源码实现

基于OpenCV的人脸对齐步骤详解及源码实现

目录 1. 前言2. 人脸对齐基本原理与步骤3. 人脸对齐代码实现 1. 前言 在做人脸识别的时候&#xff0c;前期的数据处理过程通常会遇到一个问题&#xff0c;需要将各种人脸从不同尺寸的图像中截取出来&#xff0c;再进行人脸对齐操作&#xff1a;即将人脸截取出来并将倾斜的人脸…
阅读更多...
76、基于STM32单片机车牌识别摄像头图像处理扫描设计(程序+原理图+PCB源文件+相关资料+参考PPT+元器件清单等)

76、基于STM32单片机车牌识别摄像头图像处理扫描设计(程序+原理图+PCB源文件+相关资料+参考PPT+元器件清单等)

单片机主芯片选择方案 方案一&#xff1a;AT89C51是美国ATMEL公司生产的低电压&#xff0c;高性能CMOS型8位单片机&#xff0c;器件采用ATMEL公司的高密度、非易失性存储技术生产&#xff0c;兼容标准MCS-51指令系统&#xff0c;片内置通用8位中央处理器(CPU)和Flash存储单元&a…
阅读更多...
使用Streamlit和Matplotlib创建交互式折线图

使用Streamlit和Matplotlib创建交互式折线图

大家好&#xff0c;本文将介绍使用Streamlit和Matplotlib创建一个用户友好的数据可视化Web应用程序。该应用程序允许上传CSV文件&#xff0c;并为任何选定列生成折线图。 构建Streamlit应用程序 在本文中&#xff0c;我们将指导完成创建此应用程序的步骤。无论你是专家还是刚刚…
阅读更多...
CTF安全竞赛介绍

CTF安全竞赛介绍

目录 一、赛事简介 二、CTF方向简介 1.Web&#xff08;Web安全&#xff09; &#xff08;1&#xff09;简介 &#xff08;2&#xff09;涉及主要知识 2.MISC&#xff08;安全杂项&#xff09; &#xff08;1&#xff09;介绍 &#xff08;2&#xff09;涉及主要知识 3…
阅读更多...
设计模式之行为型模式

设计模式之行为型模式

本文已收录于专栏 《设计模式》 目录 概念说明大话设计模式行为型模式 各模式详解第一组观察者模式&#xff08;Observer Pattern&#xff09;模板方法模式&#xff08;Template Method Pattern&#xff09;命令模式&#xff08;Command Pattern&#xff09;状态模式&#xff0…
阅读更多...
智能指针+拷贝构造+vector容器+多态引起的bug

智能指针+拷贝构造+vector容器+多态引起的bug

今天在调试一段代码的时候&#xff0c;VC编译提示&#xff1a; error C2280: “T485CommCtrlPara::T485CommCtrlPara(const T485CommCtrlPara &)”: 尝试引用已删除的函数 函数执行部分如下&#xff1a; 看意思是这个pComm485Pro已经消亡了&#xff0c;后续push_back到ve…
阅读更多...
【Linux】MHA高可用配置及故障切换

【Linux】MHA高可用配置及故障切换

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 MHA高可用配置及故障切换 MHA高可用配置及故障切换MHA概述MHA 的组成MHA 的特点 搭建 MySQL MHA1&#xff0e;Master、Slave1、Slave2 节点上安装 mysql5.72&#xff0e;修改…
阅读更多...
Web项目思路

Web项目思路

登录注册 账号密码登录&#xff0c;注册时生成唯一id。 找回密码 通过验证邮箱来修改现在的密码。 主页 显示最受欢迎的博主&#xff0c;点赞次数最多的博客&#xff0c;根据用户收藏夹中的偏好推荐博客等 遍历数据库&#xff0c;找到粉丝数最多的博主和点赞次数最多的博客…
阅读更多...
配置本地的application-dev.yml读取nacos上面的配置

配置本地的application-dev.yml读取nacos上面的配置

我想配置本地的application-dev.yml读取nacos上面的配置&#xff0c;我应该在配置文件里怎么写&#xff1f;并且在nacos里怎么建立 在本地的application-dev.yml文件中&#xff0c;您可以使用Spring Cloud Nacos来读取Nacos上的配置。以下是在配置文件和Nacos中设置的步骤&…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023