package pipeline

func ArraySource(arr ...int) chan int {
    out := make(chan )
    go func() {
        for a := range arr  {
            out <- a
        }

        close(out)
    }()
    return out
}

test007/pipeline
)

func main() {
    sourceChan := pipeline.ArraySource(4,1)">8,1)">19,1)">5)

     {
        if data,ok := <- sourceChan; ok {
            fmt.Println(data:else {
            break
        }
    }

    time.Sleep(time.Second * 10)
}

)
    for v := range sourceChan {
            fmt.Println()
}

)
    go func() {
        for v := range sourceChan {
            fmt.Println()
}

package pipeline

import 

func ArraySource(arr ...int) <- chan int a
        }

        time.Sleep(time.Second * 2)
        out <- 9
        //close(out)

}

 比如: 老师要进行一个sort排序
func Sort(in <-chan int) <- chan {
     第一步: 定义一个chan
    )
    
     第三步: 向管道里放数据,单独开一个goroutine
    go func() {
        
    }()
    
     第二步: return 这个channel
    
}

func InMemSort(int) <-chan  第一步: 定义一个channel变量
     第三步: 向channel中放数据
    go func() {
        var arr []int
         管道到这里会阻塞,等待close以后,才会退出这个for循环
         如果没有close,就会发生死锁
        for v := range in  {
            arr = append(arr,v)
        }
        fmt.Println(arr: for _,1)"> range arr {
             v
        }
    }()

     第二步: 返回这个channel
    
}

func Merge(in1,in2 <-chan  第一步: 定义一个管道变量
     第三步: 向管道中放入数据
 第四步: 从两个管道中取一个数据
        v1,ok1 := <- in1
        v2,ok2 := <- in2

         第五步: 如果能够从任意一个管道中取出数据,则处理
        for ok1 || ok2 {
            if !ok2 || (ok1 && v1 <= v2) {
                 v1
                v1,ok1 = <- in1
            }  {
                 v2
                v2,ok2 = <-in2
            }
        }

         第六步: 没有数据可以取出来,close管道,表示已经取完了
            }()

     第二步: 返回这个管道
    
}

package main

import (
"

)

func main() {
    mergeChan := pipeline.Merge(
        pipeline.InMemSort(pipeline.ArraySource(1,4,8,2,19,5)),pipeline.InMemSort(pipeline.ArraySource(0,29,43,1,7,9)))

    go func() {
         range mergeChan {
            fmt.Println()
}

GOROOT=/usr/local/go #gosetup
GOPATH=/Users/luoxiaoli/go #gosetup
/usr/local/go/bin/go build -o /private/var/folders/g2/74np978j3971l2864zdk7lgc0000gn/T/___go_build_main_go_darwin /Users/luoxiaoli/test007/pipeline/pipelinedemo/main.go #gosetup
/var/folders/g2/74np978j3971l2864zdk7lgc0000gn/T/___go_build_main_go_darwin #gosetup
arr:  [1 4 8 2 19 ]
arr:  [0 29 43 7 95 199 43]
data: 0
data: 147829

Process finished with exit code 0

func ArraySource(arr ... a
        }
        close()
    }()
    
}

/**
 * 从redader中读取数据
 */
func ReaderSource(reader io.Reader) <- chan  这里定义为8个字节,原因是我的机器是64位的,所以int也是64位,那么对应的字节数就是8个字节
        buffer := make([]byte,1)">   {
             reader返回两个参数,第一个是读取到的字节数,第二个是err异常
            n,err := reader.Read(buffer)
            if n >   {
                 如果读到了,就把读到的东西发给channel
                u := binary.BigEndian.Uint64(buffer)
                out <- (u)
            }

            if err != nil {
                
            }
        }
        close(
}

*
 * 只读数据,不写数据的,将读出来的数据打印出来
 * 可以打印到控制台,也可以写入到文件. 这里就写入到文件
 
