Anatomie des canaux dans Go

Bonjour, Habr! Je vous présente la traduction de l'article "Anatomie des canaux en Go" par Uday Hiwarale.

Quels sont les canaux?

Un canal est un objet de communication à travers lequel les goroutins échangent des données. Techniquement, il s'agit d'un convoyeur (ou tuyau) à partir duquel vous pouvez lire ou mettre des données. Autrement dit, un goroutin peut envoyer des données au canal, et l'autre peut lire les données placées dans ce canal.

Création de chaîne

Go fournit le mot-clé chan pour créer une chaîne. Un canal peut transmettre des données d'un seul type, les données d'autres types ne peuvent pas être transmises par ce canal.

package main

import "fmt"

func main() {
    var c chan int
    fmt.Println(c)
}

Exemple sur play.golang.org

c, int. <nil>, — nil. . , (). make.

package main

import "fmt"

func main() {
    c := make(chan int)

    fmt.Printf("type of `c` is %T\n", c)
    fmt.Printf("value of `c` is %v\n", c)
}

play.golang.org

:= make. :

type of `c` is chan int
value of `c` is 0xc0420160c0

c, . go . , , . , , .

Go <-

c <- data

c. , data c.

<- c

c. . , :

var data int
data = <- c

c, int, data. , :

data := <- c

Go , c, data .

. , , . , . , .

package main

import "fmt"

func greet(c chan string) {
    fmt.Println("Hello " + <-c + "!")
}

func main() {
    fmt.Println("main() started")
    c := make(chan string)

    go greet(c)

    c <- "John"
    fmt.Println("main() stopped")
}

play.golang.org

:

greet, c . c .
main "main() started".
, make, c string.
greet , go.
main greet, main - .
main , (greet) c. Go greet .
main "main() stopped".

Deadlock ( )

, . , , "". deadlock, .

, , , - . : , .

deadlock main , .

package main

import "fmt"

func main() {
    fmt.Println("main() started")

    c := make(chan string)
    c <- "John"

    fmt.Println("main() stopped")
}

play.golang.org

:

main() started
fatal error: all goroutines are asleep - deadlock!
goroutine 1 [chan send]:
main.main()
        program.go:10 +0xfd
exit status 2

Go , . , : val, ok := <- channel, ok , , ok false, . , close, close(channel). :

package main

import "fmt"

func greet(c chan string) {
    <-c // for John
    <-c // for Mike
}

func main() {
    fmt.Println("main() started")

    c := make(chan string, 1)

    go greet(c)
    c <- "John"

    close(c) // closing channel

    c <- "Mike"
    fmt.Println("main() stopped")
}

play.golang.org

c <- "John" , , greet . , c. , c , main close(c).

:

main() started
panic: send on closed channel

goroutine 1 [running]:
main.main()
    program.go:20 +0x120
exit status 2

, , , . for.

for

package main

import "fmt"

func squares(c chan int) {
    for i := 0; i <= 9; i++ {
        c <- i * i
    }

    close(c) // close channel
}

func main() {
    fmt.Println("main() started")
    c := make(chan int)

    go squares(c) // start goroutine

    // periodic block/unblock of main goroutine until chanel closes
    for {
        val, ok := <-c
        if ok == false {
            fmt.Println(val, ok, "<-- loop broke!")
            break // exit break loop
        } else {
            fmt.Println(val, ok)
        }
    }

    fmt.Println("main() stopped")
}

play.golang.org

, . squares, 0 9. main for.

, val, ok := <-c, ok , . squares , , , close. ok true, val ( ok). ok false, , break. :

main() started
0 true
1 true
4 true
9 true
16 true
25 true
36 true
49 true
64 true
81 true
0 false <-- loop broke!
main() stopped

, val, , , . int, 0, : 0 false <-- loop broke!

, for, Go range, , . range:

package main

import "fmt"

func squares(c chan int) {
    for i := 0; i <= 9; i++ {
        c <- i * i
    }

    close(c) // close channel
}

