Scala-21：分布式计算模拟

╰半夏微凉° 2023-10-04 16:55 3阅读 0赞

## Scala-21：分布式计算模拟 ##

分布式计算，顾名思义，就是把需要处理的数据进行分布处理，这样可以减轻每一台计算机的计算量，从而提高效率

现在就模拟一个分布式计算的简单流程

#### 案例一：一对一传输数字 ####

**假定一个简单的模拟，一台服务器和一台客户端进行一对一的传输数据进行处理，数据使用Int**

首先选择简单的Socket模仿简单的网络编程

**服务器端：**

分析：

1.  首先需要指定一个端口接受数据
2.  然后等待客户端连接，未连接就是阻塞状态
3.  实例一个接受数据的管道，既是输入流
4.  接受数据
5.  最后关闭资源

**客户端：**

分析

1.  首先创建一个客户端连接服务器
2.  然后创建一个输出流用于写数据
3.  写出数据
4.  最后关闭资源

//连接服务器
    val clinet = new Socket("localhost", 9999)    //连接本地的9999端口
    
    //创建输出流
    val out: OutputStream = client.getOutputStream
    
    //写出数据
    out.write(2)
    out.flush()
    
    //资源关闭
    out.close()
    client.close()

先开启服务端，在执行客户端，客户端之行结束，就代表客户端的数据已经传输到了服务端，就可以在服务端查看到传输的数据了

#### 案例二：一对一传输对象类型数据 ####

**在实际的应用开发中，大多数传输的都是一个一个的对象，所以简单的模拟一个对象的传输，并且在对象内部做一些运算**

分析：首先就需要先创建一个类，类里面简单的设置一个列表用于计算，以及在类中声明一个方法用于计算

**Task类**

//因为传输对象，所以要混入序列化的特质
    class Task Extends Serializable {
        
    	
    	val datas: List[Int] = List(1, 2, 3, 4)
    	
    	//一个创建逻辑，实现简单的乘2操作
    	val logic: Int => Int = _ * 2
    	
    	//计算方法
    	def compute(): List[Int] = {
        
    		datas.map(logic)
    	}
    }

**服务器端：**

分析：

这里和上述的就有一些特别的地方，因为传输的是一个对象，所以简单的输入流是不可以的，需要使用到对象流

1.  首先需要指定一个端口接受数据
2.  然后等待客户端连接，未连接就是阻塞状态
3.  实例一个接受数据的管道，既是对象流
4.  接受对象数据
5.  最后关闭资源

//启动服务器
    val server = new ServerSocket(9999)  //端口号9999
    prinln("服务器启动，等待接收数据")
    
    //等待客户端连接
    val client: Socket = server.accept()
    
    //创建一个输入流接受数据
    val in: InputStream = client.getInputStream
    //使用输入流创建对象流
    val objIn: ObjectInputStream = new ObjectInputStream(in)
    
    //实例化一个Task对象，用于传输并且计算
    val task: Task = objIn.readObject().asInstanceOf[Task]
    val ints: List[Int] = task.compute()
    
    //接收到的数据打印
    piintln("计算之后的数据是：" + ints)
    
    //资源关闭
    objIn.close()
    client.close()
    server.close()

**客户端：**

分析：

同样的，因为传输的是对象数据，所以在使用输出流的时候也要换为对象流，然后在写出数据的时候使用对象流写出

1.  首先创建一个客户端连接服务器
2.  然后创建一个对象流用于写数据
3.  写出对象数据
4.  最后关闭资源

//连接服务器
    val clinet = new Socket("localhost", 9999)    //连接本地的9999端口
    
    //创建输出流
    val out: OutputStream = client.getOutputStream
    //使用输出流创建对象流
    val objOut: ObjectOuputStream = new ObjectOutputStream(out)
    
    //创建一个Task对象
    val task: Task = new Task()
    
    //写出Task对象数据
    objOut.writeObject(task)
    objOut.flush()
    
    //资源关闭
    objOut.close()
    objOut.close()

#### 案例三：一对多传输对象类型数据 ####

以上的两个案例简单的模拟了分布式下计算机之间如何进行的传输数据，但是都是一对一的模式，在实际的开发环境下，并非是一对一的情况，大多数都是一对多的情况，比如：一个客户端发送来的数据，多个服务器共同进行处理，这样才是真正的分布式处理，可以提高效率，也可以减少单台机器的运行负荷

简单模拟一个客户端发送的数据，使用两个服务器去处理，以上一个案例为基础，对传来的对象数据，分别使用两台服务器去处理

**对象类**

在对象类的设计中，因为是要把传来的对象分散到两台机器上处理，所以还需要有一个用于分割对象的类，这个类的结构和Task类应该一致，但是因为要分发到两台机器上的数据不确定，所以在这个类中的属性和逻辑的实现上都声明成可变的

