Comprehensive application of Go coroutines and channels

Comprehensive application issues of Go coroutines and channels

2024-07-12

1. Briefly understand what coroutines and channels are

What is a coroutine?

A coroutine is a user-level lightweight thread that has its own stack space and shares the program's heap space.

It is a microthread implemented by an algorithm based on a single thread. Compared with multi-threaded programming, it has the following advantages:

The context switching of the coroutine is determined by the user, and there is no need for the context switching of the system kernel, which reduces the overhead
By default, coroutines are fully protected to prevent interruption. No atomic locks are required.
A single thread can also achieve high concurrency, and even a single-core CPU can support tens of thousands of coroutines.

What is a channel

A channel is a data structure used for communication between coroutines. It is similar to a queue, with one end being the sender and the other end being the receiver. Using channels can ensure the synchronization and order of data.

Channels are divided into buffered channels and unbuffered channels, which are declared as follows:

Buffered channel

intChan := make(chan int,<缓冲容量>)

Unbuffered Channel

intChan := make(chan int)

The difference between buffered channels and unbuffered channels:

Blocking: The sender of an unbuffered channel blocks until data is received; the sender of a buffered channel blocks until the buffer is full, and the receiver blocks until the buffer is not empty.
Data synchronization and order: Unbuffered channels guarantee data synchronization and order; buffered channels do not guarantee data synchronization and order.
Application scenarios: Unbuffered channels require strict synchronization and sequentiality; buffered channels allow asynchronous communication and improve throughput.

It should be noted that in the implementation of unbuffered channels, there must be a sender and a receiver at both ends of the channel, otherwise a deadlock will occur.

2. Coroutine-channel concurrent programming case

(1) Print letters and numbers alternately

Topic: Use coroutine-channel to print numbers 1-10 and letters AJ alternately.

Code:


package main
 
import (
	"fmt"
	"sync"
)
 
/*
无缓冲chanel：需要在写入chanel的时候要保证有另外一个协程在读取chanel。否则会导致写端阻塞，发生死锁
解决办法：
避免死锁的发生：
当i循环到10时，printAlp协程已然结束，所以此时不必再写入alp通道
*/
 
func printNum(wg *sync.WaitGroup, numCh chan struct{}, alpCh chan struct{}) {
	defer wg.Done()
 
	for i := 1; i <= 10; i++ {
		<-alpCh // 等待字母goroutine发信号
		fmt.Print(i, " ")
		//避免死锁发生
		if i < 10 {
			numCh <- struct{}{} // 发信号给字母goroutine
		}
		if i == 10 {
			close(numCh)
		}
	}
 
}
 
func printAlp(wg *sync.WaitGroup, numCh chan struct{}, alpCh chan struct{}) {
	defer wg.Done()
 
	for i := 'A'; i <= 'J'; i++ {
		<-numCh // 等待数字goroutine发信号
		fmt.Printf("%c", i)
		alpCh <- struct{}{} // 发信号给数字goroutine
	}
	close(alpCh)
}
 
func main() {
	numCh := make(chan struct{}) // 用于数字goroutine的信号通道
	alpCh := make(chan struct{}) // 用于字母goroutine的信号通道
	var wg sync.WaitGroup
 
	wg.Add(2)
 
	go printAlp(&wg, numCh, alpCh)
	go printNum(&wg, numCh, alpCh)
 
	// 启动时先给数字goroutine发送一个信号
	numCh <- struct{}{}
 
	wg.Wait()
 
}

Topic Analysis:

The question requires us to print letters and numbers alternately, which requires us to ensure the strict order of the two coroutines, in line with the application scenario of unbuffered channels. Set up two channels to store numbers and letters respectively. The two coroutines that print numbers and letters serve as senders and receivers of the two channels respectively. Print once in a loop and send a signal to remind the other coroutine to print.

It should be noted that when the last character '10' is printed, the coroutine for printing letters has ended, and the numCh channel has no receiver. At this time, it no longer meets the implementation conditions of the unbuffered channel - there must be a sender and a receiver. Sending a signal again will cause blocking deadlock. Therefore, there is no need to send a signal for the 10th time.

(2) Design a task scheduler

Title: Design a task scheduler that uses the multi-coroutine + channel programming model to implement business scenarios that process multiple tasks concurrently, and requires that the scheduling order be based on the order in which the tasks are added.

Code:


type scheduler struct {
	taskChan chan func()
	wt       sync.WaitGroup
}
 
func (td *scheduler) AddTask(task func()) {
	td.taskChan <- task
}
 
func (td *scheduler) Executer() {
	defer td.wt.Done()
	for {
		task, ok := <-td.taskChan
		task()
		if ok && len(td.taskChan) == 0 {
			break
		}
	}
}
 
func (td *scheduler) Start() {
	td.wt.Add(4)
	//假设四个消费者
	for i := 0; i < 4; i++ {
		go td.Executer()
	}
 
	td.wt.Wait()
}
 
func main() {
	sd := scheduler{
		taskChan: make(chan func(), 5),
	}
 
	go func() {
		sd.AddTask(func() {
			fmt.Println("任务1")
		})
		sd.AddTask(func() {
			fmt.Println("任务2")
		})
		sd.AddTask(func() {
			fmt.Println("任务3")
		})
		sd.AddTask(func() {
			fmt.Println("任务4")
		})
		sd.AddTask(func() {
			fmt.Println("任务5")
		})
		sd.AddTask(func() {
			fmt.Println("任务6")
		})
		close(sd.taskChan)
	}()
 
	sd.Start()
 
}

problem analysis:

Since the added tasks are multiple tasks, more than one, and asynchronous processing is required to execute these tasks, the buffered channel needs to improve throughput and asynchronous processing.