func main() {
    fmt.Println("main() started")
    c := make(chan int)

    go squares(c) // start goroutine

    // periodic block/unblock of main goroutine until chanel closes
    for val := range c {
        fmt.Println(val)
    }

    fmt.Println("main() stopped")
}

play.golang.org

val := range c , range , . :

main() started
0
1
4
9
16
25
36
49
64
81
main() stopped

for range, - dealock .

, . make 2- . — . - 0, . , , .

0, , . , , , , ( ). , , , , . , , .

:

c := make(chan Type, n)

Type n. , n+1 .

, , :

package main

import "fmt"

func squares(c chan int) {
    for i := 0; i <= 3; i++ {
        num := <-c
        fmt.Println(num * num)
    }
}

func main() {
    fmt.Println("main() started")
    c := make(chan int, 3)

    go squares(c)

    c <- 1
    c <- 2
    c <- 3

    fmt.Println("main() stopped")
}

play.golang.org

c 3. , 3 (c <- 3), ( ), main , . :

main() started
main() stopped

:

package main

import "fmt"

func squares(c chan int) {
    for i := 0; i <= 3; i++ {
        num := <-c
        fmt.Println(num * num)
    }
}

func main() {
    fmt.Println("main() started")
    c := make(chan int, 3)

    go squares(c)

    c <- 1
    c <- 2
    c <- 3
    c <- 4 // blocks here

    fmt.Println("main() stopped")
}

play.golang.org

, main , squares, , .

, . — ( ) , — . , , len, , cap, .

package main

import "fmt"

func main() {
    c := make(chan int, 3)
    c <- 1
    c <- 2

    fmt.Printf("Length of channel c is %v and capacity of channel c is %v", len(c), cap(c))
    fmt.Println()
}

play.golang.org

:

Length of channel c is 2 and capacity of channel c is 3

deadlock , 3, 2 , main. main, , .

:

package main

import "fmt"

func sender(c chan int) {
    c <- 1 // len 1, cap 3
    c <- 2 // len 2, cap 3
    c <- 3 // len 3, cap 3
    c <- 4 // <- goroutine blocks here
    close(c)
}

func main() {
    c := make(chan int, 3)

    go sender(c)

    fmt.Printf("Length of channel c is %v and capacity of channel c is %v\n", len(c), cap(c))

    // read values from c (blocked here)
    for val := range c {
        fmt.Printf("Length of channel c after value '%v' read is %v\n", val, len(c))
    }
}

play.golang.org

:

Length of channel c is 0 and capacity of channel c is 3
Length of channel c after value '1' read is 3
Length of channel c after value '2' read is 2
Length of channel c after value '3' read is 1
Length of channel c after value '4' read is 0

:

package main

import (
    "fmt"
    "runtime"
)

func squares(c chan int) {
    for i := 0; i < 4; i++ {
        num := <-c
        fmt.Println(num * num)
    }
}

func main() {
    fmt.Println("main() started")
    c := make(chan int, 3)
    go squares(c)

    fmt.Println("active goroutines", runtime.NumGoroutine())
    c <- 1
    c <- 2
    c <- 3
    c <- 4 // blocks here

    fmt.Println("active goroutines", runtime.NumGoroutine())

    go squares(c)

    fmt.Println("active goroutines", runtime.NumGoroutine())

    c <- 5
    c <- 6
    c <- 7
    c <- 8 // blocks here

    fmt.Println("active goroutines", runtime.NumGoroutine())
    fmt.Println("main() stopped")
}

play.golang.org

:

main() started
active goroutines 2
1
4
9
16
active goroutines 1
active goroutines 2
25
36
49
64
active goroutines 1
main() stopped

for range, . , :

package main

import "fmt"

func main() {
    c := make(chan int, 3)
    c <- 1
    c <- 2
    c <- 3
    close(c)

    // iteration terminates after receiving 3 values
    for elem := range c {
        fmt.Println(elem)
    }
}

play.golang.org

2 , , :

package main

import "fmt"

func square(c chan int) {
    fmt.Println("[square] reading")
    num := <-c
    c <- num * num
}

