Go 网络版外部排序

目录结构

1、基础文件

var startTime time.Time
func Init(){
    startTime = time.Now()
}

func ArraySource(a ...int) <-chan int{
    out := make(chan int)
    go func(){
        for _,v := range a{
            out <- v
        }
        close(out)
    }()
    return out
}

func InMemSort(in <-chan int) <-chan int {
    out := make(chan int,1024)
    go func() {
        a := []int{}
        for v := range in {
            a = append(a,v)
        }
        //large
        fmt.Println("read done:",time.Now().Sub(startTime))

        sort.Ints(a)
        //large
        fmt.Println("Inmemsort done:",time.Now().Sub(startTime))
        for _,v := range a{
            out <- v
        }
        close(out)
    }()
    return out
}

func Merge(in1,in2 <-chan int) <-chan int {
    out := make(chan int,1024)
    go func() {
        v1,ok1 := <-in1
        v2,ok2 := <-in2
        for ok1 || ok2 {
            if !ok2 ||(ok1 && v1 <= v2){
                out <- v1
                v1,ok1 = <-in1
            }else{
                out <- v2
                v2,ok2 = <-in2
            }
        }
        close(out)
        //large
        fmt.Println("merge done:",time.Now().Sub(startTime))
    }()
    return out
}
//3-5 加参数chunkSize
func ReaderSource(reader io.Reader,chunkSize int) <-chan int{
    out := make(chan int,1024)
    go func() {
        buffer := make([]byte,8)
        //3-5
        bytesRead := 0
        for{
            n,err := reader.Read(buffer)
            //3-5
            bytesRead += n
            if n>0 {
                v := int(binary.BigEndian.Uint64(buffer))
                out <- v
            }
            //3-5
            if err != nil || (chunkSize != -1 && bytesRead >= chunkSize){
                break
            }
        }
        close(out)
    }()
    return out
}

func WirteSink(writer io.Writer,in <-chan int){
    for v:= range in{
        buffer := make([]byte,8)
        binary.BigEndian.PutUint64(buffer,uint64(v))
        writer.Write(buffer)
    }
}

func RandomSource(count int) <-chan int{
    out := make(chan int)
    go func() {
        for i:=0;i<count;i++{
            out <-rand.Int()
        }
        //如果没有这句话会报错 fatal error: all goroutines are asleep - deadlock!
        close(out)
    }()
    return out
}


func MergeN(inputs ...<-chan int) <-chan int{
    if len(inputs) == 1{
        return inputs[0]
    }
    m := len(inputs) / 2
    return Merge(
        MergeN(inputs[:m]...),
        MergeN(inputs[m:]...))
}

main.go创建输入的文件

网络节点net_nodes

externsort

externsort排序穿件了一个createNetworkPipeline方法，这里有两个输出，其中只增加了createNetWorkPipeline其余未修改。

func printFile(filename string){
    file,err := os.Open(filename)
    if err != nil{
        panic(err)
    }
    defer file.Close()
    p := pipeline0807.ReaderSource(file,-1)
    //large 不能一口气输出 要不然要挂掉
    count := 0
    for v := range p{
        fmt.Println(v)
        count++
        if count >= 100{
            break
        }
    }
}

func writeToFile(p <-chan int,filename string){
    file,err := os.Create(filename)
    if err != nil{
        panic(err)
    }
    defer file.Close()
    writer := bufio.NewWriter(file)
    defer writer.Flush() //先flush后close
    //这个时候才拿出来
    pipeline0807.WirteSink(writer,p)
}

createNetWorkPipeline

输出1：demo2

输出2：demo3

由于第一个输出已经使用了端口，如果未及时关闭可能会提示

此时关闭即可