Go language concurrency and channel

Go Language

13. Concurrency and channel

Concurrency means that in the same period of time , Programs can perform multiple tasks .

Go How language makes use of synergy （coroutine） And channel （channel） To solve the concurrency problem .

13.1 summary

There is often a lot of time-consuming work in the program , Like uploading files 、 Download the file 、 Internet chat . Now , One thread cannot serve multiple users , There will be waiting problems caused by resource monopoly , At this time, we need to use concurrency to solve .

coroutines （coroutine）, Belong to multi-threaded programming .

13.1.1 Parallelism and concurrency

parallel （parallelism）： At the same time , There are multiple instructions executing on multiple processors at the same time .

Concurrent （concurrency）： Only one instruction can be executed at the same time , But multiple process instructions are quickly rotated , Get the macro effect of multiple processes executing at the same time , But at the micro level, it is not implemented at the same time , Just divide the time slice into several sections , Make multiple processes execute alternately quickly .

Parallelism is really simultaneous execution , Concurrency only has the effect of simultaneous execution from a macro point of view .

13.1.2 Go Concurrency advantage

Go The advantage of language is that it supports concurrency at the language level , Memory management of concurrent programming is sometimes very complex , Developers don't have to worry about the underlying logic of concurrency 、 memory management , Because these have been solved at the language level , Developers only need to write their own business logic .Go Language also provides a very powerful automatic garbage collection mechanism , Developers don't have to worry about how to destroy the created quantity .

stay Go In language , Just use “go” Adding the function name can make this function become a concurrent function .

package main

func run(arg string) {
    
}

func main() {
    
   go run("This is new thread")
}

Go Language concurrency based on CSP（Communication Sequential Process, Communication sequence process ） Model ,CSP The model is in 20 century 70 It was proposed in the s to describe two independent concurrent entities through a shared communication pipeline （channel） Concurrent model for communication .

Simply speaking ,CSP The model advocates sharing memory through communication , Instead of communicating through shared memory .

Go Language is not locked , But through communication , Send and receive data through secure channels to achieve synchronization , This greatly simplifies the writing of concurrent programming .

13.2 goroutine

goroutine yes Go The core of concurrent design , Also called Xiecheng , It is lighter than threads , Therefore, thousands of concurrent tasks can be run at the same time . More Than This ,Go The language has been implemented internally goroutine Memory sharing between , It is easier to use than threads 、 Efficient and lightweight .

13.2.1 goroutine Definition

stay Go In language , Each concurrent execution unit is called a goroutine. Want to write a concurrent task , Just add... In front of the function you call go keyword , You can make this function run in a co process .

go  Function name ( Function parameter )

If the function has a return value , The return value is ignored . therefore , Once the use go keyword , You cannot use the return value of the function to exchange data with the main process , Only channels can be used .

For users , There is little difference between a coroutine and a thread , But in fact, there is a certain difference between the two . Threads have fixed stacks , Basic it is 2 MB, Are fixed distribution ; This stack is used to hold local variables , Use . But for goroutine For this lightweight collaborative process , A fixed stack size can lead to a waste of resources , therefore Go The strategy of dynamic expansion and contraction is adopted , Initialize to 2 KB, It can be expanded to 1 GB.

Each thread has one id, This will return... When the thread is created , So it's easy to get through id Operate on a thread . however goroutine There is no such concept , This is Go At the beginning of language design, it was decided to prevent abuse , So you can't be in a process “ Kill ” Another synergy , When coding, you need to consider when to create and release the collaboration .

The most important difference between coroutines and threads ：

• Thread switching needs to fall into the kernel , Then the context switches , The collaboration process is completed by the collaboration scheduler in the user state , There's no need to get into the kernel , In this way, the cost is small
• The switching time point of the coordination process is determined by the scheduler , Not determined by the system kernel , Although their switching points are time slices that exceed a certain threshold , Or enter I/O Or sleep
• Go Garbage collection is realized , But the necessary condition for garbage collection is that the memory is in a consistent state , Therefore, you need to pause all threads . If you leave it to the system , Then all threads will be paused to make them consistent ; But if Go in , The scheduler knows when the memory is in a consistent state , There is no need to pause all running threads .