func cube(c chan int) {
    fmt.Println("[cube] reading")
    num := <-c
    c <- num * num * num
}

func main() {
    fmt.Println("[main] main() started")

    squareChan := make(chan int)
    cubeChan := make(chan int)

    go square(squareChan)
    go cube(cubeChan)

    testNum := 3
    fmt.Println("[main] sent testNum to squareChan")

    squareChan <- testNum

    fmt.Println("[main] resuming")
    fmt.Println("[main] sent testNum to cubeChan")

    cubeChan <- testNum

    fmt.Println("[main] resuming")
    fmt.Println("[main] reading from channels")

    squareVal, cubeVal := <-squareChan, <-cubeChan
    sum := squareVal + cubeVal

    fmt.Println("[main] sum of square and cube of", testNum, " is", sum)
    fmt.Println("[main] main() stopped")
}

play.golang.org

:

2 square cube, . c c int, num. c.
main squareChan cubeChan c int.
square cube .
- main testNum 3.
squareChan cubeChan. main , . , .
main (squareChan cubeChan), , (square cube) . := .
, main, .

:

[main] main() started
[main] sent testNum to squareChan
[cube] reading
[square] reading
[main] resuming
[main] sent testNum to cubeChan
[main] resuming
[main] reading from channels
[main] sum of square and cube of 3  is 36
[main] main() stopped

, . , . , , , .

make, .

roc := make(<-chan int)
soc := make(chan<- int)

roc , soc . , .

package main

import "fmt"

func main() {
    roc := make(<-chan int)
    soc := make(chan<- int)

    fmt.Printf("Data type of roc is `%T`\n", roc)
    fmt.Printf("Data type of soc is `%T\n", soc)
}

play.golang.org

:

