Spark 源码解析(二) 根据 SparkRpc 自己动手实践一个跨节点通讯 ...

 目录
一、框架流程：
二、Maven 搭建 Scala 导入POM依赖
三、根据流程举行编写
1、实例 Master
2、创建 RpcEnv
3、创建 RpcEndpoint
4、生成 RpcEndpointRef
5、RpcEndpointRef发送消息
 6、防止还没收到消息程序就结束运行
7、验证一下，看看结果
四、完整代码

---

一、框架流程：

1、实例 Master
2、创建 RpcEnv
3、Master向 RpcEnv 注册
4、生成 RpcEndpointRef
5、RpcEndpointRef发送消息
二、Maven 搭建 Scala 导入POM依赖

起首是主模块

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0"
3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5.     <modelVersion>4.0.0</modelVersion>
7.     <groupId>org.apache.spark</groupId>
8.     <artifactId>10</artifactId>
9.     <version>1.0-SNAPSHOT</version>
10.     <packaging>pom</packaging>
11.     <modules>
12.         <module>11</module>
13.     </modules>
15.     <properties>
16.         <maven.compiler.source>8</maven.compiler.source>
17.         <maven.compiler.target>8</maven.compiler.target>
18.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
19.     </properties>
21.     <build>
22.         <pluginManagement>
23.             <plugins>
24.                 <plugin>
25.                     <groupId>net.alchim31.maven</groupId>
26.                     <artifactId>scala-maven-plugin</artifactId>
27.                     <version>3.2.2</version>
28.                 </plugin>
29.                 <plugin>
30.                     <groupId>org.apache.maven.plugins</groupId>
31.                     <artifactId>maven-site-plugin</artifactId>
32.                     <version>3.5.1</version>
33.                 </plugin>
34.             </plugins>
35.         </pluginManagement>
36.         <plugins>
37.             <plugin>
38.                 <groupId>net.alchim31.maven</groupId>
39.                 <artifactId>scala-maven-plugin</artifactId>
40.                 <executions>
41.                     <execution>
42.                         <id>scala-compile-first</id>
43.                         <phase>process-resources</phase>
44.                         <goals>
45.                             <goal>add-source</goal>
46.                             <goal>compile</goal>
47.                         </goals>
48.                     </execution>
49.                     <execution>
50.                         <id>scala-test-compile</id>
51.                         <phase>process-test-resources</phase>
52.                         <goals>
53.                             <goal>testCompile</goal>
54.                         </goals>
55.                     </execution>
56.                 </executions>
57.             </plugin>
59.             <plugin>
60.                 <groupId>org.apache.maven.plugins</groupId>
61.                 <artifactId>maven-compiler-plugin</artifactId>
62.                 <executions>
63.                     <execution>
64.                         <phase>compile</phase>
65.                         <goals>
66.                             <goal>compile</goal>
67.                         </goals>
68.                     </execution>
69.                 </executions>
70.             </plugin>
72.             <plugin>
73.                 <groupId>org.apache.maven.plugins</groupId>
74.                 <artifactId>maven-shade-plugin</artifactId>
75.                 <version>2.4.3</version>
76.                 <executions>
77.                     <execution>
78.                         <phase>package</phase>
79.                         <goals>
80.                             <goal>shade</goal>
81.                         </goals>
82.                         <configuration>
83.                             <filters>
84.                                 <filter>
85.                                     <artifact>*:*</artifact>
86.                                     <excludes>
87.                                         <exclude>META-INF/*.SF</exclude>
88.                                         <exclude>META-INF/*.DSA</exclude>
89.                                         <exclude>META-INF/*.RSA</exclude>
90.                                     </excludes>
91.                                 </filter>
92.                             </filters>
93.                         </configuration>
94.                     </execution>
95.                 </executions>
96.             </plugin>
97.         </plugins>
98.     </build>
99. </project>

复制代码

然后创建一个子模块

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0"
3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5.     <modelVersion>4.0.0</modelVersion>
6.     <parent>
7.         <groupId>org.apache.spark</groupId>
8.         <artifactId>10</artifactId>
9.         <version>1.0-SNAPSHOT</version>
10.     </parent>
12.     <artifactId>11</artifactId>
14.     <properties>
15.         <maven.compiler.source>8</maven.compiler.source>
16.         <maven.compiler.target>8</maven.compiler.target>
17.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
18.     </properties>
20.     <dependencies>
21.         <dependency>
22.             <groupId>org.arakhne.afc.slf4j</groupId>
23.             <artifactId>slf4j-log4j</artifactId>
24.             <version>17.0</version>
25.         </dependency>
26.         <dependency>
27.             <groupId>org.scala-lang</groupId>
28.             <artifactId>scala-library</artifactId>
29.             <version>2.13.15</version>
30.         </dependency>
31.         <dependency>
32.             <groupId>org.apache.spark</groupId>
33.             <artifactId>spark-core_2.13</artifactId>
34.             <version>3.5.4</version>
35.         </dependency>
36.     </dependencies>
38. </project>

复制代码

三、根据流程举行编写

在子模块下创建Scala类Master
1、实例 Master

1. class Master(val rpcEnv:RpcEnv) extends ThreadSafeRpcEndpoint {
3.   override def receive: PartialFunction[Any, Unit] = {
4.     case "hello" => println("hello，这是一个测试")
5.     case _ => println("receive uncatchable information!")
6.   }
7. }

复制代码

我们起首必要伪装一下包名，否则无法调用 RpcEnv
然后继承线程安全的 ThreadSafeRpcEndpoint 成为一个Endpoint
重写receive方法，用到了偏函数。
2、创建 RpcEnv

1.     val sparkConf:SparkConf = new SparkConf()
2.     val securityManager:SecurityManager = new SecurityManager(sparkConf)
4.     // 1、创建RpcEnv
6.     val rpcEnv1 = RpcEnv.create("master","127.0.0.1",10000,sparkConf,securityManager)

复制代码

3、创建 RpcEndpoint

1.     // 2、创建RpcEndpoint
3.     val master = new Master(rpcEnv1)

复制代码

Master 继承了 ThreadSafeRpcEndpoint 以是可以直接这么写。

4、生成 RpcEndpointRef

1.     // 3、向RpcEnv注册RpcEndpoint并返回RpcEndpointRef
3.     val endpointEndpointRef = rpcEnv1.setupEndpoint("endpoint1", master)

复制代码

代码现实上很简单，由于都集成好了
5、RpcEndpointRef发送消息

1.     // 4、使用RpcEndpointRef发送消息
3.     endpointEndpointRef.send("hello")

复制代码

 6、防止还没收到消息程序就结束运行

1.     rpcEnv.awaitTermination()

复制代码

7、验证一下，看看结果

乐成！
再试试传输别的信息

1. endpointEndpointRef.send("123456")

复制代码

 没有问题。
四、完整代码

1. package org.apache.spark.psy
3. import org.apache.spark.SparkConf
4. import org.apache.spark.rpc.{RpcEnv, ThreadSafeRpcEndpoint}
5. import org.apache.spark.SecurityManager
8. class Master(val rpcEnv: RpcEnv) extends ThreadSafeRpcEndpoint{
9.   override def receive: PartialFunction[Any, Unit] = {
10.     case "hello" => println("hello，这是一个测试")
11.     case _ => println("receive uncatchable information!")
12.   }
13. }
15. object Master{
16.   def main(args: Array[String]): Unit = {
18.     val conf = new SparkConf()
19.     val securityManager = new SecurityManager(conf)
21.     //1.创建RpcEnv
22.     val rpcEnv = RpcEnv.create("SparkMaster","localhost",8888,conf,securityManager)
23.     //2.创建RpcEndPoint
24.     val master = new Master(rpcEnv)
25.     //3.使用RpcEnv注册RpcEndPoint
26.     val masterEndpointRef = rpcEnv.setupEndpoint("Master",master)
28.     //4.通过返回的RpcEndPointRef发送消息
29.     masterEndpointRef.send("123456")
31.     rpcEnv.awaitTermination()
32.   }
33. }

复制代码

OK，代码不是很难，由于都已经集成好了，接下来我们要试着自己去集成代码，自己尝试完成一个 应用 Rpc 框架
加油！坚持就是胜利！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。