Then, we need to put the tasks into the channel, and multiple receivers can take the tasks from the channel in order and execute them.

It should be noted that if the number of tasks added is greater than the channel buffer, it will be blocked when adding tasks. In order not to affect the normal startup of the consumer, it is necessary to open a separate coroutine to add tasks.

In this way, when the consumer consumes, the blocked producer will be awakened to continue adding tasks.

3. Summary

After learning the coroutine + channel programming model, in addition to what was just mentioned in the title, we should also pay attention to the following issues:

1. Why should the channel be closed after use? What are the risks of not closing it?

To avoid deadlock, closing the channel also tells the receiver that there is no data to be sent from the sender and it does not need to wait for data any more. After receiving the information that the channel is closed, the receiver stops receiving data. If the channel is not closed, the receiver will be blocked all the time, which may cause deadlock.
Release resources and avoid resource leakage. After closing the channel, the system will release the corresponding resources. Closing the channel in time can avoid resource waste and leakage.

2. How to close the channel gracefully?

First of all, the most basic principle of closing a channel is not to close a channel that has already been closed. Secondly, there is another principle for using Go channels:Do not close the channel on the receiver side or when there are multiple senders.in other words，We should only allow the sole sender on a channel to close the channel.

A crude way is to close the channel by exception recovery, but it obviously violates the above principles and may cause data competition; another way is to close the channel by sync.Once or sync.Mutex, but it does not guarantee that concurrent closing operations and sending operations on a channel will not cause data competition. Both methods have certain problems, so I will not introduce them in detail. The following is a method of how to close the channel gracefully.

Case 1: M receivers and one sender

This is the easiest case to handle. When the sender needs to finish sending, just let it close the channel. The two programming examples above are such cases.

Scenario 2: One Receiver and N Senders

According to the basic principle of Go channel, we can only close the channel when it is the only sender. So, in this case, we can't close the channel directly somewhere.But we can let the receiver close an additional signal channel to tell the sender not to send any more data。


package main
 
import (
	"log"
	"sync"
)
 
func main() {
 
	cosnt N := 5
	cosnt Max := 60000
	count := 0
 
	dataCh := make(chan int)
	stopCh := make(chan bool)
 
	var wt sync.WaitGroup
	wt.Add(1)
 
	//发送者
	for i := 0; i < N; i++ {
		go func() {
			for {
				select {
				case <-stopCh:
					return
				default:
					count += 1
					dataCh <- count
				}
			}
		}()
	}
 
	//接收者
	go func() {
		defer wt.Done()
		for value := range dataCh {
			if value == Max {
				// 此唯一的接收者同时也是stopCh通道的
				// 唯一发送者。尽管它不能安全地关闭dataCh数
				// 据通道，但它可以安全地关闭stopCh通道。
				close(stopCh)
				return
			}
			log.Println(value)
		}
	}()
 
	wt.Wait()
}

In this method, we add an additional signal channel stopCh, which the receiver uses to tell the sender that it does not need to receive any more data. In addition, this method does not close dataCh. When a channel is no longer used by any coroutine, it will gradually be garbage collected, regardless of whether it has been closed.

The elegance of this approach lies in the fact that closing one channel stops the use of the other channel, thereby indirectly closing the other channel.

Scenario 3: M receivers and N senders

We cannot allow any of the receiver and sender to close the channel used to transmit data, nor can we allow one of multiple receivers to close an additional signal channel. Both practices violate the channel closing principle.

However, we can introduceAn intermediate mediator role and closes additional signal channels to notify all receivers and senders that the work is finished。

Code example:


package main
 
import (
	"log"
	"math/rand"
	"strconv"
	"sync"
)
 
func main() {
 
	const Max = 100000
	const NumReceivers = 10
	const NumSenders = 1000
 
	var wt sync.WaitGroup
	wt.Add(NumReceivers)
 
	dataCh := make(chan int)
	stopCh := make(chan struct{})
	// stopCh是一个额外的信号通道。它的发送
	// 者为中间调解者。它的接收者为dataCh
	// 数据通道的所有的发送者和接收者。
	toStop := make(chan string, 1)
	// toStop是一个用来通知中间调解者让其
	// 关闭信号通道stopCh的第二个信号通道。
	// 此第二个信号通道的发送者为dataCh数据
	// 通道的所有的发送者和接收者，它的接收者
	// 为中间调解者。它必须为一个缓冲通道。
 
	var stoppedBy string
 
	// 中间调解者
	go func() {
		stoppedBy = <-toStop
		close(stopCh)
	}()
 
	// 发送者
	for i := 0; i < NumSenders; i++ {
		go func(id string) {
			for {
				value := rand.Intn(Max)
				if value == 0 {
					// 为了防止阻塞，这里使用了一个尝试
					// 发送操作来向中间调解者发送信号。
					select {
					case toStop <- "发送者#" + id:
					default:
					}
					return
				}
 
				select {
				case <-stopCh:
					return
				case dataCh <- value:
				}
			}
		}(strconv.Itoa(i))
	}
 
	// 接收者
	for i := 0; i < NumReceivers; i++ {
		go func(id string) {
			defer wt.Done()
 
			for {
				select {
				case <-stopCh:
					return
				case value := <-dataCh:
					if value == Max {
						// 为了防止阻塞，这里使用了一个尝试
						// 发送操作来向中间调解者发送信号。
						select {
						case toStop <- "接收者:" + id:
						default:
						}
						return
					}
 
					log.Println(value)
				}
			}
		}(strconv.Itoa(i))
	}
 
	wt.Wait()
	log.Println("被" + stoppedBy + "终止了")
 
}

Technology Sharing