class SubTask extends Serializable {
        
    	//数据
    	val datas: List[Int] = _
    	
    	//逻辑
    	val logic: Int => Int = _
    	
    	//方法不变
    	def compute(): List[Int] = {
        
    		datas.map(logic)
    	}
    }

在传输的时候使用Task类，这个类就用于我们服务器在接收到传输过来的Task对象时用于怎么分发的操作

**服务器端：**

分析：

因为是要传输的数据分发到两台服务器上进行处理，所以就需要两台机器来运行处理，就要在上述案例的基础上，在多加一台不同的端口号进行处理

然后就要考虑怎么分发的操作，其实这个也比较简单，因为在上个步骤已经创建了SubTask进行处理，所以就可以将我们的分发逻辑处理之后将值赋给SubTask对象即可

最后剩下的步骤就和上述的案例一样了

1.  首先需要两个服务器，指定两个个端口接受数据
2.  然后等待客户端连接，未连接就是阻塞状态
3.  实例一个接受数据的管道，既是对象流
4.  接受对象数据
5.  最后关闭资源

**服务器1：**

//启动服务器
    val server = new ServerSocket(9999)  //端口号9999
    prinln("服务器启动，等待接收数据")
    
    //等待客户端连接
    val client: Socket = server.accept()
    
    //创建一个输入流接受数据
    val in: InputStream = client.getInputStream
    //使用输入流创建对象流
    val objIn: ObjectInputStream = new ObjectInputStream(in)
    
    //实例化一个Task对象，用于传输并且计算
    val task: Task = objIn.readObject().asInstanceOf[Task]
    val ints: List[Int] = task.compute()
    
    //接收到的数据打印
    piintln("计算节点[9999]计算之后的的数据是：" + ints)
    
    //资源关闭
    objIn.close()
    client.close()
    server.close()

**服务器2：**

//启动服务器
    val server = new ServerSocket(8888)  //端口号8888
    prinln("服务器启动，等待接收数据")
    
    //等待客户端连接
    val client: Socket = server.accept()
    
    //创建一个输入流接受数据
    val in: InputStream = client.getInputStream
    //使用输入流创建对象流
    val objIn: ObjectInputStream = new ObjectInputStream(in)
    
    //实例化一个Task对象，用于传输并且计算
    val task: Task = objIn.readObject().asInstanceOf[Task]
    val ints: List[Int] = task.compute()
    
    //接收到的数据打印
    piintln("计算节点[8888]计算之后的的数据是：" + ints)
    
    //资源关闭
    objIn.close()
    client.close()
    server.close()

**客户端：**

因为要涉及到将传输的数据分成两份发往不同的服务器处理，所以在这里就要考虑我们的分发数据原则，将数据分发之后交由不同的服务器处理

1.  首先创建一个客户端连接服务器
2.  然后创建一个对象流用于写数据
3.  对于传出对象进行分割成两份，分别 传出到不同的端口号中
4.  写出对象数据
5.  最后关闭资源

//连接服务器，就需要连接两个端口号
    val clinet1 = new Socket("localhost", 9999)    //连接本地的9999端口
    val clinet2 = new Socket("localhost", 8888)    //连接本地的8888端口
    
    //创建两个输出流
    val out1: OutputStream = client.getOutputStream
    val out2: OutputStream = client.getOutputStream
    //使用输出流创建对象流
    val objOut1: ObjectOuputStream = new ObjectOutputStream(out1)
    val objOut2: ObjectOuputStream = new ObjectOutputStream(out2)
    
    //创建一个Task对象,是需要传输的对象
    val task: Task = new Task()
    
    //发送至9999端口
    val sub1: SubTask = new SubTask()
    //task有四条数据，将前两条发送至9999端口
    sub1.logic = task.logic
    sub1.datas = task.datas.take(2)
    
    //写出sub1对象数据到9999端口
    objOut1.writeObject(sub1)
    objOut1.flush()
    
    //资源关闭
    objOut1.close()
    objOut1.close()
    println("发送至9999端口的数据发送完毕")
    
    
    //发送至8888端口
    val sub2: SubTask = new SubTask()
    //task有四条数据，将后两条发送至8888端口
    sub2.logic = task.logic
    sub2.datas = task.datas.right(2)
    
    //写出sub1对象数据到9999端口
    objOut2.writeObject(sub2)
    objOut2.flush()
    
    //资源关闭
    objOut2.close()
    objOut2.close()
    println("发送至8888端口的数据发送完毕")

这样三个案例就简单的模拟了分布式计算的简单情况，如果要复杂一点的话，就是在我们传输的数据类型属性和逻辑的复杂，万变不离其宗，从简到难，可以慢慢实现。