首页 / 更多教程 / SparkRPC源码分析之RPC管道与消息类型

SparkRPC源码分析之RPC管道与消息类型

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了SparkRPC源码分析之RPC管道与消息类型，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含11774字，纯文字阅读大概需要17分钟。

内容图文

SparkRPC源码分析之RPC管道与消息类型
我们前面看过了netty基础知识扫盲，那我们应该明白，ChannelHandler这个组件内为channel的各种事件提供了处理逻辑，也就是主要业务逻辑写在该组建内。Spark的RPC也不会例外，因此我们看一下Spark的Handler怎么调用的。在TransPortClientFactory初始化客户端之前有一条代码为TransportChannelHandler clientHandler = context.initializePipeline(ch);这里的context定义的地方为private final TransportContext context;也就时我们接下来看TransoprtContext类的方法，代码如下

public TransportChannelHandler initializePipeline(SocketChannel channel) {
return initializePipeline(channel, rpcHandler);
}
1
2
3
可以看到这里的initializePipeline调用了另一个initializePipeline方法，它的代码如下

public TransportChannelHandler initializePipeline(
SocketChannel channel,
RpcHandler channelRpcHandler) {
try {
TransportChannelHandler channelHandler = createChannelHandler(channel, channelRpcHandler);
channel.pipeline()
.addLast("encoder", ENCODER)
.addLast(TransportFrameDecoder.HANDLER_NAME, NettyUtils.createFrameDecoder())
.addLast("decoder", DECODER)
.addLast("idleStateHandler", new IdleStateHandler(0, 0, conf.connectionTimeoutMs() / 1000))
// NOTE: Chunks are currently guaranteed to be returned in the order of request, but this
// would require more logic to guarantee if this were not part of the same event loop.
.addLast("handler", channelHandler);
return channelHandler;
} catch (RuntimeException e) {
logger.error("Error while initializing Netty pipeline", e);
throw e;
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
这里面和我们前面netty基础知识扫盲里面做的内容很类似，就是给pipeline启动添加了一些Handler处理逻辑
通过addLast添加的Handler会被依次执行顺序或倒序，那么我们就来依次看一些他的Handler都干了什么。

addLast(“encoder”, ENCODER)
服务器端用来编码服务器到客户端响应的编码器。通过调用消息的encode()方法对其进行编码。对于非数据消息，将添加一个ByteBuf到“out”，其中包含总帧长度、消息类型和消息本身。在ChunkFetchSuccess的情况下，我们还将与数据对应的ManagedBuffer添加到“Out”，以便启用零拷贝传输。一般会出现的消息类型如下

0 ChunkFetchRequest;
1 ChunkFetchSuccess;
2 ChunkFetchFailure;
3 RpcRequest;
4 RpcResponse;
5 RpcFailure;
6 StreamRequest;
7 StreamResponse;
8 StreamFailure;
9 OneWayMessage;
1 UploadStream;
-1 User
1
2
3
4
5
6
7
8
9
10
11
12
.addLast(TransportFrameDecoder.HANDLER_NAME, NettyUtils.createFrameDecoder())
- 一种允许截取原始数据的定制帧解码器。
- 类似于Netty的帧解码器(具有符合此库需要的硬编码参数)但是不同的是它在封装成帧之前允许去装拦截器直接去读取数据
- 与Netty的帧解码器不同，每个帧在解码后立即被发送给子处理程序，而不是尽可能多地放入当前缓冲区一次性发出去。这允许子处理程序在需要时安装拦截器。
- 如果安装了拦截器，则停止封装成帧，数据将直接输入拦截器，当拦截器指示它不需要读取任何更多数据时,封装恢复

.addLast(“decoder”, DECODER)
客户端用来解码服务器到客户端响应的解码器。消息类型和加密端一样不再重复写了

.addLast(“idleStateHandler”, new IdleStateHandler(0, 0, conf.connectionTimeoutMs() / 1000))
在服务器和客户端之间一定时间内没有数据交互时, 即处于 idle【空闲】状态时, 客户端或服务器会发送一个特殊的数据包给对方, 当接收方收到这个数据报文后, 也立即发送一个特殊的数据报文, 回应发送方, 此即一个 PING-PONG 交互，确保TCP连接有效

.addLast(“handler”, channelHandler);

channelHandler的创建代码如下

TransportChannelHandler channelHandler = createChannelHandler(channel, channelRpcHandler);
1
createChannelHandler代码如下

private TransportChannelHandler createChannelHandler(Channel channel, RpcHandler rpcHandler) {
TransportResponseHandler responseHandler = new TransportResponseHandler(channel);
TransportClient client = new TransportClient(channel, responseHandler);
TransportRequestHandler requestHandler = new TransportRequestHandler(channel, client,
rpcHandler, conf.maxChunksBeingTransferred());
return new TransportChannelHandler(client, responseHandler, requestHandler,
conf.connectionTimeoutMs(), closeIdleConnections);
}
1
2
3
4
5
6
7
8
值得注意的一点，我们可以看到这里面有客户端的初始化new TransportClient(channel, responseHandler);也许大家会有疑惑，我们前面才看了代码TransportClientFactory中有初始化TransportClient的代码，怎么这里也有呢？

这里分析一下TransportClientFactory中创建TransportClient时的情况，可以看到代码如下

final AtomicReference<TransportClient> clientRef = new AtomicReference<>();
final AtomicReference<Channel> channelRef = new AtomicReference<>();
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
TransportChannelHandler clientHandler = context.initializePipeline(ch);
clientRef.set(clientHandler.getClient());
channelRef.set(ch);
}
});
…… 省略掉一部分代码
TransportClient client = clientRef.get();
…… 省略掉一部分代码
return client;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
可以看到这里的客户端与其说时创建不如说是获取，从clientHandler中获取，这么看来，客户端真正的创建的地方是在new关键字出现的地方，也就是这里，而TransportClientFactory中的创建不过是从这边取到的而已。

