Golang collaboration and channel usage

brief introduction

Coroutines is golang A major feature and selling point of . coroutines (goroutine) Is a lightweight execution thread , Use go Keyword to function or lamba Expressions can quickly start a coroutine . The return value of the covariance function is discarded . The scheduling of threads is managed by the operating system , It's preemptive scheduling . And the process is different , The coroner needs to work with each other , Take the initiative to surrender the right of execution .

To configure

GOMAXPROCS Set logic CPU Number , In general use CPU Core quantity value . Use enough threads to improve golang Parallel execution efficiency . If your business is IO For dense type, you can set multiple times and CPU For better performance . If Go If the program is executed in the container, the value should be reduced according to the situation , Because the container cannot use all the cores of the host . Setting a smaller value can avoid the overhead of thread switching .

Use channel Carry out process communication

Define the channel

ch1 := make(chan string) // Defines an unbuffered string passageway 
ch2 := make(chan string , 4) // Defined a 4 An element of string passageway 

Channel operator

ch1 <- "Chengdu" // Write data to channel 
itemOfCh1 := <- ch1 // from ch1 The channel reads a piece of data 
<- ch1 // Read the next value of the channel 
var in_only chan<- int // Only receive channels 
var out_only <-chan int // Read only the channel 
close(ch) // Close channel 

The channel is blocked

By default, channels are synchronized unbuffered , The sender is blocked before the receiver is ready . If there is no data in the channel, the receiver is blocked .

package main

import (
    "fmt"
    "time"
)

func f1(in chan int) {
    data := <-in
    fmt.Println(data)
}

func main() {
    out := make(chan int)
    out <- 2
    fmt.Println(v)
    go f1(out)
    time.Sleep(100 * time.Millisecond)
}

The above procedure will panic sign out , because out Write data has no receiver , therefore main The main process is blocked . The following code will never be executed , So the channel will never have data , There's a deadlock . modify out:=make(chan int , 1) If the channel has a buffer, it will not deadlock . Or start the reading process before writing . Or reading in another coroutine can solve this problem .

Use semaphores

The semaphore can be used to make the main process wait for the completion of the sub process to exit the execution .

package main

import (
    "fmt"
    "time"
)

func f1(in chan int, done chan int) {
    data := <-in
    fmt.Println(data)
    time.Sleep(10e9)
    done <- 1
}

func main() {
    out := make(chan int)
    done := make(chan int)
    go f1(out, done)
    out <- 2
    <-done
}

Output 2 after 10 Seconds before the program exits , We don't need to use it sleep To let the main process execute .

Close channel

Explicitly close the channel , Closing the channel means that the sender will not send new data to the receiver . Only the sender needs to close the channel .

ch := make(chan int )
defer close(ch)

data,ok := <-ch // When data is received ok by  true, Use ok It can detect whether the channel is closed or blocked 

This is the case , The read channel will not report a deadlock error in the main process , Because after the channel is checked to be closed, the channel reading will not be carried out and the loop will jump out , Therefore, the channel that is not written will not be read . So there is no deadlock .

package main

import "fmt"

func makeStream(n int) chan bool {
    ch := make(chan bool, n)
    go func() {
        for i := 0; i < n; i++ {
            ch <- true
        }
        close(ch)
    }()
    return ch
}

func main() {
    stream := makeStream(5)

    for {
        v, ok := <-stream
        if !ok {
            break
        }
        fmt.Println(v)
    }
}

Use select Switch coroutines

To get values from different concurrent processes, you can use select Keyword to carry out rotation training . Usually and for To use together

• If both are blocked, wait for one of them to be processed
• If more than one can be processed, randomly select one . If there is a cycle in the outer layer, the remaining... Will be processed next time
• If there is no channel to handle but default execute default. Otherwise, it will be blocked all the time
• without case, this select It will keep blocking
• have access to break Jump out of select

package main

import (
    "fmt"
    "time"
)

func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)

    go func() {
        for i := 0; i < 10; i++ {
            ch1 <- fmt.Sprintf("A%d", i)
        }
    }()

    go func() {
        for i := 0; i < 10; i++ {
            ch2 <- fmt.Sprintf("B%d", i)
        }

    }()

    go func() {
        for {
            select {
            case v := <-ch1:
                fmt.Println(v)
            case v := <-ch2:
                fmt.Println(v)
            }
        }
    }()

    time.Sleep(1e9)
}

You can use this pattern as a server to cycle through client requests

timer (Ticker)

type Ticker struct {
    C <-chan Time // The channel on which the ticks are delivered.
    r runtimeTimer
}

Timer C The variable will be based on the timer time you created , Write time to the channel within a given time

package main

import (
    "fmt"
    "time"
)

func main() {
    t := time.NewTicker(time.Second)

    go func() {
        for {
            v := <-t.C
            fmt.Println(v)
        }
    }()

    time.Sleep(10e9) // <-time.After(10e9)  Use channels to set timeouts 
}

Use time.Tick(duration) You can get the channel directly , similar time.NewTicker(1e9).C

time.After(duration) Send only once . You can use this channel to handle timeouts

Recovery of the process

Xiecheng is encountering panic Exit safely , Without affecting other processes

package main

import (
    "log"
    "time"
)

func doWork() {
    time.Sleep(4e9)
    panic("fk")
}

func main() {

    go func() {
        for {
            log.Printf("another worker")
            time.Sleep(1e9)
        }
    }()

    go func() {

        defer func() {
            if err := recover(); err != nil {
                log.Printf(" Something is wrong.  %s", err)
            }
        }()

        doWork()
    }()

    time.Sleep(10e9)
}

Use lock or channel

In one scene , There are multiple missions , One worker Deal with a task . This scenario is suitable for using channels and coroutines to solve problems

package main

type Task struct{}
type Result struct{}

func process(Task *Task) Result {
    return Result{}
}

func main() {

    tasks, results := make(chan Task), make(chan Result)

    workCount := 10

    // Create tasks 
    go func() {
        for i := 0; i < workCount; i++ {
            tasks <- Task{}
        }
    }()

    // start-up worker
    for i := 0; i < workCount; i++ {
        go func() {
            for {
                t := <-tasks
                result := process(&t) // Processing data 
                results <- result     // Write structure 
            }
        }()
    }

    // The result of consumption 

}

• The use of locks ：

  • Accessing cached information in shared data structures
  • Save application context and state information data

• The scenario of using channels ：

  • Interact with the results of an asynchronous operation
  • Distribute tasks
  • Transfer data ownership