并发

很多goroutine映射到一个线程。多对1的关系。

chan是go里的一等公民。我们可以往chan里放不同类型的数据。也可以从一个声明的chan里取数据。

demo 从8:00开始

- pipeline：一個library

- pipeline：一個library

Go语言编程范式：面向接口、函数式编程、并发编程。

各个编程语言的使用场景.

外部排序pipeline

节点的组装，

传统语言与Go语言 对比

go run [file]
go build [file]
[bin]

进一步学习

• go tool tour

• http://docscn.studygolang.com/doc/

• 期待我的 go 语言实战课

go 语言三大特色

1. 面向接口- Reader/ Writer 接口的使用

2. 函数式编程

3. 并发编程

课程回顾

原始数据

网络的扩展

网络的扩展

单机版外部排序 pipeline

• Reader Source 节点-支持分块

• Mergen-搭建归并节点组

• pipeline 的搭建及运行——CPU 及线程数量的观测

func ReadSource(reader io.Reader) <-chan int {
   out := make(chan int)
   go func() {
      buffer := make([]byte, 8)
      for  {
         n, err := reader.Read(buffer)
         if n > 0 {
            v := int(binary.BigEndian.Uint64(buffer))
            out <- v
         }
         if err != nil {
            break
         }
      }
      close(out)
   }()
   return out
}

func WriteSink(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()
      }
      close(out)
   }()
   return out
}

func main() {
   const filename = "large.in"
   const n = 100000000
   file, err := os.Create(filename)
   if err != nil {
      panic(err)
   }
   defer file.Close()

   p := pipeline.RandomSource(n)
   writer := bufio.NewWriter(file)
   pipeline.WriteSink(writer, p)
   writer.Flush()

   file, err = os.Open(filename)
   if err != nil {
      panic(err)
   }
   defer file.Close()

   p = pipeline.ReadSource(bufio.NewReader(file))
   count := 0
   for v := range p {
      fmt.Println(v)
      count++
      if count >= 100 {
         break
      }
   }
}

package pipeline

import (
   "encoding/binary"
   "io"
   "sort"
)

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)
   go func() {
      // Read into memory
      var a []int
      for v := range in {
         a = append(a, v)
      }

      // Sort
      sort.Ints(a)

      // Output
      for _, v := range a {
         out <- v
      }
      close(out)
   }()
   return out
}

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(out)
   }()
   return out
}

package main

import (
   "ccmouse/gointro/pipeline"
   "fmt"
)

func main() {
   p := pipeline.Merge(
      pipeline.InMemSort(
         pipeline.ArraySource(3, 2, 6, 7, 4)),
      pipeline.InMemSort(
         pipeline.ArraySource(7, 4, 0, 3, 2, 13, 8)))
   for v := range p {
      fmt.Println(v)
   }
}

一些节点

• 数组数据源节点- channel 的关闭及检测

• 内部排序节点

• 归并节点

外部排序 Pipeline

节点的组装

节点多输入和多输出

Pipeline

外部排序归并节点

外部排序切分

将数据分为左右两半，分别归并排序，再把两个有序数据归并

如何归并：

• [1, 3, 6, 7], [1, 2, 3, 5] → 1

• [3, 6, 7]，[1, 2, 3, 5]   → 1

• [3, 6, 7], [2, 3, 5]        → 2

• [3,6,7]，[3, 5]            → 3

• ……

package main

import (
   "fmt"
   "sort"
)

func main() {
   // create a slice of int
   a := []int{3, 6, 2, 1, 9, 10,8}
   sort.Ints(a)

   for _, v := range a {
      fmt.Println(v)
   }
}

package main

import (
   "fmt"
)

func main() {
   ch := make(chan string)

   for i:=0; i < 5000; i++ {
      // go starts a goroutine
      go printHelloWorld(i, ch)
   }

   for  {
      msg := <- ch
      fmt.Println(msg)
   }
}

func printHelloWorld(i int, ch chan string) {
   for  {
      ch <- fmt.Sprintf("Hello world from go routine %d!\n", i)
   }
}

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)
}

Hello World

• 普通 Hello World

• Hello World Server - 读取 url 中的参数

• 并发 Hello World - 初识 goroutine 与 channel

• 一个简单的排序