13.2.2 establish goroutine

Just add go keyword , You can create concurrent execution units . Developers don't need to know any execution details , The scheduler will automatically schedule it to the appropriate system thread for execution .

Example

package main

import (
   "fmt"
   "time"
)

func Task1() {
    
   for {
    
      fmt.Println(time.Now().Format("15:04:05"), " Dealing with Task1 The task of ！")
      time.Sleep(time.Second * 3)
   }
}

func Task2() {
    
   for {
    
      fmt.Println(time.Now().Format("15:04:05"), " Dealing with Task2 The task of ！")
      time.Sleep(time.Second * 1)
   }
}

func main() {
    
   go Task1()
   go Task2()
   for {
    
      fmt.Println(time.Now().Format("15:04:05"), " Processing tasks of main process ！")
      time.Sleep(time.Second * 2)
   }
}

If you put go Keyword deletion , Obviously ,Task1 → Fall into a dead cycle

13.2.3 main function

When a program starts , Its main function is in a separate goroutine Run in , We call it main goroutine. new goroutine Will use go To create .

package main

import (
   "fmt"
   "time"
)

func Task1() {
    
   for {
    
      fmt.Println(time.Now().Format("15:04:05"), " Dealing with Task1 The task of ！")
      time.Sleep(time.Second * 3)
   }
}

func main() {
    
   go Task1()
   time.Sleep(time.Second * 10)
}

It can be seen that ,10 Seconds later , be-all goroutine stay main At the end of the function , It's over together .

goroutine Although similar to the thread concept , But the scheduling performance is not as detailed as that of threads , And the degree of detail depends on goroutine Implementation and running environment of scheduler . End goroutine The best way is to naturally return directly in the function .

13.2.4 Use anonymous functions to create goroutine

go func( parameter list ) {
    
     The body of the function 
} ( Call parameter list )

package main

import (
   "fmt"
   "time"
)

func main() {
    
   go func() {
    
      for {
    
         fmt.Println(time.Now().Format("15:04:05"), " Dealing with Task1 The task of ！")
         time.Sleep(time.Second * 3)
      }
   }()
   time.Sleep(time.Second * 8)
}

13.2.5 runtime package

Go In language runtime（ Runtime ） Package implements a small task scheduler . The working principle of this scheduler is similar to the system's scheduling of threads ,Go The language scheduler can efficiently integrate CPU Resources are allocated to each task .

1. Gosched()

   func Gosched()
   

   Gosched() Make current Go The coroutine abandons the processor , So that others Go Cooperation operation . It does not suspend the current Go coroutines , So the current Go The cooperation process will resume in the future .

   Go The coordination of language is preemptive scheduling , When encountering long-term execution or system call , Will take the initiative to goroutine Of CPU § Transfer out , Let others goroutine Can be scheduled and executed . Generally, the following situations occur ,goroutine Scheduling will occur ：

   • syscall

   • C Function call

   • Active call runtime.Gosched

   • Some goroutine The call time of is longer than 100ms, And this goroutine A non inline function was called

     【 Inline function refers to when the compiler finds that a piece of code is calling an inline function , It's not going to call the function , Instead, insert the entire code segment of the function into the current location . The advantage of this is that it eliminates the process of calling , Speed up the program .】

   Example

   package main
   
   import "fmt"
   
   func main() {
         
      go func() {
         
         for i := 0; i < 3; i++ {
         
            fmt.Println("go")
         }
      }()
   
      for i := 0; i < 2; i++ {
         
         fmt.Println("main")
      }
   }
   

   

   because main The second of the functions for The cycle has been preempted CPU control power , It didn't happen until the end of the operation goroutine Dispatch , therefore , first Go The coroutine anonymous function cannot get CPU control power , As a result, it cannot be implemented from beginning to end

   If you want to be the first for Loop execution , Can be in the second for Use in loop runtime.Gosched() To prevent it from gaining control

   package main
   
   import (
      "fmt"
      "runtime"
   )
   
   func main() {
         
      go func() {
         
         for i := 0; i < 3; i++ {
         
            fmt.Println("go")
         }
      }()
   
      for i := 0; i < 2; i++ {
         
         runtime.Gosched()
         fmt.Println("main")
      }
   }
   

   

