当前位置 : 主页 > 编程语言 > 其它开发 >

使用Go搭建并行排序处理管道笔记

来源:互联网 收集:自由互联 发布时间:2022-05-30
一、并行管道搭建: 总结下实现思路: 归并排序:进行集合元素排序(节点),并两两节点归并排序;每个节点元素要求有序的(排序),当然终点最小节点元数个数为1必是有序的;
一、并行管道搭建:

总结下实现思路:

  1. 归并排序:进行集合元素排序(节点),并两两节点归并排序;每个节点元素要求有序的(排序),当然终点最小节点元数个数为1必是有序的;
  2. 节点:任务处理单元,归并排序节点是处理输出有序集合任务的单元;文件过大单台机排不了需要多台机集群;
  3. 根据粒度,单机版:单任务版每个节点可以是排序方法,并发版每个节点可以是一个线程/协程去处理(异步排序),集群版节点是一个主机;
  4. 单机版,不管并发还是非并发,节点采用的是内存共享数据;集群版节点则需要网络连接请求应答来共享数据;
  5. go语言异步数据传输通道通过channel实现的;
  6. 每个节点将处理的数据异步发送到各自channel中,等待一个主节点获取归并,集群版多了网络的数据传输(在并发版加上网络的接口)。

 

二、代码实现:
  1. 本地节点 nodes.go:
    package pipeline
    
    import (
    	"encoding/binary"
    	"fmt"
    	"io"
    	"math/rand"
    	"sort"
    	"time"
    )
    
    var startTime time.Time
    
    func Init() {
    	startTime = time.Now()
    }
    
    //内部处理方法
    //这里是排序:异步处理容器元素排序
    func InMemSort(in <-chan int) <-chan int {
    	out := make(chan int, 1024)
    	go func() {
    		a := []int{}
    		for v := range in {
    			a = append(a, v)
    		}
    		fmt.Println("Read done:", time.Since(startTime))
    
    		sort.Ints(a)
    		fmt.Println("InMemSort done:", time.Since(startTime))
    
    		for _, v := range a {
    			out <- v
    		}
    		close(out)
    	}()
    	return out
    }
    
    //两路和并,每路通过内部方法异步处理
    //这里是排序:in1,in2元素需要排好序(经过内部方法InMemSort异步处理)的容器单元(channel 异步容器/队列)
    func Merge(in1, in2 <-chan int) <-chan int {
    	out := make(chan int, 1024)
    	// go func() {
    	// 	v1, ok1 := <-in1
    	// 	v2, ok2 := <-in2
    	// 	for {
    	// 		if ok1 || ok2 {
    	// 			if !ok2 || (ok1 && v1 <= v2) { //v2无值或v1值比v2大
    	// 				out <- v1
    	// 				v1, ok1 = <-in1
    	// 			} else {
    	// 				out <- v2
    	// 				v2, ok2 = <-in2
    	// 			}
    	// 		} else {
    	// 			close(out)
    	// 			break
    	// 		}
    	// 	}
    	// }()
    	go func() {
    		v1, ok1 := <-in1
    		v2, ok2 := <-in2
    		for ok1 || ok2 {
    			if !ok2 || (ok1 && v1 <= v2) { //v2无值或v1值比v2大
    				out <- v1
    				v1, ok1 = <-in1
    			} else {
    				out <- v2
    				v2, ok2 = <-in2
    			}
    		}
    		close(out)
    
    		fmt.Println("Merge done:", time.Since(startTime))
    	}()
    	return out
    }
    
    //读取原数据
    //chunkSize=-1全读
    func ReadSource(r io.Reader, chunkSize int) <-chan int {
    	out := make(chan int, 1024)
    	go func() {
    		buffer := make([]byte, 8) //int长度根据操作系统来的,64位为int64,64位8个字节
    		bytesRead := 0
    		for { //持续读取
    			n, err := r.Read(buffer) //读取一个int 8byte
    			bytesRead += n
    			if n > 0 {
    				out <- int(binary.BigEndian.Uint64(buffer)) //字节数组转int
    			}
    			if err != nil || (chunkSize != -1 && bytesRead >= chunkSize) { //-1全读
    				break
    			}
    		}
    		close(out)
    	}()
    	return out
    }
    
    //写处理后(排序)数据
    func WriteSink(w io.Writer, in <-chan int) {
    	for v := range in {
    		buffer := make([]byte, 8)
    		binary.BigEndian.PutUint64(buffer, uint64(v))
    		w.Write(buffer)
    	}
    }
    
    //随机生成数据源
    func RandomSource(count int) <-chan int {
    	out := make(chan int)
    	go func() {
    		for i := 0; i < count; i++ {
    			out <- rand.Int()
    		}
    		close(out)
    	}()
    	return out
    }
    
    //多路两两归并,每路通过内部方法异步处理
    //这里是排序:ins元素需要排好序(经过内部方法InMemSort异步处理)的容器单元(channel 异步容器/队列)
    func MergeN(ins ...<-chan int) <-chan int {
    	if len(ins) == 1 {
    		return ins[0]
    	}
    	m := len(ins) / 2
    	return Merge(
    		MergeN(ins[:m]...),
    		MergeN(ins[m:]...)) //chennel异步并发归并
    }
    

      

  2. 网络节点:
    package pipeline
    
    import (
    	"bufio"
    	"net"
    )
    
    //节点服务端数据写入到Network中
    //开启服务后,用goroutine等连接,避免创建pipeline阻塞
    func NetworkSink(addr string, in <-chan int) {
    	//net必须是面向流的网络:"tcp"、"tcp4"、"tcp6"、"unix"或"unixpacket"
    	listener, err := net.Listen("tcp", addr) //addr ip:port
    	if err != nil {
    		panic(err)
    	}
    	go func() { //不能等待阻塞
    		for {
    			conn, err := listener.Accept()
    			if err != nil {
    				continue
    			}
    			w := bufio.NewWriter(conn)
    			WriteSink(w, in)
    			w.Flush()    //使用bufio Writer最后一定要Flush把缓存数据发出去  defer
    			conn.Close() //关闭
    		}
    		// defer listener.Close()
    		// conn, err := listener.Accept()
    		// if err != nil {
    		// 	panic(err)
    		// }
    		// defer conn.Close()
    		// w := bufio.NewWriter(conn)
    		// WriteSink(w, in)
    		// defer w.Flush()
    	}()
    }
    
    //Network向节点服务端读取数据源
    func NetworkSource(addr string) <-chan int {
    	out := make(chan int)
    	go func() {
    		conn, err := net.Dial("tcp", addr)
    		if err != nil {
    			panic(err)
    		}
    		defer conn.Close()
    
    		r := ReadSource(bufio.NewReader(conn), -1)
    		for v := range r {
    			out <- v
    		}
    		close(out)
    	}()
    	return out
    }
    

      

  3. 创建管道:
    package main
    
    import (
    	"bufio"
    	"fmt"
    	"goBase/pipelinedemo/pipeline"
    	"os"
    	"strconv"
    )
    
    const sourceFilename = "../large.in"
    const resultFilename = "../large.out"
    
    //单机版而言,并发使用channel效率肯定是下降的
    //好处,当文件过大,一台机器排不了,多机排序
    func main() {
    	p, files := createNetworkPipeline(sourceFilename, 800000000, 4) //平均每个文件读取int64数:800000000/8/4
    	defer func() {
    		for _, file := range files {
    			file.Close()
    		}
    	}()
    	writeToFile(p, resultFilename) //该方法运行,通道才真正打开
    	printFile(resultFilename)
    }
    
    
    //创建并行处理管道
    //fileSize 文件字节数
    //chunkCount 节点数 读取文件分块数
    func createNetworkPipeline(filename string, fileSize, chunkCount int) (<-chan int, []*os.File) {
    
    	chunkSize := fileSize / chunkCount //每个节点读取文件字节数
    
    	//outs := make([]<-chan int, chunkCount)
    	outs := []<-chan int{}
    	sortAddr := []string{}
    
    	files := []*os.File{}
    
    	pipeline.Init() //开始计时
    
    	//#region 节点服务端工作
    
    	for count := 0; count < chunkCount; count++ {
    		file, err := os.Open(filename) //这里file没有close,需要返回*[]File,在外面close
    		if err != nil {
    			panic(err)
    		}
    		files = append(files, file)
    
    		//Seek设置下一次读/写的位置。offset为相对偏移量,
    		//whence决定相对位置:0为相对文件开头,1为相对当前位置,2为相对文件结尾
    		file.Seek(int64(count*chunkSize), 0) //读文件字节范围
    
    		source := pipeline.ReadSource(bufio.NewReader(file), chunkSize)
    
    		// outs = append(outs, pipeline.InMemSort(source))
    		//本机地址
    		addr := ":" + strconv.Itoa(7000+count)                 //将数字转换成对应的字符串类型的数字
    		pipeline.NetworkSink(addr, pipeline.InMemSort(source)) //开启节点服务监听,收到请求发送数据将写入到Network,异步不能等待阻塞
    
    		sortAddr = append(sortAddr, addr)
    	}
    
    	//#endregion
    
    
    	//#region Network工作
    
    	for _, addr := range sortAddr {
    		outs = append(outs, pipeline.NetworkSource(addr))
    	}
    	//构建管道,goroutine还没有运行,不能确定InMemSort是否全部排序完成,不能在该方法close file
    	return pipeline.MergeN(outs...), files
    
    	//#endregion
    }
    
    func writeToFile(in <-chan int, filename string) {
    	file, err := os.Create(filename)
    	if err != nil {
    		panic(err)
    	}
    	defer file.Close()
    
    	w := bufio.NewWriter(file)
    	defer w.Flush()
    
    	pipeline.WriteSink(w, in)
    }
    
    func printFile(filename string) {
    	file, err := os.Open(filename)
    	if err != nil {
    		panic(err)
    	}
    	defer file.Close()
    	count := 0
    	all := pipeline.ReadSource(bufio.NewReader(file), -1)
    	for s := range all {
    		fmt.Println(s)
    		count++
    		if count > 100 {
    			break
    		}
    	}
    }
上一篇:python操作MySQL;MySQL补充知识
下一篇:没有了
网友评论