Data type of roc is `<-chan int`
Data type of soc is `chan<- int

? e, , , .

, , , / . ?

Go .

import "fmt"

func greet(roc <-chan string) {
    fmt.Println("Hello " + <-roc + "!")
}

func main() {
    fmt.Println("main() started")
    c := make(chan string)

    go greet(c)

    c <- "John"
    fmt.Println("main() stopped")
}

play.golang.org

greet , . , :

"invalid operation: roc <- "some text" (send to receive-only type <-chan string)"

. , .
:

package main

import "fmt"

func main() {
    fmt.Println("main() started")
    c := make(chan string)

    // launch anonymous goroutine
    go func(c chan string) {
        fmt.Println("Hello " + <-c + "!")
    }(c)

    c <- "John"
    fmt.Println("main() stopped")
}

play.golang.org

.

, , , / . :

package main

import "fmt"

// gets a channel and prints the greeting by reading from channel
func greet(c chan string) {
    fmt.Println("Hello " + <-c + "!")
}

// gets a channels and writes a channel to it
func greeter(cc chan chan string) {
    c := make(chan string)
    cc <- c
}

func main() {
    fmt.Println("main() started")

    // make a channel `cc` of data type channel of string data type
    cc := make(chan chan string)

    go greeter(cc) // start `greeter` goroutine using `cc` channel

    // receive a channel `c` from `greeter` goroutine
    c := <-cc

    go greet(c) // start `greet` goroutine using `c` channel

    // send data to `c` channel
    c <- "John"

    fmt.Println("main() stopped")
}

play.golang.org

select

select switch , . select , case.

, :

package main

import (
    "fmt"
    "time"
)

var start time.Time
func init() {
    start = time.Now()
}

func service1(c chan string) {
    time.Sleep(3 * time.Second)
    c <- "Hello from service 1"
}

func service2(c chan string) {
    time.Sleep(5 * time.Second)
    c <- "Hello from service 2"
}

func main() {
    fmt.Println("main() started", time.Since(start))

    chan1 := make(chan string)
    chan2 := make(chan string)

    go service1(chan1)
    go service2(chan2)

    select {
    case res := <-chan1:
        fmt.Println("Response from service 1", res, time.Since(start))
    case res := <-chan2:
        fmt.Println("Response from service 2", res, time.Since(start))
    }

    fmt.Println("main() stopped", time.Since(start))
}

play.golang.org

select switch, , . select , default( ). case, main . case ?

case , select , case . , case ( : , ).

, . 2 . select c case . case chan1 chan2. , . case select , case .

select main ( ), select, service1 service2. service1 3 , chan1. service1 service2, 5 chan2. service1 , service2, case chan1, case . main, .

:

main() started 0s
Response from service 1 Hello from service 1 3s
main() stopped 3s

-, . , select, , , , .

, , case , Sleep .

package main

import (
    "fmt"
    "time"
)

var start time.Time
func init() {
    start = time.Now()
}

func service1(c chan string) {
    c <- "Hello from service 1"
}

func service2(c chan string) {
    c <- "Hello from service 2"
}

func main() {
    fmt.Println("main() started", time.Since(start))

    chan1 := make(chan string)
    chan2 := make(chan string)

    go service1(chan1)
    go service2(chan2)

    select {
    case res := <-chan1:
        fmt.Println("Response from service 1", res, time.Since(start))
    case res := <-chan2:
        fmt.Println("Response from service 2", res, time.Since(start))
    }

    fmt.Println("main() stopped", time.Since(start))
}

play.golang.org

:

main() started 0s
service2() started 481µs
Response from service 2 Hello from service 2 981.1µs
main() stopped 981.1µs

:

main() started 0s
service1() started 484.8µs
Response from service 1 Hello from service 1 984µs
main() stopped 984µs

, chan1 chan2 , .

, case , .

package main

import (
    "fmt"
    "time"
)

var start time.Time

func init() {
    start = time.Now()
}

func main() {
    fmt.Println("main() started", time.Since(start))
    chan1 := make(chan string, 2)
    chan2 := make(chan string, 2)

    chan1 <- "Value 1"
    chan1 <- "Value 2"
    chan2 <- "Value 1"
    chan2 <- "Value 2"

    select {
    case res := <-chan1:
        fmt.Println("Response from chan1", res, time.Since(start))
    case res := <-chan2:
        fmt.Println("Response from chan2", res, time.Since(start))
    }

    fmt.Println("main() stopped", time.Since(start))
}

play.golang.org

:

main() started 0s
Response from chan2 Value 1 0s
main() stopped 1.0012ms

:

main() started 0s
Response from chan1 Value 1 0s
main() stopped 1.0012ms

2. 2 , select. , , case , Go case .

default case

switch, select default. default , , default select . , ( ) .

- , select case. , default.

package main

import (
    "fmt"
    "time"
)

var start time.Time

func init() {
    start = time.Now()
}

func service1(c chan string) {
    fmt.Println("service1() started", time.Since(start))
    c <- "Hello from service 1"
}

func service2(c chan string) {
    fmt.Println("service2() started", time.Since(start))
    c <- "Hello from service 2"
}

func main() {
    fmt.Println("main() started", time.Since(start))

    chan1 := make(chan string)
    chan2 := make(chan string)

    go service1(chan1)
    go service2(chan2)

    select {
    case res := <-chan1:
        fmt.Println("Response from service 1", res, time.Since(start))
    case res := <-chan2:
        fmt.Println("Response from service 2", res, time.Since(start))
    default:
        fmt.Println("No response received", time.Since(start))
    }

    fmt.Println("main() stopped", time.Since(start))
}

play.golang.org

:

main() started 0s
No response received 0s
main() stopped 0s

, , default. select default, .

default select , . main , time.Sleep. , main, .

package main

import (
    "fmt"
    "time"
)

var start time.Time

func init() {
    start = time.Now()
}

func service1(c chan string) {
    fmt.Println("service1() started", time.Since(start))
    c <- "Hello from service 1"
}

func service2(c chan string) {
    fmt.Println("service2() started", time.Since(start))
    c <- "Hello from service 2"
}

func main() {
    fmt.Println("main() started", time.Since(start))

    chan1 := make(chan string)
    chan2 := make(chan string)

    go service1(chan1)
    go service2(chan2)

    time.Sleep(3 * time.Second)

    select {
    case res := <-chan1:
        fmt.Println("Response from service 1", res, time.Since(start))
    case res := <-chan2:
        fmt.Println("Response from service 2", res, time.Since(start))
    default:
        fmt.Println("No response received", time.Since(start))
    }

    fmt.Println("main() stopped", time.Since(start))
}

play.golang.org

:

main() started 0s
service1() started 0s
service2() started 0s
Response from service 1 Hello from service 1 3.0001805s
main() stopped 3.0001805s

, :

main() started 0s
service1() started 0s
service2() started 0s
Response from service 2 Hello from service 2 3.0000957s
main() stopped 3.0000957s

Deadlock

, deadlock, default, , Go , .

package main

import (
    "fmt"
    "time"
)

var start time.Time

func init() {
    start = time.Now()
}

func main() {
    fmt.Println("main() started", time.Since(start))

    chan1 := make(chan string)
    chan2 := make(chan string)

    select {
    case res := <-chan1:
        fmt.Println("Response from chan1", res, time.Since(start))
    case res := <-chan2:
        fmt.Println("Response from chan2", res, time.Since(start))
    default:
        fmt.Println("No goroutines available to send data", time.Since(start))
    }

    fmt.Println("main() stopped", time.Since(start))
}

play.golang.org

:

main() started 0s
No goroutines available to send data 0s
main() stopped 0s

, default, , ( ).

nil

, — nil, - . select, .

package main

import "fmt"

func service(c chan string) {
    c <- "response"
}

func main() {
    fmt.Println("main() started")

    var chan1 chan string

    go service(chan1)

    select {
    case res := <-chan1:
        fmt.Println("Response from chan1", res)
    }

    fmt.Println("main() stopped")
}

play.golang.org

:

main() started
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [select (no cases)]:
main.main()
    program.go:17 +0xc0

goroutine 6 [chan send (nil chan)]:
main.service(0x0, 0x1)
    program.go:6 +0x40
created by main.main
    program.go:14 +0xa0

, select (no cases) , select , case . select{} main , service, nil , : chan send (nil chan). , , default.

package main

import "fmt"

func service(c chan string) {
    c <- "response"
}

func main() {
    fmt.Println("main() started")

    var chan1 chan string

    go service(chan1)

    select {
    case res := <-chan1:
        fmt.Println("Response from chan1", res)
    default:
        fmt.Println("No response")
    }

    fmt.Println("main() stopped")
}

play.golang.org

:

main() started
No response
main() stopped

case , default . service. , , , , nil.

timeout

- , default . , , , default. , case , . After (package) time. :

package main

import (
    "fmt"
    "time"
)

var start time.Time

func init() {
    start = time.Now()
}

func service1(c chan string) {
    time.Sleep(3 * time.Second)
    c <- "Hello from service 1"
}

func service2(c chan string) {
    time.Sleep(5 * time.Second)
    c <- "Hello from service 2"
}

func main() {
    fmt.Println("main() started", time.Since(start))

    chan1 := make(chan string)
    chan2 := make(chan string)

    go service1(chan1)
    go service2(chan2)

    select {
    case res := <-chan1:
        fmt.Println("Response from service 1", res, time.Since(start))
    case res := <-chan2:
        fmt.Println("Response from service 2", res, time.Since(start))
    case <-time.After(2 * time.Second):
        fmt.Println("No response received", time.Since(start))
    }

    fmt.Println("main() stopped", time.Since(start))
}

play.golang.org

2 :

main() started 0s
No response received 2s
main() stopped 2s

, <-time.After(2 * time.Second) main 2 . time.After , . chan1 chan2 , 3- , .

, . time.After(2 * time.Second) time.After(10 * time.Second) service1.

select

for{}, select{} , . select , case , select case, , , deadlock.

package main

import "fmt"

func service() {
    fmt.Println("Hello from service!")
}

func main() {
    fmt.Println("main() started")

    go service()

    select {}

    fmt.Println("main() stopped")
}

play.golang.org

:

main() started
Hello from service!
fatal error: all goroutines are asleep - deadlock!
goroutine 1 [select (no cases)]:
main.main()
        program.go:16 +0xba
exit status 2

WaitGroup

, , ( : , ). , select. .

WaitGroup. , , ( : , , , , , , ). , .

:

package main

import (
    "fmt"
    "sync"
    "time"
)

func service(wg *sync.WaitGroup, instance int) {
    time.Sleep(2 * time.Second)
    fmt.Println("Service called on instance", instance)
    wg.Done() // decrement counter
}

func main() {
    fmt.Println("main() started")
    var wg sync.WaitGroup // create waitgroup (empty struct)

    for i := 1; i <= 3; i++ {
        wg.Add(1) // increment counter
        go service(&wg, i)
    }

    wg.Wait() // blocks here
    fmt.Println("main() stopped")
}

play.golang.org

WaitGroup, noCopy state1 (https://golang.org/src/sync/waitgroup.go?s=574:929#L10). : Add, Wait Done. .

Add int , delta () WaitGroup. — , 0. . WaitGroup , 0, , delta Add. , , , Add.

Wait , . 0, . - .

Done . . ( : sync, , Add(-1)).

, wg, for 1 3 . 1. 3 , WaitGroup 3. , wg . , Done , .

for, wg.Wait(), , , , main , , 0. main , .

:

main() started
Service called on instance 1
Service called on instance 3
Service called on instance 2
main() stopped

, - .

, — , . WaitGroup , , . , - , .

package main

import (
    "fmt"
    "time"
)

// worker than make squares
func sqrWorker(tasks <-chan int, results chan<- int, id int) {
    for num := range tasks {
        time.Sleep(time.Millisecond) // simulating blocking task
        fmt.Printf("[worker %v] Sending result by worker %v\n", id, id)
        results <- num * num
    }
}

func main() {
    fmt.Println("[main] main() started")

    tasks := make(chan int, 10)
    results := make(chan int, 10)

    // launching 3 worker goroutines
    for i := 0; i < 3; i++ {
        go sqrWorker(tasks, results, i)
    }

    // passing 5 tasks
    for i := 0; i < 5; i++ {
        tasks <- i * 2 // non-blocking as buffer capacity is 10
    }

    fmt.Println("[main] Wrote 5 tasks")

    // closing tasks
    close(tasks)

    // receving results from all workers
    for i := 0; i < 5; i++ {
        result := <-results // blocking because buffer is empty
        fmt.Println("[main] Result", i, ":", result)
    }

    fmt.Println("[main] main() stopped")
}

play.golang.org

:

[main] main() started
[main] Wrote 5 tasks
[worker 0] Sending result by worker 0
[worker 2] Sending result by worker 2
[worker 1] Sending result by worker 1
[main] Result 0 : 4
[main] Result 1 : 0
[main] Result 2 : 16
[worker 2] Sending result by worker 2
[main] Result 3 : 64
[worker 0] Sending result by worker 0
[main] Result 4 : 36
[main] main() stopped

, , :

sqrWorker tasks, results, id. — , tasks, results.
main, tasks result , 10. , , . — .
sqrWorker id, , .
5 tasks, , .
tasks, . , , .
for 5 , results. , . , , main .
, . , results, , , . , tasks. , tasks, for , tasks . deadlock, tasks .
, main , results .
, , main , results, .

, , . , . time.Sleep(), , , .

, .

WaitGroup . WaitGroup, , .

package main

import (
    "fmt"
    "sync"
    "time"
)

// worker than make squares
func sqrWorker(wg *sync.WaitGroup, tasks <-chan int, results chan<- int, instance int) {
    for num := range tasks {
        time.Sleep(time.Millisecond)
        fmt.Printf("[worker %v] Sending result by worker %v\n", instance, instance)
        results <- num * num
    }

    // done with worker
    wg.Done()
}

func main() {
    fmt.Println("[main] main() started")

    var wg sync.WaitGroup

    tasks := make(chan int, 10)
    results := make(chan int, 10)

    // launching 3 worker goroutines
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go sqrWorker(&wg, tasks, results, i)
    }

    // passing 5 tasks
    for i := 0; i < 5; i++ {
        tasks <- i * 2 // non-blocking as buffer capacity is 10
    }

    fmt.Println("[main] Wrote 5 tasks")

    // closing tasks
    close(tasks)

    // wait until all workers done their job
    wg.Wait()

    // receving results from all workers
    for i := 0; i < 5; i++ {
        result := <-results // non-blocking because buffer is non-empty
        fmt.Println("[main] Result", i, ":", result)
    }

    fmt.Println("[main] main() stopped")
}

play.golang.org

:

[main] main() started
[main] Wrote 5 tasks
[worker 0] Sending result by worker 0
[worker 2] Sending result by worker 2
[worker 1] Sending result by worker 1
[worker 2] Sending result by worker 2
[worker 0] Sending result by worker 0
[main] Result 0 : 4
[main] Result 1 : 0
[main] Result 2 : 16
[main] Result 3 : 64
[main] Result 4 : 36
[main] main() stopped

, , results main , results - wg.Wait(). WaitGroup, () ( ), 7 9 . .

— Go. , race condition( ). , . , , . , . :

package main

import (
    "fmt"
    "sync"
)

var i int // i == 0

// goroutine increment global variable i
func worker(wg *sync.WaitGroup) {
    i = i + 1
    wg.Done()
}

func main() {
    var wg sync.WaitGroup

    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go worker(&wg)
    }

    // wait until all 1000 goroutines are done
    wg.Wait()

    // value of i should be 1000
    fmt.Println("value of i after 1000 operations is", i)
}

play.golang.org

1000 , i, 0. WaitGroup, , 1000 i , , , 1000. main wg.Wait(), i. :

value of i after 1000 operations is 937

? 1000? . , , race condition. , .

i = i + 1 :

i
1
i

, . , 2 1000 , . G1 G2.

G1 , i 0, i 1. , G1 i 1 3, G2 , . G2, i 0, i 1, G1 i 1. , 3- , 2 i 2, . , , i 1000.

, , , . i = i + 1 , Go ?

stackoverflow. Go .

, 3 , . — , . , , , . , .

Go — , sync. Go , race condition, , mutex.Lock(). i = i + 1 , , mutext.Unlock(). i , , . i. , , Lock Unlock, , .

, .

package main

import (
    "fmt"
    "sync"
)

var i int // i == 0

// goroutine increment global variable i
func worker(wg *sync.WaitGroup, m *sync.Mutex) {
    m.Lock() // acquire lock
    i = i + 1
    m.Unlock() // release lock
    wg.Done()
}

func main() {
    var wg sync.WaitGroup
    var m sync.Mutex

    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go worker(&wg, &m)
    }

    // wait until all 1000 goroutines are done
    wg.Wait()

    // value of i should be 1000
    fmt.Println("value of i after 1000 operations is", i)
}

, i, , m.Lock(), i , m.Unlock(). :

value of i after 1000 operations is 1000

, race condition. .

race condition Go, race, . go run -race program.go. .

, . , .

, . , . , , . , .

package main

import "fmt"

// fib returns a channel which transports fibonacci numbers
func fib(length int) <-chan int {
    // make buffered channel
    c := make(chan int, length)

    // run generation concurrently
    go func() {
        for i, j := 0, 1; i < length; i, j = i+j, i {
            c <- i
        }
        close(c)
    }()

    // return channel
    return c
}

func main() {
    // read 10 fibonacci numbers from channel returned by `fib` function
    for fn := range fib(10) {
        fmt.Println("Current fibonacci number is", fn)
    }
}

Current fibonacci number is 0
Current fibonacci number is 1
Current fibonacci number is 1
Current fibonacci number is 2
Current fibonacci number is 3
Current fibonacci number is 5
Current fibonacci number is 8

fib, , . fib, . . , . for, . main, range, , fib.

Fan-in Fan-out

Fan-in — , . Fan-out — , .


package main

import (
    "fmt"
    "sync"
)

// return channel for input numbers
func getInputChan() <-chan int {
    // make return channel
    input := make(chan int, 100)

    // sample numbers
    numbers := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}

    // run goroutine
    go func() {
        for num := range numbers {
            input <- num
        }
        // close channel once all numbers are sent to channel
        close(input)
    }()

    return input
}

// returns a channel which returns square of numbers
func getSquareChan(input <-chan int) <-chan int {
    // make return channel
    output := make(chan int, 100)

    // run goroutine
    go func() {
        // push squares until input channel closes
        for num := range input {
            output <- num * num
        }

        // close output channel once for loop finishes
        close(output)
    }()

    return output
}

// returns a merged channel of `outputsChan` channels
// this produce fan-in channel
// this is variadic function
func merge(outputsChan ...<-chan int) <-chan int {
    // create a WaitGroup
    var wg sync.WaitGroup

    // make return channel
    merged := make(chan int, 100)

    // increase counter to number of channels `len(outputsChan)`
    // as we will spawn number of goroutines equal to number of channels received to merge
    wg.Add(len(outputsChan))

    // function that accept a channel (which sends square numbers)
    // to push numbers to merged channel
    output := func(sc <-chan int) {
        // run until channel (square numbers sender) closes
        for sqr := range sc {
            merged <- sqr
        }
        // once channel (square numbers sender) closes,
        // call `Done` on `WaitGroup` to decrement counter
        wg.Done()
    }

    // run above `output` function as groutines, `n` number of times
    // where n is equal to number of channels received as argument the function
    // here we are using `for range` loop on `outputsChan` hence no need to manually tell `n`
    for _, optChan := range outputsChan {
        go output(optChan)
    }

    // run goroutine to close merged channel once done
    go func() {
        // wait until WaitGroup finishes
        wg.Wait()
        close(merged)
    }()

    return merged
}

func main() {
    // step 1: get input numbers channel
    // by calling `getInputChan` function, it runs a goroutine which sends number to returned channel
    chanInputNums := getInputChan()

    // step 2: `fan-out` square operations to multiple goroutines
    // this can be done by calling `getSquareChan` function multiple times where individual function call returns a channel which sends square of numbers provided by `chanInputNums` channel
    // `getSquareChan` function runs goroutines internally where squaring operation is ran concurrently
    chanOptSqr1 := getSquareChan(chanInputNums)
    chanOptSqr2 := getSquareChan(chanInputNums)

    // step 3: fan-in (combine) `chanOptSqr1` and `chanOptSqr2` output to merged channel
    // this is achieved by calling `merge` function which takes multiple channels as arguments
    // and using `WaitGroup` and multiple goroutines to receive square number, we can send square numbers
    // to `merged` channel and close it
    chanMergedSqr := merge(chanOptSqr1, chanOptSqr2)

    // step 4: let's sum all the squares from 0 to 9 which should be about `285`
    // this is done by using `for range` loop on `chanMergedSqr`
    sqrSum := 0

    // run until `chanMergedSqr` or merged channel closes
    // that happens in `merge` function when all goroutines pushing to merged channel finishes
    // check line no. 86 and 87
    for num := range chanMergedSqr {
        sqrSum += num
    }

    // step 5: print sum when above `for loop` is done executing which is after `chanMergedSqr` channel closes
    fmt.Println("Sum of squares between 0-9 is", sqrSum)
}

.

chanInputNums, getInputChan. getInputChan , , , numbers .
(fan-out) (chanOptSqr1 chanOptSqr2), getSquareChan. getSquareChan , , , .
(fan-in), merge. merge WaitGroup, (merged), outputsChan, , , merged, , . . , . merged.
Nous lisons les données du canal à l' chanMergedSqraide de foret range, et résumons les données reçues.
Au final, nous dérivons notre résultat.

Sortie du programme:

Sum of squares between 0-9 is 285

MISE À JOUR:

Autres publications sur les canaux internes:

All Articles