go 搭建并行处理管道
一. Go语言并发编程
- 采用了CSP(Communication Seuential process)模型
- 不需要锁, 不需要callback
- 并发编程 vs 并行计算
1.1 CSP并发模型
CSP模型是上个世纪七十年代提出的，用于描述两个独立的并发实体通过共享的通讯 channel(管道)进行通信的并发模型。 CSP中channel是第一类对象，它不关注发送消息的实体，而关注与发送消息时使用的channel。
1.2 Golang CSP
Golang 就是借用CSP模型的一些概念为之实现并发进行理论支持，其实从实际上出发，go语言并没有，完全实现了CSP模型的所有理论，仅仅是借用了 process和channel这两个概念。process是在go语言上的表现就是 goroutine 是实际并发执行的实体，每个实体之间是通过channel通讯来实现数据共享。
1.3 不需要锁, 不需要callback
go使用CSP模型进行通信, 不需要使用锁, 其实, 这里不需要锁指的是用户在使用go语言进行并发通信的时候不需要使用锁. 也不需要使用callback. 但是, go底层其实还是使用了锁和callback的.
2.1 模拟服务器启动, 打印内容到页面
package main
import (
"fmt"
"net/http"
)
func main() {
http.HandleFunc("/", func(writer http.ResponseWriter, request *http.Request) {
fmt.Fprintf(writer, "<h1>Hello World %s</h1>", request.FormValue("name"))
})
http.ListenAndServe(":8888", nil)
}这里面需要注意的是启动一个服务器的方式
2.2 主方法main和Hello world方法进行通信, 通信使用的是channel
package main
import (
"fmt"
"time"
)
func main() {
ch := make(chan string)
for i := 0; i< 5000; i++ {
go hello(i, ch)
}
for {
str := <-ch
fmt.Println(str)
}
time.Sleep(time.Millisecond)
}
func hello(i int, ch chan string) {
for {
ch <- fmt.Sprintf("hello world, %d", i)
}
}2.3 内部排序
func main() {
arr := []int {2, 5, 1, 9, 23, 01}
sort.Ints(arr)
for i, v := range arr {
fmt.Printf("%d, %d\n", i, v)
}
}内部排序使用的是内建函数
2.4 go实现外部排序 pipeline
我们使用外部排序的时候,会用到归并排序, 先来看看什么是归并排序?
将数据分为左右两半, 分别归并排序, 再把两个有序数据归并
有一个大的集合, 首先将其分为两个小的集合, 使用内部排序对两个小的集合进行内部排序
得到排好序的两个集合, 然后进行归并排序
第一步: 取出两个集合的首元素. 比较, 如果一样, 取左侧元素---->1
第二步: 再次取出两个集合的首元素, 比较, 右侧小--->1
第三步: 再次取出两个集合的首元素, 比较, 左侧小--->2
........
依次类推
接下来,我们的外部排序, 采用二路归并的方式实现.
数据源,来自多方. 比如hadoop中的多个hdfs
然后将数据发送到节点1, 进行二路归并, 归并后的结果在发送到节点2, 再次进行二路归并.......一次类推,直到最后只有一路数据,就是我们要的结果
3. 代码实现
我们使用案例来说明管道的使用
3.1 channel是goroutine和goroutine之间的通信
首先, 将数据放入管道中.....这里有个疑问, 为什么要将数据放入管道中呢?
假如: 这每一个数组都是一个对象, 一个很大的对象, 处理链路比较长. 这时候, 放入管道中. 就可以并发处理了. 不影响后面的流程. 该怎么处理, 就怎么处理.
package pipeline
func ArraySource(arr ...int) chan int {
out := make(chan int)
go func() {
for a := range arr {
out <- a
}
close(out)
}()
return out
}这是一个将数组放入管道的过程. 有两点
- 创建一个channel. 然后将这个channel return
- 记住: channel是goroutine和goroutine之间的通信. 我们在给channel赋值的时候, 不能直接赋值, 要将其放在一个单独的goroutine中.
- 注意: 这里有一个close, 意思是:我把数据向管道中放完了, 没有数据了. 后面取数据的人, 请不要再取了
所以, 这里定义了一个单独的goroutine,用来向管道中存数据
接下来写一个demo, 取数据
package main
import (
"fmt"
"test007/pipeline"
"time"
)
func main() {
sourceChan := pipeline.ArraySource(1, 4, 8, 2, 19, 5)
for {
if data, ok := <- sourceChan; ok {
fmt.Println("data:", data)
} else {
break
}
}
time.Sleep(time.Second * 10)
}首先, 我将一个数组, 放入管道中.
然后, 从管道中持续不断的去取数据. 所以有一个for死循环. 循环的时候, 使用了ok来判断, 是否还有数据可以取出, 如果没有了, 就退出
注意: 这里使用了close关闭了管道, 所以, 我们就不会不停的取数据.
还有一种方式, 直接有数据取出, 不用我们手动判断
package main
import (
"fmt"
"test007/pipeline"
"time"
)
func main() {
sourceChan := pipeline.ArraySource(1, 4, 8, 2, 19, 5)
for v := range sourceChan {
fmt.Println("data:", v)
}
time.Sleep(time.Second * 10)
}如果使用了range的方式, 从管道中取数据, 那么....必须要手动close, 否则, for循环不知道该何时退出, 将会发生死锁的现象.
为什么死锁呢?
因为, 这里的for循环不知道何时退出, 一直处于等待状态, 后面的代码没有办法执行, 所以就发生了死锁.
那么: 通常情况下, 我们的管道是不会手动close的. 当管道中有数据的时候, 我们就取出, 如果管道中没有数据, 我们就等待. 这个怎么做呢?
也就是说, 我们不会手动close, 如何才能让goroutine一直等待, 直到有数据到来呢? 接收方也使用单独的goroutine, 单独来接收数据.这样就不会阻塞主goroutine了.
package main
import (
"fmt"
"test007/pipeline"
"time"
)
func main() {
sourceChan := pipeline.ArraySource(1, 4, 8, 2, 19, 5)
go func() {
for v := range sourceChan {
fmt.Println("data:", v)
}
}()
time.Sleep(time.Second * 10)
}加上一个单独的goroutine为什么不会发生死锁?
我猜: 因为单开一个goroutine去sourceChan中接收数据, 那么....他就不会阻塞主线程向下运行. 单独的goroutine的作用是: 如果有数据, 就处理, 没有, 就等待. 哪怕等一年,两年....它都可以等.
上面这个demo需要记住的重点:
1. channel是goroutine和goroutine之间的通信
2. 如果不想要发生死锁, 那么向channel中放数据和从channel中取数据都要在一个单独的goroutine中进行. 很有可能这两个goroutine永远都不工作, 但是活着, 不会影响主goroutine
3.2 管道是有方向的
管道的输入, 和管道的返回值都是有方向的.
还是上面的demo. 如果ArraySource的返回值是一个 <- chan int, 那么, 表示, 返回的是一个可以取数据的管道. 那么, 后面接收这个返回值的变量, 就不可以向其中放数据
package pipeline
import "time"
func ArraySource(arr ...int) <- chan int {
out := make(chan int)
go func() {
for a := range arr {
out <- a
}
time.Sleep(time.Second * 2)
out <- 9
//close(out)
}()
return out
}注意, 这里的返回值类型的管道是可以取数据的管道.
那么像下面这样, 在管道里放数据是不ok的, 直接报错
3.3. 学习定义一个返回值为channel的func
学习老师写管道方法的模板
// 比如: 老师要进行一个sort排序
func Sort(in <-chan int) <- chan int{
// 第一步: 定义一个chan
out := make(chan int)
// 第三步: 向管道里放数据, 单独开一个goroutine
go func() {
}()
// 第二步: return 这个channel
return out
}第一步: 定义一个chan of int 类型的管道变量
第二步: 将这个管道变量返回
第三步: 向管道中放数据. 放什么样的数据, 那就是业务逻辑了.
这样定义, 就不会发生死锁. 因为,开一个goroutine是很快的.
3.4 定义排序方法
func InMemSort(in <-chan int) <-chan int {
// 第一步: 定义一个channel变量
out := make(chan int)
// 第三步: 向channel中放数据
go func() {
var arr []int
// 管道到这里会阻塞, 等待close以后, 才会退出这个for循环
// 如果没有close,就会发生死锁
for v := range in {
arr = append(arr, v)
}
fmt.Println("arr: ",arr)
sort.Ints(arr)
fmt.Println("arr: ",arr)
for _, v := range arr {
out <- v
}
}()
// 第二步: 返回这个channel
return out
}这个排序方法, 就使用了老师定义一个channel 方法的三个步骤
在第三步: 处理业务. 首先, 从输入的in管道中, 取出数据.
取出数据使用的是for循环. 循环从管道中取数据. 这里会发生阻塞. 直到所有的数据全部都被取出, 且in管道close. 否则, 会无限循环等待下去.
3.5 归并算法, 将两个数组中的数据进行合并
依然采用的是三步走方法,
第一步: 定一个channel变量,
第二步: 返回channel 变量
第三步: 定义goroutine放入数据
func Merge(in1, in2 <-chan int) <-chan int {
// 第一步: 定义一个管道变量
out := make(chan int)
// 第三步: 向管道中放入数据
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管道, 表示已经取完了
//close(out)
}()
// 第二步: 返回这个管道
return out
}这里业务逻辑在goroutine里面. 因为是管道. 所以,每次从管道中取一个数据出来, 循环从管道取数据比较.
然后, 我们造两个数组, 测试将两个数组合并
package main
import (
"fmt"
"test007/pipeline"
"time"
)
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() {
for v := range mergeChan {
fmt.Println("data:", v)
}
}()
time.Sleep(time.Second * 10)
}输出结果:
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
/private/var/folders/g2/74np978j3971l2864zdk7lgc0000gn/T/___go_build_main_go_darwin #gosetup
arr: [1 4 8 2 19 5]
arr: [0 29 43 1 7 9]
arr: [1 2 4 5 8 19]
arr: [0 1 7 9 29 43]
data: 0
data: 1
data: 1
data: 2
data: 4
data: 5
data: 7
data: 8
data: 9
data: 19
data: 29
data: 43
Process finished with exit code 0这里遇到一个问题:
in1中的数据取完了, 发现, 后面的代码就不执行了. 虽然没有发生死锁, 但是, 阻止了后面数据的输出.
究其原因,在这里:
在做内部排序的时候, 排序完了, 没有close. 这样就导致, range 管道的时候, 如果没有数据, 就一直等待, 如果始终没有, 就卡在那里了.
3.6 改变数据源为从文件读取
之前,我们的数据源是自己定义的一个数组ArraySource. 传进来一个数组, 然后, 我们将数组放入管道中进行处理. 如下情况:
func ArraySource(arr ...int) <- chan int {
out := make(chan int)
go func() {
for _, a := range arr {
out <- a
}
close(out)
}()
return out
}真实情况是. 我们通常是从文件读取数据源, 比如日志文件. 而且, 数据源可能不止一个. 比如: hdfs有n个, 我们要读取n个hdfs数据源.
下面, 我们从文件读取数据源. 这里定义一种通用读取的方式. 不管是什么数据源, 我们都使用reader来读取. 读取的内容, 放入管道中.
/**
* 从redader中读取数据
*/
func ReaderSource(reader io.Reader) <- chan int{
out := make(chan int)
go func() {
// 这里定义为8个字节, 原因是我的机器是64位的, 所以int也是64位, 那么对应的字节数就是8个字节
buffer := make([]byte, 8)
for {
// reader返回两个参数, 第一个是读取到的字节数, 第二个是err异常
n, err := reader.Read(buffer)
if n > 0 {
// 如果读到了, 就把读到的东西发给channel
u := binary.BigEndian.Uint64(buffer)
out <- int(u)
}
if err != nil {
break
}
}
close(out)
}()
return out
}接下来, 我们有读取数据源了, 还有最后只进不出的Sink, Sink从管道里读数据, 将读到的数据, 输出
/**
* 只读数据, 不写数据的, 将读出来的数据打印出来
* 可以打印到控制台, 也可以写入到文件. 这里就写入到文件
*/
func WriteSink(writer io.Writer, in <-chan int) {
for v := range in {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(v))
writer.Write(b)
}
}最后为了方便操作, 我们定义一个生成随机数的代码. 将随机数写入到文件中
/**
* 生成一个随机数
*/
func RandomSource(count int) chan int {
out := make(chan int)
go func() {
// 生成count个随机数
for i := 0; i<count ; i++ {
out <- rand.Int()
}
close(out)
}()
return out
}下面对以上操作进行一个整合. 整合代码如下:
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, e := os.Open(fileName)
if e != 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 {
break
}
}
}最后,生成随机数, 就是一个造数据的过程. 我们可以利用这个main方法, 早两类数据, 一个是小数据, 一个是大数据.
其实, 这个部分,就是练习channel. go语言是如何通过channel进行通信的. 每一个部分, 都是channel
3.7 多路合并
接下来,我们实现多路合并, 其实这里的多路合并, 是多个数据源通过传古来, 但最终还是两两进行合并.
/**
* N个节点两两归并
*/
func MergeN(inputs ... <-chan int) <-chan int{
if len(inputs) == 1 {
return inputs[0]
}
middle := len(inputs) / 2
// 两个两个的合并
return Merge(MergeN(inputs[:middle]...), MergeN(inputs[middle:]...))
}先将输入分为两半, 然后递归再去两半, 直到最后将两个通道数据合并.
3.8 单机外部排序
单机外部排序分为3部分:
1. 读取数据, 将数据进行合并
2. 将数据写入文件
3. 将写入文件的数据读出来
首先来看第一部分: 从文件读取数据, 将数据源两两合并, 最后返回合并后的数据通道
/**
* @param fileName: 文件名
* @param fileSize: 文件大小
* @param chunkCount: 将文件分成多少块
*/
func createPipeline(fileName string, fileSize, chunkCount int) <- chan int{
pipeline.Init()
// 每次读取的内容的字节大小
chunkSize := fileSize/chunkCount
sortResult := []<-chan int{}
for i := 0; i < chunkCount; i++ {
file, e := os.Open(fileName)
if e != nil {
panic(e)
}
// offset: 从文件的什么位置开始读 whence: 从第几个字符开始读
file.Seek(int64(chunkSize*i) , 0)
// 读取文件的内容
source := pipeline.ReaderSource(bufio.NewReader(file), chunkSize)
// 在内存中对内容进行排序
sortResult = append(sortResult, pipeline.InMemSort(source))
}
// 合并所有的内部排序后的结果
return pipeline.MergeN(sortResult...)
}第二部分: 将数据写入到文件
/**
* 将合并后的结果写入到文件
*/
func writeToFile(p <- chan int, fileName string) {
file, e := os.Create(fileName)
if e != nil {
panic(e)
}
defer file.Close()
writer := bufio.NewWriter(file)
defer writer.Flush()
// 这一步是将读取的内容写入到文件
pipeline.WriteSink(writer, p)
}第三部分: 将文件中的数据打印到控制台
func printFile(fileName string) {
file, e := os.Open(fileName)
if e != nil {
panic(e)
}
defer file.Close()
reader := bufio.NewReader(file)
source := pipeline.ReaderSource(reader, -1)
count := 0
for v := range source {
fmt.Println(v)
count ++
if count > 100 {
break
}
}
}这里, 我们在上面生成了一个512k的文件, 那么最后合并后的数据也应该是512k
大数据是8000000字节, 那么最后合并后也应该是80000000字节.
阶段总结:
通过上面的demo, 可以,所有的方法, 都是通过管道进行通信. 传一个管道到过去, 管道里现在可能没有数据, 那么就等待, 直到有数据了, 才能取出
上面的demo由两部分:
第一部分: 造数据. 随机生成数据, 然后保存到文件
第二部分: 将文件中的数据, 分片读取, 分了4片. 对每一片数据, 进行内部排序, 排序后, n片数据, 在两两归并排序, 最后输出一路数据. 然后将数据输出到文件
如上图分析:
可以看到, 基本都是使用管道进行的通信, 读取数据的时候, 并不是说, 最开始放入一个数据, 到最后, 输出一个数据, 中间有等待的过程.
只要有等待, 那么就可能发生死锁, 所以, 放完了数据, 一定要调用close. 这样, 取数据的一方就不会不停的等待.
这个就是搭建的管道通信方式
还有一个部分就是带有缓冲的管道. 发现,如果管道没有缓冲, 那就, 就要一直等待. 我放一个, 有人取走了,我再放一个, 走人再去走了, 我再放一个, 知道放数据的一边说, 我放完了. 然后, 取数据的一遍就结束了, 不在取了
这样1对1 效率有些低, 因此, 我们将给管道增加一个缓冲, 比如这里增加了1000个数据的缓冲, 也就是里面可以放1000个数据, 这样就大大提高了效率
// 定义一个节点, 处理数据
func InMemSort(in <-chan int) <-chan int {
// 第一步: 定义一个channel变量
out := make(chan int, 1000)
// 第三步: 向channel中放数据
go func() {
var arr []int
// 管道到这里会阻塞, 等待close以后, 才会退出这个for循环
// 如果没有close,就会发生死锁
for v := range in {
arr = append(arr, v)
}
fmt.Println("read data: ", time.Now().Sub(startTime))
sort.Ints(arr)
fmt.Println("sorted data: ", time.Now().Sub(startTime))
for _, v := range arr {
out <- v
}
close(out)
}()
// 第二步: 返回这个channel
return out
}
func Merge(in1, in2 <-chan int) <-chan int {
// 第一步: 定义一个管道变量
out := make(chan int, 1000)
// 第三步: 向管道中放入数据
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管道, 表示已经取完了
close(out)
fmt.Println("merged data: ", time.Now().Sub(startTime))
}()
// 第二步: 返回这个管道
return out
}
/**
* 从redader中读取数据
* 将reader改为分块读取, 每次读取指定字符长度
*/
func ReaderSource(reader io.Reader, trunkSize int) <- chan int{
out := make(chan int, 1000)
go func() {
// 这里定义为8个字节, 原因是我的机器是64位的, 所以int也是64位, 那么对应的字节数就是8个字节
buffer := make([]byte, 8)
readered := 0
for {
// 记录已经读取的个数
// reader返回两个参数, 第一个是读取到的字节数, 第二个是err异常
n, err := reader.Read(buffer)
readered += n
if n > 0 {
// 如果读到了, 就把读到的东西发给channel
u := binary.BigEndian.Uint64(buffer)
out <- int(u)
}
if err != nil || (trunkSize != -1 && readered >= trunkSize){
break
}
//fmt.Println("已读字符数", readered)
}
close(out)
}()
return out
}标红的部分, 增加了管道缓冲, 提高了管道处理的效率
3.9 网络版外部排序
通过上面的排序结果, 我们看到, 一个800M的文件排序时间大概是40-50秒. 其实这个时间并不快, 或者说, 如果不用管道, 那么会更快. 用了管道反而更慢了, 那我们为什么还要用管道呢?
首先, 用了管道为什么会变慢呢? 因为, 管道之间的通信, 有等待的过程. 肯定是要比直接处理要慢的.
第二: 虽然用管道会慢, 但我们依然用它,为什么么? 这里是开启了4路并行处理. 文件一共800M, 那么如果是8G呢?800G呢?我们能用一个线程单独去执行么? 显然不可以. 一定要用这种并行的方式.
----------------------------
通常服务器的日志都是放在不同的机器上的, 某几台机器接收日志文件. 然后传输给其他机器进行数据处理. 数据处理以后, 在发送给其他机器, 进行数据合并, 最后入库. 这几个步骤可能都发生在不同的机器上. 接下来, 我们就真实模拟一下, 服务器之间, 是如何传输这些数据的.
接下来我们要做的事情是这样的
将InMemSort在内存中排序和ReaderSource读取数据进行合并, 这两个步骤分开. 分别在两个服务器上执行.
原理: 有多少个节点, 就开多少个server, 然后merge节点去接这些server.
---------------------------
现在要做的有两件事情
1. 从文件读取到的数据, 放入到server中, 然后将数据通过网络发送里给连接到客户端的client
2. 客户端对数据进行Merge后输出到文件
提取第一部分: 将数据源的文件读取后发送到server中
func NetWorkSink(addr string, in <- chan int) {
// 第一步: 开启服务器的监听端口
listener, e := net.Listen("tcp", addr)
if e != nil {
panic(e)
}
go func() {
defer listener.Close()
// 第二步: 等待客户端连接
conn, e := listener.Accept()
if e != nil {
panic(e)
}
defer conn.Close()
// 第三步: 将数据通过网络发送出去
writer := bufio.NewWriter(conn)
defer writer.Flush()
WriteSink(writer, in)
}()
}将数据发送给连接的客户端
第二部分: 客户端接收到数据后, 读取数据,并发送到通道里面
func NetWorkSource(addr string) <-chan int{
out := make(chan int)
go func() {
conn, e := net.Dial("tcp", addr)
if e != nil {
panic(e)
}
defer conn.Close()
reader := bufio.NewReader(conn)
source := ReaderSource(reader, -1)
for s := range source {
out <- s
}
close(out)
}()
return out
}接下来创建一个网络pipeline
/**
* @param fileName: 文件名
* @param fileSize: 文件大小
* @param chunkCount: 将文件分成多少块
*/
func createNetWorkPipeline(fileName string, fileSize, chunkCount int) <- chan int{
pipeline.Init()
// 每次读取的内容的字节大小
chunkSize := fileSize/chunkCount
sortResult := []<-chan int{}
sortAddr := []string{}
for i := 0; i < chunkCount; i++ {
file, e := os.Open(fileName)
if e != nil {
panic(e)
}
// offset: 从文件的什么位置开始读 whence: 从第几个字符开始读
file.Seek(int64(chunkSize*i) , 0)
// 读取文件的内容
source := pipeline.ReaderSource(bufio.NewReader(file), chunkSize)
sort := pipeline.InMemSort(source)
addr := ":" + strconv.Itoa(7000 + i)
pipeline.NetWorkSink(addr, sort)
// 在内存中对内容进行排序
//sortResult = append(sortResult, pipeline.InMemSort(source))
sortAddr = append(sortAddr, addr)
}
for _, s := range sortAddr {
sortResult = append(sortResult, pipeline.NetWorkSource(s))
}
// 合并所有的内部排序后的结果
return pipeline.MergeN(sortResult...)
}最后测试:网络版的文件接收通信,
func main() {
// 第一步: 生成pipeline文件
p := createNetWorkPipeline("large.in",
800000000, 4)
//time.Sleep(time.Hour)
// 第二步: 写入到文件
writeToFile(p, "large.out")
// 第三步: 打印出来
printFile("large.out")
}运行结果:
总结:
网络版这一块做的事情, 是在讲什么? 在模拟真实的使用场景.
总结: 再次体验了整个go是如何使用chan进行通信的. 几乎每一部都是在使用chan进行通信.
最后这个网络版的的排序, 只是一个简单的模拟, 不过, 真实情况也许就是这样的. 修改一下就可以使用在客户端和服务端了.
3.10 附源码
项目结构
1. nodes.go
package pipeline
import (
"encoding/binary"
"fmt"
"io"
"math/rand"
"sort"
"time"
)
var startTime time.Time
func Init() {
startTime = time.Now()
}
func ArraySource(arr ...int) <- chan int {
out := make(chan int, 1000)
go func() {
for _, a := range arr {
out <- a
}
close(out)
}()
return out
}
// 定义一个节点, 处理数据
func InMemSort(in <-chan int) <-chan int {
// 第一步: 定义一个channel变量
out := make(chan int, 1000)
// 第三步: 向channel中放数据
go func() {
var arr []int
// 管道到这里会阻塞, 等待close以后, 才会退出这个for循环
// 如果没有close,就会发生死锁
for v := range in {
arr = append(arr, v)
}
fmt.Println("read data: ", time.Now().Sub(startTime))
sort.Ints(arr)
fmt.Println("sorted data: ", time.Now().Sub(startTime))
for _, v := range arr {
out <- v
}
close(out)
}()
// 第二步: 返回这个channel
return out
}
func Merge(in1, in2 <-chan int) <-chan int {
// 第一步: 定义一个管道变量
out := make(chan int, 1000)
// 第三步: 向管道中放入数据
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管道, 表示已经取完了
close(out)
fmt.Println("merged data: ", time.Now().Sub(startTime))
}()
// 第二步: 返回这个管道
return out
}
/**
* 从redader中读取数据
* 将reader改为分块读取, 每次读取指定字符长度
*/
func ReaderSource(reader io.Reader, trunkSize int) <- chan int{
out := make(chan int, 1000)
go func() {
// 这里定义为8个字节, 原因是我的机器是64位的, 所以int也是64位, 那么对应的字节数就是8个字节
buffer := make([]byte, 8)
readered := 0
for {
// 记录已经读取的个数
// reader返回两个参数, 第一个是读取到的字节数, 第二个是err异常
n, err := reader.Read(buffer)
readered += n
if n > 0 {
// 如果读到了, 就把读到的东西发给channel
u := binary.BigEndian.Uint64(buffer)
out <- int(u)
}
if err != nil || (trunkSize != -1 && readered >= trunkSize){
break
}
//fmt.Println("已读字符数", readered)
}
close(out)
}()
return out
}
/**
* 只读数据, 不写数据的, 将读出来的数据打印出来
* 可以打印到控制台, 也可以写入到文件. 这里就写入到文件
*/
func WriteSink(writer io.Writer, in <-chan int) {
for v := range in {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(v))
writer.Write(b)
}
}
/**
* 生成一个随机数
*/
func RandomSource(count int) chan int {
out := make(chan int, 1000)
go func() {
// 生成count个随机数
for i := 0; i<count ; i++ {
out <- rand.Int()
}
close(out)
}()
return out
}
/**
* N个节点两两归并
*/
func MergeN(inputs ... <-chan int) <-chan int{
if len(inputs) == 1 {
return inputs[0]
}
middle := len(inputs) / 2
// 两个两个的合并
return Merge(MergeN(inputs[:middle]...), MergeN(inputs[middle:]...))
}2. main.go
package main
import (
"bufio"
"fmt"
"os"
"test007/pipeline"
"time"
)
func main() {
// 第一步: 造数据, 生成100个随机数, 写入到文件
const fileName = "large.in"
const count = 100000000
// 第一步: 将随机生产的数字保存到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, e := os.Open(fileName)
if e != nil {
panic(e)
}
defer f.Close()
readerSource := pipeline.ReaderSource(bufio.NewReader(f), -1)
var num = 0
for rs := range readerSource{
fmt.Println(rs)
num ++
if num > 100 {
break
}
}
}
func MergeDemo() {
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() {
for v := range mergeChan {
fmt.Println("data:", v)
}
}()
time.Sleep(time.Second * 10)
}第三步: net_nodes.go
package pipeline
import (
"bufio"
"net"
)
func NetWorkSink(addr string, in <- chan int) {
// 第一步: 开启服务器的监听端口
listener, e := net.Listen("tcp", addr)
if e != nil {
panic(e)
}
go func() {
defer listener.Close()
// 第二步: 等待客户端连接
conn, e := listener.Accept()
if e != nil {
panic(e)
}
defer conn.Close()
// 第三步: 将数据通过网络发送出去
writer := bufio.NewWriter(conn)
defer writer.Flush()
WriteSink(writer, in)
}()
}
func NetWorkSource(addr string) <-chan int{
out := make(chan int)
go func() {
conn, e := net.Dial("tcp", addr)
if e != nil {
panic(e)
}
defer conn.Close()
reader := bufio.NewReader(conn)
source := ReaderSource(reader, -1)
for s := range source {
out <- s
}
close(out)
}()
return out
}4. sort.go
package main
import (
"bufio"
"fmt"
"os"
"strconv"
"test007/pipeline"
)
func main() {
// 第一步: 生成pipeline文件
p := createNetWorkPipeline("large.in",
800000000, 4)
//time.Sleep(time.Hour)
// 第二步: 写入到文件
writeToFile(p, "large.out")
// 第三步: 打印出来
printFile("large.out")
}
func printFile(fileName string) {
file, e := os.Open(fileName)
if e != nil {
panic(e)
}
defer file.Close()
reader := bufio.NewReader(file)
source := pipeline.ReaderSource(reader, -1)
count := 0
for v := range source {
fmt.Println(v)
count ++
if count > 100 {
break
}
}
}
/**
* 将合并后的结果写入到文件
*/
func writeToFile(p <- chan int, fileName string) {
file, e := os.Create(fileName)
if e != nil {
panic(e)
}
defer file.Close()
writer := bufio.NewWriter(file)
defer writer.Flush()
// 这一步是将读取的内容写入到文件
pipeline.WriteSink(writer, p)
}
/**
* @param fileName: 文件名
* @param fileSize: 文件大小
* @param chunkCount: 将文件分成多少块
*/
func createPipeline(fileName string, fileSize, chunkCount int) <- chan int{
pipeline.Init()
// 每次读取的内容的字节大小
chunkSize := fileSize/chunkCount
sortResult := []<-chan int{}
for i := 0; i < chunkCount; i++ {
file, e := os.Open(fileName)
if e != nil {
panic(e)
}
// offset: 从文件的什么位置开始读 whence: 从第几个字符开始读
file.Seek(int64(chunkSize*i) , 0)
// 读取文件的内容
source := pipeline.ReaderSource(bufio.NewReader(file), chunkSize)
// 在内存中对内容进行排序
sortResult = append(sortResult, pipeline.InMemSort(source))
}
// 合并所有的内部排序后的结果
return pipeline.MergeN(sortResult...)
}
/**
* @param fileName: 文件名
* @param fileSize: 文件大小
* @param chunkCount: 将文件分成多少块
*/
func createNetWorkPipeline(fileName string, fileSize, chunkCount int) <- chan int{
pipeline.Init()
// 每次读取的内容的字节大小
chunkSize := fileSize/chunkCount
sortResult := []<-chan int{}
sortAddr := []string{}
for i := 0; i < chunkCount; i++ {
file, e := os.Open(fileName)
if e != nil {
panic(e)
}
// offset: 从文件的什么位置开始读 whence: 从第几个字符开始读
file.Seek(int64(chunkSize*i) , 0)
// 读取文件的内容
source := pipeline.ReaderSource(bufio.NewReader(file), chunkSize)
sort := pipeline.InMemSort(source)
addr := ":" + strconv.Itoa(7000 + i)
pipeline.NetWorkSink(addr, sort)
// 在内存中对内容进行排序
//sortResult = append(sortResult, pipeline.InMemSort(source))
sortAddr = append(sortAddr, addr)
}
for _, s := range sortAddr {
sortResult = append(sortResult, pipeline.NetWorkSource(s))
}
// 合并所有的内部排序后的结果
return pipeline.MergeN(sortResult...)
}参考资料: