Go並行処理パターン

https://classmethod.connpass.com/event/55140/ の発表資料

Goの並行処理に必要な幾つかの機能と、それをつかったサンプルをご紹介

Transcript

1. (Pฒߦॲཧύλʔϯ 
2. ϞόΠϧόοΫΤϯυ
   େࡕࣄ຿ॴ!̍ਓ
   3BJMTͱ͔/PEFͱ͔࢖ͬͯ"1*࡞ͬͯ·͢
   (Pྺϲ݄ 
3. جૅ 
4. ฒߦॲཧ͍ͨ͠ཧ༝ w $16ίΞΛෳ਺࢖͍͍ͨ
   w ϊϯϒϩοΩϯάͰॲཧ͍ͨ͠ 
5. $16ͷΫϩοΫ਺೥ද 0 750 1500 2250 3000 1980 1990 2000 2010 2020 ͋ͨ·͏ͪ ٸܹʹ৳ͼͨ 
6. ίΞ ίΞ ॲཧΛ෼ׂ ฒྻ ίΞ ඵ ඵ 
7. ଴ͪΛղফ ฒߦ ॲཧ ॲཧ *0଴ͪ *0଴ͪ ॲཧ ॲཧ *0଴ͪ *0଴ͪ *0଴ͪ *0଴ͪ *0଴ͪ *0଴ͪ ଴ͪ࣌ؒΛͳ͘͠
   ϨεϙϯεΛૣ͘͢Δ 
8. $41  $PNNVOJDBUJOH
   4FRVFOUJBM
   1SPDFTTFT
9. ໊લ͕ࣔ͢ͱ͓Γɺ$41͸ಠཱͨ͠ϓϩηε܈͕ϝοηʔδύογ ϯάʹΑͬͯ௨৴͢Δ͜ͱͰ૬ޓʹ΍ΓऔΓ͍ͯ͠Δ΋ͷͱͯ͠γ εςϜΛهड़͢Δɻ͔͠͠ɺ$41ͷ໊শʹؚ·ΕΔ
   4FRVFOUJBM ʢஞ࣍తʣͱ͍͏෦෼͸ޡղΛੜ͡ΔՄೳੑ͕͋Δɻͱ͍͏ͷ΋࠷ ۙͷ$41Ͱ͸ɺϓϩηε͸୯ͳΔஞ࣍తϓϩηε͚ͩͰͳ͘ɺΑΓ جຊతͳϓϩηε܈ͷฒྻ߹੒Ͱੜ੒͞ΕΔϓϩηε΋ؚ·ΕΔ͔ ΒͰ͋Δɻϓϩηεؒͷؔ܎΍ϓϩηε͕पғͱ௨৴͢Δํ๏͸ɺ ֤छϓϩηε୅਺ԋࢉࢠΛ࢖ͬͯද͞ΕΔɻ͜ͷΑ͏ͳ୅਺తख๏ Λ࢖͏͜ͱͰɺগ਺ͷϓϦϛςΟϒཁૉ͔Β༰қʹۃΊͯෳࡶͳϓ ϩηεΛߏஙͰ͖Δɻ 8JLJQFEJBΑΓ 
10. ͳΔ΄ͲΘ͔ΒΜ 
11. ฒߦॲཧͷ໰୊ɹ 1SPDFTT  1SPDFTT  $POUFYU
    QSPHSFTT
    
     1SPDFTTؒͷ$POUFYUͷڞ༗͕೉͍͠ 
12. ฒߦॲཧͷछྨ w 4IBSFE
    .FNPSZ
    w .FTTBHF
    1BTTJOH 
13. 4IBSFE
    .FNPSZ  5ISFBE
     5ISFBE
     .VUFY -PDL -PDL
    "RVJSFE -PDL 6OMPDL -PDL
    "RVSJFE 6OMPDL Block
14. %FBEMPDL  5ISFBE
     5ISFBE
     .VUFY -PDL -PDL
    "RVJSFE -PDL -PDL
    "RVSJFE .VUFY -PDL -PDL
15. .VUFY౳Λ࢖͍जΕ෺Λ৮ΔΑ͏ʹ
    ϝϞϦΛอޢ͠ͳ͕Β΍ΓͱΓΛߦ͏
    Ұൠతʹ೉͍͠ॲཧ 
16. .FTTBHF
    1BTTJOH 1SPDFTT  1SPDFTT  ϓϩηεಉ͕࢜ϝοηʔδΛૹΓ߹ͬͯ΍ΓͱΓ͢Δ  ༻ҙ͞Εͨ఻ୡखஈΛ࢖͏͜ͱʹ
    Αͬͯ೉͍͜͠ͱΛ
    ͋Δఔ౓͓·͔ͤͰ͖Δ WBS
17. %P
    OPU
    DPNNVOJDBUF
    CZ
    TIBSJOH
    NFNPSZ
    JOTUFBE
    TIBSF
    NFNPSZ
    CZ
    DPNNVOJDBUJOH 
18. (Pಛ༗ͷߏจ 
19. func onExit() { fmt.Println("exit!") } func main() { defer onExit() fmt.Println("end") } > go run main.go end exit! EFGFS ؔ਺ऴྃ࣌ʹ࣮ߦ͢Δؔ਺Λొ࿥͢Δ
20. func main() { arr := []int{1,2,3} for _, i := range arr { fmt.Println(i) } } > go run main.go 1 2 3 SBOHF ഑ྻ΍$IBOOFM͔Βஞ࣍తʹ஋ΛऔΓग़͢
21. ฒߦॲཧͷಓ۩ɹ 
22. (PSPVUJOF  ॲཧͷ࣮ߦ୯Ґ
23. ಛ௃ w εϨουͱ΄΅ಉ͡
    w εϨουΑΓܰྔ
    w εϨουΑΓܰྔͳͷͰͨ͘͞Μ্ཱͪ͛ͯ΋ಈ࡞͢Δ
    w (PͷεέδϡʔϥʹΑΓ(PSPVUJOFΛෳ਺ͷεϨο υʹׂΓ౰ͯΒΕΔͨΊɺ(PSPVUJOFͰฒߦॲཧ͢Δ ͱಁաతʹ$16ίΞ͕ෳ਺࢖ΘΕΔ 
24. ίετ w ͭ
    όΠτ
    εϨουͷελοΫαΠ ζ͸͍͍ͩͨ.όΠτ
    w ੾Γସ͕͑εϨουΑΓܰྔ 
25. ࢖͍ํ go func() {}(); 25 HPΩʔϫʔυʹؔ਺ݺͼग़͠Ͱ
    (PSPVUJOFΛੜ੒͠ฒߦॲཧΛߦ͏ HPΩʔϫʔυ
    
    ؔ਺ݺͼग़͠
26. ཁૉ go func() {}(); 26 ಗ໊ؔ਺
    +BWB4DSJQUͰ͍͏
    GVODUJPO \^ ؔ਺ݺͼग़͠ɹ HPΩʔϫʔυɹ
27. ࢖͍ํ func foo() { time.Sleep(1 * time.Second) fmt.Println("2") } func main() { fmt.Println("1") // go キーワードの後に関数呼び出しを行うとGoroutineが // 生成され並行に実行される go foo() fmt.Println("3") time.Sleep(2 * time.Second) } 27 > go run main.go 1 3 2
28. $IBOOFM 28 (PSPVUJOFಉ࢜ͷϝοηʔδΛதܧ͢ΔύΠϓ PL PL
29. ಛ௃ w (PSPVUJOFಉ࢜ͷ஋ͷड͚౉͠ͷͨΊͷಓ۩
    w ϝοηʔδύογϯάΛ࣮ݱ͢Δಓ۩
    w ૹ৴ɺड৴͕Ͱ͖Δ·ͰHPSPVUJOFΛϒϩοΫ͢Δ
    w #VGGFSΛࢦఆͰ͖Δ
    w ϑΝʔετΫϥεΦϒδΣΫτͳͷͰม਺ͱͯ͠ड͚౉ ͕͠Ͱ͖Δ 29
30. ࢖͍ํ // 送信する値の型を指定しChannelを生成 ch := make(chan int) // 受信(Channelから <-演算子で取り出す) val := <- ch // 変数に代入しなくてもいい <- ch // 送信(Channelに向かって <-演算子で送信) ch <- 1 // 閉じる close(ch) 30
31. ࢖͍ํ func main() { ch := make(chan int) go func() { // 送信 ch <- 1 }() // 受信 i := <-ch fmt.Println(i) } 31 > go run main.go 1
32. ࢖͍ํ func main() { ch := make(chan int) go func() { // 送信 ch <- 1 }() // 受信 i := <-ch fmt.Println(i) } 32 DI 
33. GPS
    SBOHF
    Λ࢖ͬͯ
    ഑ྻͷΑ͏ʹϧʔϓͰ͖Δ func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() // Channelがcloseされるまでループする for i := range ch { fmt.Println(i) } } 33 > go run main.go 1 2
34. #VGGFSFE
    $IBOOFM // bufferの数を指定するとbuffer有りのChannelになる ch := make(chan int, 2) ch <- 1 ch <- 1 ch <- 1 // bufferの数を超えて書き込もうとするとブロック 34
35. DMPTF func main() { ch := make(chan int, 1) ch <- 1 close(ch) // ch <- 1 CloseされたChannelには送信できない // Closeされても読み出せる fmt.Println(<-ch) // 呼び出せる値がなくなった場合は // ゼロ値(intの場合0、stringの場合"")が返ってくる fmt.Println(<-ch) fmt.Println(<-ch) } 35 > go run main.go 1 0 0
36. 4FMFDU 36 ෳ਺ͷ$IBOOFMΛಉ࣌ʹૢ࡞͢Δ
37. ಛ௃ w ෳ਺ͷ$IBOOFMΛѻ͏ͱ͖ʹ࢖͏
    w ͍ͣΕ͔ͷ$IBOOFM͕ड৴ɺૹ৴Ͱ͖Δঢ়ଶʹͳΔ· ͰϒϩοΫ͢Δ
    w ֤DBTFͷॲཧ͕ಉ࣌ʹ૸Βͳ͍͜ͱ͕อূ͞Ε͍ͯΔ
    w EFGBVMU۟Λॻ͘ͱɺશͯͷ$IBOOFM͕ड৴ɺૹ৴ Ͱ͖ͳ͍৔߹ʹ࣮ߦ͞ΕΔɻ ϒϩοΫ͞Εͳ͘ͳΔ 37
38. select { case v := <-ch1: // Ch1 から受信 case <-ch2: // 受信した場合でも代入式は必須じゃない case ch3 <- 1: // 送信することも可能 default: // 全てのcaseで送受信不能な場合 }  ࢖͍ํ
39. dataCh1 := make(chan int) dataCh2 := make(chan int) closeCh := make(chan struct{}) go func() { for { // selectは一度処理すると抜けてしまうのでforループで囲む fmt.Println("select start") select { case d := <-dataCh1: fmt.Println(">>> 1:", d) time.Sleep(1 * time.Second) fmt.Println("<<< 1:", d) case d := <-dataCh2: fmt.Println(">>> 2:", d) time.Sleep(1 * time.Second) fmt.Println("<<< 2:", d) case <-closeCh: fmt.Println("close") return } } }() fmt.Println("Send 1 => dataCh1") dataCh1 <- 1 fmt.Println("Send 2 => dataCh1") dataCh1 <- 2 fmt.Println("Send 1 => dataCh2") dataCh2 <- 1 fmt.Println("Send true => closeCh") closeCh <- struct{}{}  > go run main.go Send 1 => dataCh1 select start >>> 1: 1 Send 2 => dataCh1 <<< 1: 1 select start >>> 1: 2 Send 1 => dataCh2 <<< 1: 2 select start >>> 2: 1 Send true => closeCh <<< 2: 1 select start close
40. TZODύοέʔδ (PͰ͸ϝοηʔδύογϯά͚ͩͰશͯΛ࿫͏ͷ͸
    ݱ࣮తͰͳ͍ͱ͍͏ཧ༝Ͱ4IBSFE
    .FNPSZͷ
    πʔϧ΋༻ҙ͞Ε͍ͯ·͢ 40
41. छྨ w TZOD8BJU(SPVQ
    w TZOD0ODF
    w TZOD.VUFY 41
42. छྨ w TZOD8BJU(SPVQ
    w TZOD0ODF
    w TZOD.VUFY 42
43. ಛ௃ w ෳ਺ͷ(PSPVUJOFͷऴྃΛ଴ͭͨΊͷಓ۩ 43
44. wg := new(sync.WaitGroup) // AddでWaitする数を追加する wg.Add(1) // Done()でAddした数をデクリメント wg.Done() // Addした数が0になるまでSleep wg.Wait()  ࢖͍ํ
45. func main() { wg := new(sync.WaitGroup) for i := 0; i < 10; i++ { // 待つ数を登録する。1回のループにつきGoroutineを // 1つ生成するので1を追加していく wg.Add(1) go func(i int) { // deferキーワードでこの関数を抜けるときに // wg.Done()が呼ばれるように登録する defer wg.Done() time.Sleep(1 * time.Second) fmt.Println(i) }(i) } // wg.Addで追加した数の合計数wg.Done()が呼ばれるまで待機 fmt.Println("** wait **") wg.Wait() fmt.Println("** done **") }  > go run main.go ** wait ** 0 8 6 2 1 3 4 5 7 9 ** done ** ࢖͍ํ
46. ύλʔϯ 46
47. 4ZOD (PSVUJOFಉ࢜ͷ଴ͪ߹Θͤ 47
48. // 終了のお知らせを伝えるためのChannel wait := make(chan struct{}) go func() { time.Sleep(1 * time.Second) // 処理が終了したらwaitに値を送信 wait <- struct{}{} }() // 値が書き込まれるまでブロック <-wait 
49. 'BO*O 49 ෳ਺ͷ1SPDFTT͔Β
    ͭͷ1SPDFTTʹू໿͢Δ
50. 1SPEVDFS 1SPEVDFS $POTVNFS 50
51. func producer(out chan<- int, v int) { for i := 0; i < 10; i++ { out <- v time.Sleep(time.Duration(v) * time.Second) } } func consumer() { ch := make(chan int) go producer(ch, 1) go producer(ch, 2) // producerが送信した値を集約して出力 for i := range ch { fmt.Println(i) } } 
52. 'BO0VU 52 ͭͷ2VFVFΛෳ਺ͷϓϩηεͰॲཧ͢Δ
53. $POTVNFS $POTVNFS ͭͷ2VFVF͔Βෳ਺ͷϓϩηεʹ౉͢ 1SPEVDFS 53
54. func consumer(ch <-chan int, wg *sync.WaitGroup) { wg.Add(1) defer wg.Done() for i := range ch { fmt.Println(i) } } func producer(size int) { queue := make(chan int) wg := new(sync.WaitGroup) go consumer(queue, wg) go consumer(queue, wg) for i := 0; i < size; i++ { queue <- i } close(queue) wg.Wait() } 
55. (FOFSBUPS ഑ྻͷ୅ΘΓʹ(PSPVUJOFͰ஋Λੜ੒͠
    $IBOOFMͰૹ৴͢Δ͜ͱͰϝϞϦΛઅ໿͢Δ 55
56. func Generator(n int) chan int { ch := make(chan int) go func() { for i := 0; i < n; i++ { // 与えられた数まで順番にChannelに書き込んで行く ch <- i } close(ch) }() return ch } func main() { for i := range Generator(10000) { fmt.Printf("%d", i) } } 
57. 1JQFMJOF (PSPVUJOFͷग़ྗ͕ผͷ(PSPVUJOFͷೖྗʹͳͬͯΔ
    $IBOOFMΛ௨ͯ͠ஞ࣍తʹॲཧ͞ΕΔ 57
58. (FOFSBUPS EPVCMF QSJOUFS ΠϯΫϦϝϯτ
    ͨ͠஋ΛPVUʹ
    ॻ͖ࠐΉ JOͰड͚औͬͨ
    ஋Λഒͯ͠PVUʹ
    ॻ͖ࠐΉ JOͰड͚औͬͨ
    ग़ྗ͢Δ    ʜ   ʜ
59. func generator(n int) <-chan int { out := make(chan int) go func() { defer close(out) for i := 0; i < n; i++ { out <- i } }() return out } func double(in <-chan int) <-chan int { out := make(chan int) go func() { defer close(out) for i := range in { out <- i * 2 } }() return out } func print(in <-chan int) { for i := range in { fmt.Println(i) } } func main() { ch1 := generator(10) ch2 := double(ch1) print(ch2) }  > go run main.go 0 2 4 6 8 10 12 14 16 18
60. Τϥτεςωεͷᝲ ࢀߟ
    IUUQTUBMLTHPMBOHPSH DPODVSSFODZTMJEF  ࢦఆ͞Εͨ੔਺ҎԼͷૉ਺ΛશͯٻΊΔΞϧΰϦζϜ
61. ୳ࡧϦετ
    
    
    
    
    ʜ ૉ਺Ϧετ 9·Ͱͷ஋Λ୳ࡧϦετ
 ʹೖΕΔ ୳ࡧϦετ
    
    
    
    
    ʜ ૉ਺Ϧετ
     ୳ࡧϦετͷઌ಄Λ
    ૉ਺ϦετʹҠ͢ ୳ࡧϦετ
    
    
    
    ʜ ૉ਺Ϧετ
     ୳ࡧϦετ͔ΒҠͨ͠
    ૉ਺ͷഒ਺Λ࡟আ͢Δ ୳ࡧϦετͷઌ಄Λ
    ૉ਺ϦετʹҠ͢ ୳ࡧϦετ
    
    
    
    ʜ ૉ਺Ϧετ
    
     Ҏ߱܁Γฦ͠ʜ
62. // Send the sequence 2, 3, 4, ... to channel 'ch'. func Generate(ch chan<- int) { for i := 2; ; i++ { ch <- i // Send 'i' to channel 'ch'. } } // Copy the values from channel 'in' to channel 'out', // removing those divisible by 'prime'. func Filter(in <-chan int, out chan<- int, prime int) { for { i := <-in // Receive value from 'in'. if i%prime != 0 { out <- i // Send 'i' to 'out'. } } } // The prime sieve: Daisy-chain Filter processes. func main() { ch := make(chan int) // Create a new channel. go Generate(ch) // Launch Generate goroutine. for i := 0; i < 10; i++ { prime := <-ch fmt.Println(prime) ch1 := make(chan int) go Filter(ch, ch1, prime) ch = ch1 } } 62
63. (FOFSBUF ૉ਺ͷ 'JMUFS ૉ਺ͷ 'JMUFS    ʜ */ */ 065     ʜ ૉ਺ͷ 'JMUFS 065 */    ʜ ૉ਺͕ݟ͔ͭΔຖʹ 'JMUFS͕૿͍͑ͯ͘ ϑΟϧλ͞Εͳ͔ͬͨ
    ࠷ॳͷ਺͕ૉ਺ QSJNF 
64. func Generate(ch chan<- int) { for i := 2; ; i++ { ch <- i // 2から順に最初のFilterに送信していきます } } 64
65. // in => 前のFilter(Generator)から数値を渡されるChannel // out => 次のFilterに数値を渡すChannel // prime => Filterに設定される素数 func Filter(in <-chan int, out chan<- int, prime int) { for { i := <-in // Filter(Generator)から数値が渡ってくる if i%prime != 0 { out <- i // 自身に渡された素数で割り切れないもの次のFilterに送る } } } 65
66. func main() { ch := make(chan int) go Generator(ch) // 素数を10個見つけるまでループ for i := 0; i < 10; i++ {
 // 待機リストの最初の数値を素数として受信 prime := <-ch fmt.Println(prime) // FilterのoutになるChannelを用意 ch1 := make(chan int) go Filter(ch, ch1, prime) // 次のフィルタのinになるChannel変数に // 今回のontのChannelを代入する ch = ch1 } } 66
67. ͔ͬ͜Β͸࣌ؒ༨ͬͨ࣌༻
    ଓ͖͸ϒϩάͰʂ 
68. #BDLHSPVOE ॏ͍ͨॲཧΛόοΫάϥ΢ϯυͰ࣮ߦ͠
    શମͷॲཧΛૣ͘͢Δ 68
69. ֓ཁ w 8&#Ϋϩʔϥʔ
    w αʔόʔ͸໿NT͔͔Δ
    w ಺༰ͱͯ͠͸3FTQPOTFͷTUBUVTίʔυ ͷ਺ΛΧ΢ϯτͯ͠Δ͚ͩ 
70. func fetch(url string) int { res, _ := http.Get(url) return res.StatusCode } func crawl(urls []string) { m := make(map[int]int) for _, url := range urls { status := fetch(url) m[status]++ } } ~/w/g/s/g/c/g/p/crawler ››› go build; time ./crawler -n 100 ./crawler -n 100 0.08s user 0.12s system 0% cpu 30.554 total ฒߦॲཧͳ͠ 
71. ԟ෮ʹ໿NTʹ͔͔Γ
    ௚ྻͰ࣮ߦ͞ΕΔͷͰ ϦΫΤετ਺͔͔Δ 71 'FUDI 4FSWFS ϦΫΤετ Ϩεϙϯε ໿300ms ໿300ms +α ϦΫΤετ Ϩεϙϯε
72. func fetch(url string) int { // レスポンスが返るまでブロック res, _ := http.Get(url) return res.StatusCode } ͜ͷϒϩοΫ͕ ϘτϧωοΫ 
73. func crawl(urls []string) { m := make(map[int]int) // statusを受信するChannel ch := make(chan int) for _, url := range urls { go func() { // ここが並行処理 ch <- fetch(url) }() } // urlsの数だけchから読み取る for range urls { m[<-ch]++ } } ~/w/g/s/g/c/g/p/crawler ››› go build; time ./crawler -n 100⏎ ./crawler -n 100 0.04s user 0.03s system 20% cpu 0.338 total ඵ͔Β NT
    ·Ͱ୹ॖ 
74. ϦΫΤετຖʹผͷ(PSPVUJOFͰ
    ॲཧ͠଴͕ͪ࣌ؒͳ͘ͳͬͨͷͰ
    NTͰऴྃ 74 4FSWFS 'FUDI Ϩεϙϯε ϦΫΤετ ϦΫΤετ Ϩεϙϯε
75. func crawl(urls []string) { m := make(map[int]int) // statusを受信するChannel ch := make(chan int) for _, url := range urls { go func() { // ここが並行処理 ch <- fetch(url) }() } // urlsの数だけchから読み取る for range urls { m[<-ch]++ } } ௒͑Δ͘Β͍Ͱ
    Τϥʔ͕සൃ
    .BD
    #PPL
    "JS ~/w/g/s/g/c/g/p/crawler ››› go build; time ./crawler -n 500 panic: Get http://localhost:65067: dial tcp [::1]:65067: getsockopt: connection refused 
76. αʔόʔ ಉ࣌઀ଓ਺͕ଟ͗͢αʔόʔෛՙ্͕͕Γ͗ͯ͢
    ઀ଓ͕Ͱ͖ͳ͘ͳ͍ͬͨͨ 76 ☓ ☓
77. 4FNBQIPSF ಉ࣌ʹ࣮ߦ͢Δ(PSPVUJOFͷ਺Λ੍ݶ͢Δ 77
78. αʔόʔ (PSPVUJOFͷ਺Λ੍ݶ͠ಉ࣮࣌ߦ਺Λ཈੍ 78
79. func crawl(urls []string) { m := make(map[int]int) ch := make(chan int) // Buffered Channelの指定回数以上送信したら // 受信されるまで送信がブロックされるという特性を利用します sem := make(chan struct{}, 100) go func() { for _, url := range urls { // 100回まで送信するとブロック sem <- struct {}{} go func() { ch <- fetch(url) // 処理が終了すればChannelから受信しバッファを開けることで //同時並行数を開放していく <-sem }() } }() for range urls { m[<-ch]++ } } ~/w/g/s/g/c/g/p/crawler ››› go build; time ./crawler -n 500 ./crawler -n 500 0.17s user 0.14s system 16% cpu 1.895 total ஗͘ͳ͕ͬͨ
    ਖ਼ৗऴྃ͢ΔΑ͏ʹ 
80. 4JHOBM
    )BOEMFS (PͰ͸4JHOBM
    )BOEMFS΋$IBOOFMͰॲཧ͢Δ 80
81. // Signalを受け取るChannelを宣言 sigCh := make(chan os.Signal, 1) // 受け取るSignalを登録 signal.Notify(sigCh, syscall.SIGINT, syscall.SIGUSR1) go func() { for { select { // Signalを待ち受け case sig := <-sigCh: switch sig { case syscall.SIGINT: fmt.Println("SIGINT") os.Exit(0) case syscall.SIGUSR1: fmt.Println("SIGUSR") } } } }() 
82. 5JDL 5JDL Ұఆ࣌ؒຖͷॲཧ ΍5JNFPVUॲཧ΋
    (PͰ͸$IBOOFMΛ࢖ͬͯߦ͍·͢ 82
83. func main() { // 3秒毎にChannelに書き込まれる tick := time.Tick(3 * time.Second) // 15秒後にChannelに書き込まれる timeout := time.After(15 * time.Second) for { select { case <-tick: fmt.Println("tick") case <-timeout: fmt.Println("timeout") return } } 83
84. 1BSBMMFM
    8SJUF
    'JMF ͭͷϑΝΠϧΛෳ਺ͷ(PSVUJOFͰ
    ෼ׂͯ͠ॻ͖ࠐΜͰ͍͘ 84
85. QSPEVDFS ֤8PSLFS͕
    ͭͷϑΝΠϧͷผͷ
    ։࢝Ґஔ͔Βಉ࣌ʹॻ͖ࠐΜͰ͍͘  8PSLFS 8PSLFS 8PSLFS BB CC DD CZUFʙ CZUFʙ CZUFʙ +PC
    ։࢝Ґஔͱ
    ॻ͖ࠐΉCZUF഑ྻ
86. // ジョブ構造体 type Job struct { // 書き込むbyte配列 b []byte // 書き込むFileの開始位置 pos int64 } 86
87. type Worker struct { f *os.File // Fileポインター wg *sync.WaitGroup // 処理を待ち合わせるためのWaitGroup queue chan *Job // Jobを受信するChannel stopCh chan bool // 全てのJobが消化されWorkerを終了させたい時のChannel } // 慣例的にNew${構造体名}のメソッドを用意してインスタンスを返します func NewWorker(f *os.File, wg *sync.WaitGroup, queue chan *Job, stopCh chan bool) Worker { return Worker{ f: f, wg: wg, queue: queue, stopCh: stopCh, } } 87
88. func (w *Worker) Start() { w.wg.Add(1) go func() { defer w.wg.Done() for { select { case j := <-w.queue: // queueを受け取ったらその情報でFileに書き込み w.f.WriteAt(j.b, j.pos) case <-w.stopCh: // stopChに値が書き込まれたら関数を抜けて処理を終了します // select内は同時に走らないので最後に受け取ったqueueの処理が終わってから // このセクションが実行される事が保証される return } } }() } 88
89. func parallelWrite(lines []string) { wg := new(sync.WaitGroup) f, _ := os.Create("test.txt") defer f.Close() stopCh := make(chan bool) queue := make(chan *Job) const workers = 10 // Workerを指定回数スタートさせる for i := 0; i < workers; i++ { worker := NewWorker(f, wg, queue, stopCh) worker.Start() } pos := int64(0) for _, line := range lines { lineBytes := []byte(line + "\n") // Queueを送信。全てのWorkerが処理中なら書き込めずにここでブロック queue <- &Job{ b: lineBytes, pos: pos, } // 書き込まれたbyte数分書き込み位置をすすめる pos += int64(len(lineBytes)) } // workerを終了させるためにstopChに値を書き込んでいく for i := 0; i < workers; i++ { stopCh <- true } wg.Wait() } 
90. ·ͱΊ w (PͰ͸؆୯ʹฒߦॲཧ͕Ͱ͖Δಓ۩͕ἧͬ ͯΔ
    w (PͰ͸ܰྔͳ(PSPVUJOF͕࢖͑ΔͷͰฏ ߦ਺Λ্͛΍͍͢
91. ͏͍͋ʔ͸͍͋ΓΜ͙  େࡕࣄ຿ॴͰҰॹʹಇ͍ͯ͘ΕΔਓʂ