func WriteSink(writer io.Writer, {
        b := make([])
        binary.BigEndian.PutUint64(b,uint64(v))
        writer.Write(b)
    }
}

*
 * 生成一个随机数
 
 func RandomSource(count  {
     )
     go func() {
          生成count个随机数
        0; i<count ; i++ rand.Int()
        }

        close()
    }()
     
 }

func main() {
     第一步: 造数据,生成100个随机数,写入到文件
    const fileName  = small.in"
    const count = 100
     第一步: 将随机生产的数字保存到small.in文件
     构造第一个数据源
    file,e := os.Create(fileName)
    if e != nil {
        panic(e)
    }
    defer file.Close()
    dataSource := pipeline.RandomSource(count)
    writer := bufio.NewWriter(file)
    pipeline.WriteSink(writer,dataSource)
    writer.Flush()
    
     第二步: 从文件中读取文件内容,在控制台打印
     从第一个数据源读取出数据
    f,1)"> os.Open(fileName)
     nil {
        panic(e)
    }
    defer f.Close()
    readerSource := pipeline.ReaderSource(bufio.NewReader(f))

    var num = 0
    for rs := range readerSource{
        fmt.Println(rs)
        num ++
        if num > 100
        }
    }
}

*
 * N个节点两两归并
 
 func MergeN(inputs ... <-chan {
     if len(inputs) ==  {
         return inputs[]
     }

     middle := len(inputs) / 2
      两个两个的合并
     return Merge(MergeN(inputs[:middle]...),MergeN(inputs[middle:]...))
 }

*
 * @param fileName: 文件名
 * @param fileSize: 文件大小
 * @param chunkCount: 将文件分成多少块
 
func createPipeline(fileName string,fileSize,chunkCount {
    pipeline.Init()
     每次读取的内容的字节大小
    chunkSize := fileSize/chunkCount
    sortResult := []<-chan {}
    0; i < chunkCount; i++ {
        file,e := os.Open(fileName)
         nil {
            panic(e)
        }
         offset: 从文件的什么位置开始读 whence: 从第几个字符开始读
        file.Seek(int64(chunkSize*i),1)">)

         读取文件的内容
        source := pipeline.ReaderSource(bufio.NewReader(file),chunkSize)

         在内存中对内容进行排序
        sortResult = append(sortResult,pipeline.InMemSort(source))
    }

     合并所有的内部排序后的结果
     pipeline.MergeN(sortResult...)
}

*
 * 将合并后的结果写入到文件
 
func writeToFile(p <- chan ) {

    file,1)"> nil {
        panic(e)
    }
    defer file.Close()

    writer := bufio.NewWriter(file)
    defer writer.Flush()

     这一步是将读取的内容写入到文件
    pipeline.WriteSink(writer,p)

}

func printFile(fileName ) {
    file,1)"> nil {
        panic(e)
    }
    defer file.Close()

    reader := bufio.NewReader(file)
    source := pipeline.ReaderSource(reader,-)
    count :=  range source {
        fmt.Println(v)
        count ++
        if count > 
        }
    }
}

 定义一个节点,处理数据
func InMemSort( 第一步: 定义一个channel变量
    out := make(chan int,1000)

    read data: sorted data:  v
        }
        close()
    }()

    
}


func Merge(in1,in2 <-chan  第一步: 定义一个管道变量
    out := make(chan int,表示已经取完了
        close()
        fmt.Println(merged data: 
}

*
 * 从redader中读取数据
 * 将reader改为分块读取,每次读取指定字符长度
 
func ReaderSource(reader io.Reader,trunkSize {
    out := make(chan int,1000)
    go func() {
        )
        readered := 0
         记录已经读取的个数
             reader.Read(buffer)
            readered += n
            if err != nil || (trunkSize != -1 && readered >= trunkSize){
                
            }
            fmt.Println("已读字符数",readered)
        }
        close(
}