2. Goexit()

   func Goexit()
   

   Goexit() Terminate the call to it Go coroutines , But other Go The coordination process will not be affected .Goexit() Will terminate the Go Perform all tasks before proceeding defer Function of .

   In the process of main go When the coroutine calls this function , Will end the Go coroutines , Instead of letting main return . This is because main Function did not return , The program will continue to execute other Go coroutines . If all the others Go Xie Cheng has quit , The program will crash .

   Example

   package main
   
   import (
      "fmt"
      "runtime"
      "time"
   )
   
   func Task1() {
         
      defer fmt.Println("task1 stop")
      fmt.Println("task1 start")
      fmt.Println("task1 work")
   }
   
   func Task2() {
         
      defer fmt.Println("task2 stop")
      fmt.Println("task2 start")
      runtime.Goexit() // Effect and return equally 
      fmt.Println("task2 work")
   }
   
   func main() {
         
      go Task1()
      go Task2()
      time.Sleep(time.Second * 5)
   }
   

   

3. GOMAXPROCS()

   GOMAXPROCS(n int) The function can set the CPU Count , It is most used in later programming .Go By default, language programs will use the maximum CPU Count .

   func GOMAXPROCS(n int) int
   

   GOMAXPROCS() Set the maximum that can be executed simultaneously CPU Count , And return to the previous settings . if n ＜ 1, It will not change the current settings . The logic of the local machine CPU The number can be passed through NumCPU Inquire about . This function will be removed after the scheduler is optimized .

   Example

   package main
   
   import (
      "fmt"
      "runtime"
      "time"
   )
   
   func main() {
         
      n := runtime.GOMAXPROCS(1)
      fmt.Println(" The previous CPU The number of cores is set to ：", n)
   
      last := time.Now()
      for i := 0; i < 100000; i++ {
         
         go func() {
         
            // Time consuming tasks 
            a := 999999 ^ 999999
            a = a + 1
         }()
      }
      now := time.Now()
      fmt.Println(now.Sub(last))
   }
   

   

   

13.3 channel

goroutine Run in the same address space , So access to shared memory must be synchronized . Reference type channel yes CSP The concrete embodiment of the mode , For more than one goroutine Communication between . Its internal synchronization , Ensure concurrency security .

13.3.1 channel type

channel It's a special type , and map similar ,channel It is also a corresponding make Reference to the underlying data structure created .

var  Channel variables  chan  Channel type 

Channel variables are reference variables that hold channels ; Channel type refers to the data type that the channel can transmit . When we copy one channel Or when passing function parameters , We just copied one channel quote , Therefore, both the caller and the callee will reference the same object . Like other reference types ,channel It's also zero nil.

Define a channel when , Also need definition to send to channel The type of value of .

make(chan Type) //  Equivalent to make(chan Type, 0)
make(chan Type, capacity)

When capacity by 0 when ,channel It's read and write without buffer ; When capacity Greater than 0 when ,channel Yes, there is buffer 、 Non blocking , Until writing is full. capacity Only elements block writing .

channel By the operator “＜-” To receive and send data .

channel <- value // send out value To channel
<-channel // Receive and discard 
x := <-channel // from channel Data received in , And assign it to x
x, ok := <-channel // ditto , And check whether the channel is closed , Assign this state to ok

By default ,channel Both receiving and sending data are blocked , Unless the other end is ready to receive , So that makes goroutine Synchronization of is simpler , Without explicit locks .

package main

import (
   "fmt"
   "time"
)

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

   go func() {
    
      fmt.Println(<-ch)
   }()
   ch <- "test"
   time.Sleep(time.Second)
}

The program defines and creates a transportable string Type of ch Channel variables , In anonymous coprocessor function , from ch Receive data in the channel and print .

13.3.2 Buffer mechanism

channel According to whether the buffer is supported or not, it can be divided into unbuffered channels （unbuffered channel） And buffered channels （buffered channel）.