接着看TransportChannelHandler这个类到底为何方神圣？

从类图上可以看出来，这个类实现了ChannelInboundHandler接口，那么这个接口是干什么的呢？

ChannelInboundHandler是一个netty的组件，它是一个常用的Handler。这个Handler的作用就是处理接收到数据时的事件，我们的业务逻辑一般就是写在这个Handler里面。

这个TransportChannelHandler的处理业务逻辑是什么呢？看下面代码可知它重写了channelRead方法

@Override
public void channelRead(ChannelHandlerContext ctx, Object request) throws Exception {
if (request instanceof RequestMessage) {
requestHandler.handle((RequestMessage) request);
} else if (request instanceof ResponseMessage) {
responseHandler.handle((ResponseMessage) request);
} else {
ctx.fireChannelRead(request);
}
}
1
2
3
4
5
6
7
8
9
10
这里它主要判断请求是什么类型的数据，根据类型交给TransportResponseHandler或者TransportRequestHandler的对象去处理。

这里可以看出无论是TransportRequestHandler还是TransportResponseHandler都是继承于MessageHandler抽象类。
那么我们就来看一下MessageHandler，看一下他的方法发现上面调用的handle方法都是来自于重写该类的方法.

public abstract class MessageHandler<T extends Message> {
//处理单个消息的接收。
public abstract void handle(T message) throws Exception;
//当MessageHandler所在的通道处于活动状态时调用
public abstract void channelActive();
//在通道上捕获异常时调用
public abstract void exceptionCaught(Throwable cause);
//当MessageHandler所在的通道处于非活动状态时调用
public abstract void channelInactive();
}
1
2
3
4
5
6
7
8
9
10
那么我们先看TransportRequestHandler重写的handle方法