Unbuffered channels are those that do not have the ability to hold any values before receiving .

This type of channel requires sending goroutine And receiving goroutine At the same time be ready to , To complete the sending and receiving operations . If two goroutine Not ready at the same time , It will cause the sending or receiving operation to be performed first goroutine Block waiting .

The interaction between sending and receiving channels is synchronous in itself , None of these operations can exist alone without the other operation .

Unbuffered channel Create format ：

make(chan Type) // Equivalent to make(chan Type, 0)

If no buffer capacity is specified , Then the channel is synchronized .

Example

package main

import (
   "fmt"
   "time"
)

func main() {
    
   ch := make(chan int, 0)
   go func() {
    
      for i := 0; i < 3; i++ {
    
         fmt.Printf("len(ch) = %v,cap(ch) = %v\n", len(ch), cap(ch))
         ch <- i
      }
   }()
   for i := 0; i < 3; i++ {
    
      time.Sleep(time.Second)
      fmt.Println(<-ch)
   }
}

The program creates an unbuffered channel ch, Because there is no buffer , So only when the receiver receives the data , The sender can continue to send data .

A buffered channel is a channel that can store one or more values before being received .

make(chan Type, capacity)

This type of channel is not mandatory goroutine Both receiving and sending must be completed at the same time . The conditions of channel blocking sending and receiving will also be different . Only if there is no value to receive in the channel , Receiving action will block . Only if the channel has no available buffer to hold the value being sent , Send action will block .

This leads to a big difference between buffered channels and unbuffered channels ： Unbuffered channels guarantee the ability to send and receive goroutine Data exchange will take place at the same time , There is no such guarantee for buffered channels .

package main

import (
   "fmt"
   "time"
)

func main() {
    
   ch := make(chan int, 3)
   go func() {
    
      for i := 0; i < 3; i++ {
    
         fmt.Printf("len(ch) = %v,cap(ch) = %v\n", len(ch), cap(ch))
         ch <- i
      }
   }()

   for i := 0; i < 3; i++ {
    
      time.Sleep(time.Second)
      fmt.Println(<-ch)
   }
}

Due to buffer （ Capacity of 3）, When the buffer is not full , The program will not be blocked

13.3.3 close and range

When the sender knows that there are no more values to send to channel when , It is necessary for the receiver to know in time that there is no more value to receive , Because in this way, the receiver can stop waiting unnecessarily . This can be done through the built-in close Functions and range Keyword to implement .

1. close

   Use close close channel Need to pay attention to ：

   • hannel It doesn't need to be closed as often as files , Only when you really don't have any data to send , Or want to end explicitly range Circulation and so on , To shut it down channel.
   • close channel after , Unable to contact channel Send data again , Sending again will cause panic error .
   • close channel after , You can continue from channel receive data .
   • about nil channel, Both receive and send will be blocked .

   Example

   package main
   
   import (
      "fmt"
   )
   
   func main() {
         
      ch := make(chan int, 3)
      go func() {
         
         for i := 0; i < 3; i++ {
         
            fmt.Printf("len= %v,cap=%v\n", len(ch), cap(ch))
            ch <- i
         }
         close(ch)
      }()
   
      for {
         
         if val, ok := <-ch; ok == true {
         
            fmt.Println(val)
         } else {
         
            return
         }
      }
   }
   

   

   After the sender sends the data , Use close Closed the channel , After the channel is closed ,ok Of false It will become false, Finally, the program ends

2. range

   Use range Keyword traversal channel

   package main
   
   import "fmt"
   
   func main() {
         
      ch := make(chan int, 3)
      go func() {
         
         for i := 0; i < 3; i++ {
         
            fmt.Printf("len=%v,cap=%v\n", len(ch), cap(ch))
            ch <- i
         }
         close(ch)
      }()
   
      for data := range ch {
         
         fmt.Println(data)
      }
   }
   

   

13.3.4 A one-way channel

By default , The channel is bidirectional , That is, you can send data to the channel , You can also receive data from the channel . however , Create a channel as a parameter to pass , Often want to use only one way , That is, either only send data , Or only receive data , At this time, you can specify the direction of the channel , That is, use a one-way channel .

A one-way channel Declaration of variables ：

var ch1 chan int // ch1 For a two-way channel 
var ch2 chan<- int // ch2 It is a one-way channel that can only receive 
var <-chan int // ch3 It is a one-way channel that can only be sent 

“chan＜-” Indicates that the data enters the channel , Just write the data into the channel , For the caller, it is output .“＜-chan” Indicates that the data comes out of the channel , For the caller, it is to get the data of the channel , That is input .

producer – Consumer example

package main

import "fmt"

func producer(out chan<- int) {
    
   for i := 0; i < 10; i++ {
    
      out <- i
   }
   close(out)
}

func consumer(in <-chan int) {
    
   for val := range in {
    
      fmt.Printf(" Consumer consumption %d\n", val)
   }
}

func main() {
    
   ch := make(chan int)
   go producer(ch)
   consumer(ch)
}

Ordinary two-way channel Implicitly convert to one-way channel, One way can't be channel Convert to bidirectional channel.

13.3.5 Timer

stay Go Language standard library time In bag , The implementation of timer uses one-way channel.

Use the timer to print a string of characters at regular intervals

package main

import (
   "fmt"
   "time"
)

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

   for {
    
      <-ticker.C
      fmt.Println("loop")
   }
}

every other 1s

13.4 select

Go In language , By keyword select Can monitor channel Data flow on .select Usage and switch Very similar , from select Start a new selection block , Each selection is made by case To describe .

13.4.1 select effect

And switch Statement can select any condition that can use equality comparison ,select There are more restrictions , The biggest limitation is each case There must be a I/O operation .

select {
    
case <-chan1:
    //  If chan1 Successfully read the data , Then execute the case sentence 
case chan2 <- 1:
    //  If we succeed in chan2 Write data , Then execute the case sentence 
default:
    // If the top case None of them were successfully implemented , Then execute the default sentence 
}

In a select In the sentence ,Go The language evaluates each send and receive statement in order from beginning to end , If any of these statements can continue （ It's not blocked ）, Then randomly choose one of those executable statements to use .

If no statement can be executed （ That is, all channels are blocked ）, It will be executed by default default sentence , At the same time, the execution of the program will start from select Restore... In the statement after the statement . without default sentence , be select The statement will be blocked , Until there is one channel It can go on .

package main

import (
   "fmt"
   "time"
)

func main() {
    
   ch := make(chan int)
   go func() {
    
      for i := 0; i < 3; i++ {
    
         ch <- i
      }
   }()

   for {
    
      select {
    
      case msg := <-ch:
         fmt.Println(msg)
      default:
         time.Sleep(time.Second)
      }
   }
}

I've been waiting for

13.4.2 Overtime

Sometimes there will be goroutine Blocking condition , In order to prevent the program from blocking for a long time , We can use select To implement the blocking timeout mechanism .

package main

import (
   "fmt"
   "time"
)

func main() {
    
   ch := make(chan int)
   done := make(chan bool)

   go func() {
    
      for {
    
         select {
    
         case val := <-ch:
            fmt.Println(val)
         case <-time.After(time.Second * 3):
            fmt.Println(" Timeout 3 second ")
            done <- true
         }
      }
   }()

   for i := 0; i < 10; i++ {
    
      ch <- i
   }

   <-done
   fmt.Println(" Termination of procedure ")
}

13.4.3 Deadlock

You may encounter deadlocks when writing concurrent programs . What is a deadlock ？ That is, all threads or processes are waiting for the release of resources .

package main

func main() {
    
    ch := make(chan int)
    <- ch //  Blocking main goroutine,  channel ch Locked 
}

There is only one coroutine in the program , Add data or save data , Will lock the channel , And block the current process . That's all goroutine（ In fact, only main A thread ） Are waiting for the opening of the channel （ If no one takes the data , The channel will not be open ）, This creates a deadlock .

If an inflow occurs on an unbuffered channel, there is no outflow , Or outflow without inflow , Will cause deadlock . similarly , Use select keyword , There will also be a deadlock without adding any code .

package main

func main() {
    
   select {
    }
}