@Override
public void handle(RequestMessage request) {
if (request instanceof ChunkFetchRequest) {
processFetchRequest((ChunkFetchRequest) request);
} else if (request instanceof RpcRequest) {
processRpcRequest((RpcRequest) request);
} else if (request instanceof OneWayMessage) {
processOneWayMessage((OneWayMessage) request);
} else if (request instanceof StreamRequest) {
processStreamRequest((StreamRequest) request);
} else if (request instanceof UploadStream) {
processStreamUpload((UploadStream) request);
} else {
throw new IllegalArgumentException("Unknown request type: " + request);
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
可以看出，在这里用了if-else逻辑判断消息的类型，然后再交给相应的方法去处理。那么一共有多少消息呢？它都可以处理什么消息呢？请看类图

那么现在再看一下TransportResponseHandler它复写的handle方法逻辑如下

@Override
public void handle(ResponseMessage message) throws Exception {
if (message instanceof ChunkFetchSuccess) {
ChunkFetchSuccess resp = (ChunkFetchSuccess) message;
ChunkReceivedCallback listener = outstandingFetches.get(resp.streamChunkId);
if (listener == null) {
logger.warn("Ignoring response for block {} from {} since it is not outstanding",
resp.streamChunkId, getRemoteAddress(channel));
resp.body().release();
} else {
outstandingFetches.remove(resp.streamChunkId);
listener.onSuccess(resp.streamChunkId.chunkIndex, resp.body());
resp.body().release();
}
} else if (message instanceof ChunkFetchFailure) {
ChunkFetchFailure resp = (ChunkFetchFailure) message;
ChunkReceivedCallback listener = outstandingFetches.get(resp.streamChunkId);
if (listener == null) {
logger.warn("Ignoring response for block {} from {} ({}) since it is not outstanding",
resp.streamChunkId, getRemoteAddress(channel), resp.errorString);
} else {
outstandingFetches.remove(resp.streamChunkId);
listener.onFailure(resp.streamChunkId.chunkIndex, new ChunkFetchFailureException(
"Failure while fetching " + resp.streamChunkId + ": " + resp.errorString));
}
} else if (message instanceof RpcResponse) {
RpcResponse resp = (RpcResponse) message;
RpcResponseCallback listener = outstandingRpcs.get(resp.requestId);
if (listener == null) {
logger.warn("Ignoring response for RPC {} from {} ({} bytes) since it is not outstanding",
resp.requestId, getRemoteAddress(channel), resp.body().size());
} else {
outstandingRpcs.remove(resp.requestId);
try {
listener.onSuccess(resp.body().nioByteBuffer());
} finally {
resp.body().release();
}
}
} else if (message instanceof RpcFailure) {
RpcFailure resp = (RpcFailure) message;
RpcResponseCallback listener = outstandingRpcs.get(resp.requestId);
if (listener == null) {
logger.warn("Ignoring response for RPC {} from {} ({}) since it is not outstanding",
resp.requestId, getRemoteAddress(channel), resp.errorString);
} else {
outstandingRpcs.remove(resp.requestId);
listener.onFailure(new RuntimeException(resp.errorString));
}
} else if (message instanceof StreamResponse) {
StreamResponse resp = (StreamResponse) message;
Pair<String, StreamCallback> entry = streamCallbacks.poll();
if (entry != null) {
StreamCallback callback = entry.getValue();
if (resp.byteCount > 0) {
StreamInterceptor<ResponseMessage> interceptor = new StreamInterceptor<>(
this, resp.streamId, resp.byteCount, callback);
try {
TransportFrameDecoder frameDecoder = (TransportFrameDecoder)
channel.pipeline().get(TransportFrameDecoder.HANDLER_NAME);
frameDecoder.setInterceptor(interceptor);
streamActive = true;
} catch (Exception e) {
logger.error("Error installing stream handler.", e);
deactivateStream();
}
} else {
try {
callback.onComplete(resp.streamId);
} catch (Exception e) {
logger.warn("Error in stream handler onComplete().", e);
}
}
} else {
logger.error("Could not find callback for StreamResponse.");
}
} else if (message instanceof StreamFailure) {
StreamFailure resp = (StreamFailure) message;
Pair<String, StreamCallback> entry = streamCallbacks.poll();
if (entry != null) {
StreamCallback callback = entry.getValue();
try {
callback.onFailure(resp.streamId, new RuntimeException(resp.error));
} catch (IOException ioe) {
logger.warn("Error in stream failure handler.", ioe);
}
} else {
logger.warn("Stream failure with unknown callback: {}", resp.error);
}
} else {
throw new IllegalStateException("Unknown response type: " + message.type());
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
详细代码暂且不管，大体可以看出也是使用if-else逻辑判断消息的类型，然后分别进行处理。那么我么来看一些这里的消息。
---------------------

内容总结

以上是互联网集市为您收集整理的SparkRPC源码分析之RPC管道与消息类型全部内容，希望文章能够帮你解决SparkRPC源码分析之RPC管道与消息类型所遇到的程序开发问题。如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/803685.html

来源：【匿名】

【上一篇】月薪5K和月薪50K的程序员，差距都在哪里？【下一篇】关于IE的RegExp.exec的问题

更多 ►

【SparkRPC源码分析之RPC管道与消息类型】教程文章相关的互联网学习教程文章

飞鸽传书源码分析四-消息发送【代码】【图】

转载请注明出处：http://blog.csdn.net/mxway/article/details/44569831 本篇文章是在飞鸽传书源码v2.06的基础上进行分析的飞鸽传书是一款工作在局域网的软件，支持局域网里不同设备之间的消息发送及文件的传输（消息发送使用udp，文件传输使用tcp）。发送消息及文件传输是在飞鸽传书的发送对话框中进行，而发送对话框的打开是通过双击拖盘（win7系统）到任务栏的图标。而拖盘到任务栏的这个图标就是飞鸽传书的主窗口，对应的源...

精尽MyBatis源码分析 - MyBatis初始化（四）之 SQL 初始化（下）

摘自：https://www.cnblogs.com/lifullmoon/p/14015075.html该系列文档是本人在学习 Mybatis 的源码过程中总结下来的，可能对读者不太友好，请结合我的源码注释（Mybatis源码分析 GitHub 地址、Mybatis-Spring 源码分析 GitHub 地址、Spring-Boot-Starter 源码分析 GitHub 地址）进行阅读 MyBatis 版本：3.5.2 MyBatis-Spring 版本：2.0.3 MyBatis-Spring-Boot-Starter 版本：2.1.4MyBatis的初始化在MyBatis初始化过程中，大致会有...

xxl-job源码分析【代码】【图】

xxl-job 系统说明安装安装部署参考文档：分布式任务调度平台xxl-job 功能定时调度、服务解耦、灵活控制跑批时间（停止、开启、重新设定时间、手动触发） XXL-JOB是一个轻量级分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用概念 1、执行器列表：一个执行器是一个项目 2、任务：一个任务是一个项目中的 JobHandler 3、一个xxl-job服务可以有多个执...

Dubbo 源码分析 - 集群容错之 Router【代码】

1. 简介上一篇文章分析了集群容错的第一部分 – 服务目录 Directory。服务目录在刷新 Invoker 列表的过程中，会通过 Router 进行服务路由。上一篇文章关于服务路由相关逻辑没有细致分析，一笔带过了，本篇文章将对此进行详细的分析。首先，先来介绍一下服务目录是什么。服务路由包含一条路由规则，路由规则决定了服务消费者的调用目标，即规定了服务消费者可调用哪些服务提供者。Dubbo 目前提供了三种服务路由实现，分别为条件路由...

Mybaits源码分析九之sql执行流程【代码】

SubjectDao subjectDao=sqlSession.getMapper(SubjectDao.class);对应的源码为：1 @Override 2 public <T> T getMapper(Class<T> type) { 3 return configuration.<T>getMapper(type, this); 4 }1 public <T> T getMapper(Class<T> type, SqlSession sqlSession) { 2 return mapperRegistry.getMapper(type, sqlSession); 3 } 1 @SuppressWarnings("unchecked")2 public <T> T getMapper(Class<T> type, SqlSess...

spring boot启动源码分析 afterRefresh【代码】

1 protected void afterRefresh(ConfigurableApplicationContext context,2 ApplicationArguments args) {3 callRunners(context, args);4 }5 6 private void callRunners(ApplicationContext context, ApplicationArguments args) {7 List<Object> runners = new ArrayList<Object>();8 runners.addAll(context.getBeansOfType(ApplicationRunner.class).values());9 ru...

一、dubbo源码分析【图】

一、整体设计dubbo整体设计以及调用用链路参照官网?http://dubbo.apache.org/zh-cn/docs/dev/design.html 二、dubbo的注册中心1、注册中心参照官网 ?http://dubbo.apache.org/zh-cn/docs/user/references/registry/introduction.html；2、zk注册中心详解2.1、目录结构+- dubbo+- com.demo.service.HelloService+- consumers+- consumer://192.168.1.102/com.demo.service.HelloService?application=dubbo-demo-annotation-consumer...

ForkJoinPool 源码分析

ForkJoinPool ForkJoinPool 是一个运行 ForkJoinTask 任务、支持工作窃取和并行计算的线程池核心参数+创建实例// 工作者线程驻留任务队列索引位static final int SWIDTH = 16; // 低 16 位掩码static final int SMASK = 0xffff; // 最大并行度：#workers - 1static final int MAX_CAP = 0x7fff; // 最大工作队列数、提交队列数static final int SQMASK = 0x007e; // 工作者线...

3.2spring源码系列----循环依赖源码分析【代码】【图】

首先,我们在3.1 spring5源码系列--循环依赖之手写代码模拟spring循环依赖中手写了循环依赖的实现. 这个实现就是模拟的spring的循环依赖. 目的是为了更容易理解spring源码. 下面我们就进入正题, 看看spring的循环依赖源码. 一、getBean整体流程目标很明确了, 就是要看看spring如何解决循环依赖的. 代码入口是refresh()#finishBeanFactoryInitialization(beanFactory); 二、拆解研究流程中的每一步调用方法beanFactory.preI...

Mybatis执行流程源码分析【代码】【图】

第一部分：项目结构 user_info表：只有id和username两个字段 User实体类: public class User {private String username;private String password;public String getUsername() {return username;}public void setUsername(String username) {this.username = username;}public String getPassword() {return password;}public void setPassword(String password) {this.password = password;} }mapper:UserMapper 为根据id查询用户...

ArrayList源码分析--jdk1.8【代码】【图】

ArrayList概述??1. ArrayList是可以动态扩容和动态删除冗余容量的索引序列，基于数组实现的集合。??2. ArrayList支持随机访问、克隆、序列化，元素有序且可以重复。??3. ArrayList初始默认长度10，使用Object[]存储各种数据类型。ArrayList数据结构??数据结构是集合的精华所在，数据结构往往也限制了集合的作用和侧重点，了解各种数据结构是我们分析源码的必经之路。??ArrayList的数据结构如下：ArrayList源码分析 /** 用数组实现的...

SparkRdd实现单词统计源码分析【代码】

SparkRdd实现单词统计源码分析 1 手写单词统计 //设置任务名字 local本地模式 val conf=new SparkConf().setAppName("WC").setMaster("local") //通向spark集群的入口 val sc =new SparkContext(conf) // sc.textFile(args(0)).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).sortBy(_._2,false).saveAsTextFile(args(1)) 2 本地Debug调试信息 3 本地debug调试 (1) MapPartitionsRDD[7] at sortBy at SparkWordCou...

[Abp vNext 源码分析] - 5. DDD 的领域层支持(仓储、实体、值对象)【图】

一、简要介绍 ABP vNext 框架本身就是围绕着 DDD 理念进行设计的，所以在 DDD 里面我们能够见到的实体、仓储、值对象、领域服务，ABP vNext 框架都为我们进行了实现，这些基础设施都存放在 Volo.Abp.Ddd.Domain 项目当中。本篇文章将会侧重于理论讲解，但也只是一个抛砖引玉的作用，关于 DDD 相关的知识可以阅读 Eric Evans 所编写的《领域驱动设计：软件核心复杂性应对之道》。PS：该书也是目前我正在阅读的 DDD 理论书籍，因为...

Dubbo源码分析之处理请求（上）

提供端接收请求class NettyServerHandler extends ChannelDuplexHandler public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);try {handler.received(channel, msg);} finally {NettyChannel.removeChannelIfDisconnected(ctx.channel());} } 获取或者创建新的netty包装通道，后续发送响应需要使用网络通道stat...

Django REST framework认证权限和限制源码分析【图】

1.首先我们进入这个initial（）里面看下他内部是怎么实现的。 2.我们进入里面看到他实现了3个方法，一个认证，权限频率 3.我们首先看下认证组件发生了什么权限：啥都没返回，self.permission 条件不执行了。 4.频率组件小总结一下继续：

首页 / 更多教程 / SparkRPC源码分析之RPC管道与消息类型

SparkRPC源码分析之RPC管道与消息类型

内容导读

内容图文

内容总结

内容备注

内容手机端

【SparkRPC源码分析之RPC管道与消息类型】教程文章相关的互联网学习教程文章

飞鸽传书源码分析四-消息发送【代码】【图】

精尽MyBatis源码分析 - MyBatis初始化（四）之 SQL 初始化（下）

xxl-job源码分析【代码】【图】

Dubbo 源码分析 - 集群容错之 Router【代码】

Mybaits源码分析九之sql执行流程【代码】

spring boot启动源码分析 afterRefresh【代码】

一、dubbo源码分析【图】

ForkJoinPool 源码分析

3.2spring源码系列----循环依赖源码分析【代码】【图】

Mybatis执行流程源码分析【代码】【图】

ArrayList源码分析--jdk1.8【代码】【图】

SparkRdd实现单词统计源码分析【代码】

[Abp vNext 源码分析] - 5. DDD 的领域层支持(仓储、实体、值对象)【图】

Dubbo源码分析之处理请求（上）

Django REST framework认证权限和限制源码分析【图】

更多教程 - 最新教程

更多教程 - 最热教程