Problemas abrangentes de aplicação de corrotinas e canais Go

2024-07-12

1. Entenda resumidamente o que são corrotinas e canais

O que é co-rotina

Uma corrotina é um thread leve em nível de usuário que possui um espaço de pilha independente e compartilha o espaço de heap do programa.

É um microthread implementado por meio de algoritmos baseados em thread único. Em comparação com a programação multithread, possui as seguintes vantagens:

A troca de contexto da corrotina é decidida pelo usuário, sem a necessidade de troca de contexto do kernel do sistema, reduzindo a sobrecarga.
Por padrão, as corrotinas são totalmente protegidas para evitar interrupções.Não é necessário bloqueio de operação atômica
Um único thread também pode atingir alta simultaneidade, e até mesmo uma CPU de núcleo único pode suportar dezenas de milhares de corrotinas.

o que é canal

Canal é uma estrutura de dados usada para comunicação entre corrotinas. Semelhante a uma fila, uma extremidade é o remetente e a outra extremidade é o destinatário. O uso de canais pode garantir a sincronização e a ordem dos dados.

Os canais são divididos em canais com buffer e canais sem buffer, que são declarados da seguinte forma:

Existe um canal de buffer

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

canal sem buffer

intChan := make(chan int)

A diferença entre canais com buffer e canais sem buffer:

Bloqueio: O remetente de um canal sem buffer bloqueará até que os dados sejam recebidos; o remetente de um canal com buffer bloqueará até que o buffer esteja cheio e o receptor bloqueará até que o buffer não esteja vazio.
Sincronização e sequência de dados: canais sem buffer garantem sincronização e sequência de dados;
Cenários de aplicação: Canais sem buffer exigem sincronização e sequência rigorosas; canais com buffer podem se comunicar de forma assíncrona e melhorar o rendimento;

O que precisa ser observado na implementação de canais sem buffer é que deve haver um remetente e um receptor em ambas as extremidades do canal, caso contrário ocorrerá um deadlock.

2. Caso de programação simultânea de canal de rotina

(1) Imprima letras e números alternadamente

Significado da pergunta: Use o canal de co-rotina para imprimir alternadamente os números de 1 a 10 e as letras AJ.

Código:


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()
 
}

Análise do tópico:

A questão exige que imprimamos letras e números alternadamente, por isso precisamos garantir a ordem estrita das duas corrotinas, o que é consistente com o cenário de aplicação de canais sem buffer. Configure dois canais para armazenar números e letras, respectivamente. Duas corrotinas que imprimem números e letras servem como remetente e destinatário dos dois canais, respectivamente. Imprima uma vez em um loop e envie um sinal uma vez para lembrar outra corrotina de imprimir.

Deve-se notar que quando o último caractere '10' é impresso, a rotina de impressão de letras terminou e o canal numCh não possui receptor. Neste momento, as condições de implementação do canal sem buffer não são mais atendidas - deve haver. um remetente e um receptor, caso contrário, enviar o sinal novamente causará um impasse de bloqueio. Portanto, não há necessidade de enviar o sinal novamente pela décima vez.

(2) Projete um agendador de tarefas

Título: Projete um agendador de tarefas que use o modelo de programação multi-rotina + canal para implementar cenários de negócios de processamento simultâneo de multitarefas e exija que a ordem de agendamento esteja na ordem em que as tarefas são adicionadas.

Código:


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()
 
}

analise de problemas:

Como as tarefas adicionadas são multitarefas, há mais de uma e é necessário processamento assíncrono para executar essas tarefas. A conformidade com canais em buffer requer melhor rendimento e processamento assíncrono.

Então, precisamos colocar a tarefa no canal, e vários receptores podem pegar as tarefas do canal em ordem e executá-las.

O problema que precisa de atenção é que se o número de tarefas adicionadas for maior que o buffer do canal, causará bloqueio na adição de tarefas. Para não afetar a inicialização normal do consumidor, ele precisa abrir uma corrotina separada para adicionar tarefas.

Desta forma, quando o consumidor consumir, o produtor bloqueador será despertado para continuar adicionando tarefas.

3. Resumo

Depois de estudar o modelo de programação de corrotina + canal, além do que acabamos de mencionar no título, devemos também ficar atentos às seguintes questões:

1. Por que o canal deve ser fechado depois de esgotado? Quais são os riscos de não fechá-lo?

Para evitar impasse. Fechar o canal também informa ao receptor que não há mais dados para enviar do remetente e que não há necessidade de continuar aguardando pelos dados. Após receber a informação de fechamento do canal, o receptor para de receber dados; se o canal não for fechado, o receptor permanecerá bloqueado e há risco de deadlock;
Libere recursos e evite vazamentos de recursos. Após fechar o canal, o sistema liberará os recursos correspondentes. Fechar o canal a tempo pode evitar desperdício e vazamento de recursos.

2. Como fechar o canal normalmente?

Em primeiro lugar, o princípio mais básico do encerramento de canais é não fechar canais que foram fechados. Em segundo lugar, existe outro princípio para usar canais Go:Não feche o canal no receptor de dados ou quando houver vários remetentes.em outras palavras，Devemos permitir apenas que o único remetente de um canal feche este canal.

Uma maneira rude é fechar o canal por meio de recuperação de exceção, mas isso obviamente viola os princípios acima e pode causar corrida de dados. Outra maneira é fechar o canal com sync.Once ou sync.Mutex, o que não é garantido. operações e operações de envio em um canal não criam corridas de dados. Ambos os métodos têm alguns problemas, então não vou apresentá-los em detalhes. Aqui está um método sobre como fechar o canal normalmente.

Cenário 1: M receptores e um remetente

Uma das situações mais fáceis de lidar. Quando o remetente precisar finalizar o envio, basta deixá-lo fechar o canal. Este é o caso dos dois exemplos de programação acima.

Cenário 2: Um receptor e N remetentes

De acordo com os princípios básicos dos canais Go, só podemos fechar o canal no único remetente do canal. Portanto, neste caso, não podemos fechar o canal diretamente em algum lugar.Mas podemos deixar o receptor fechar um canal de sinal adicional para dizer ao remetente para não enviar mais dados.。


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()
}

Neste método, adicionamos um canal de sinal adicional stopCh, que o receptor usa para informar ao remetente que não precisa mais receber dados. Além disso, este método não fecha o dataCh. Quando um canal não é mais usado por nenhuma corrotina, ele será gradualmente coletado como lixo, independentemente de ter sido fechado.

A elegância desse método é que, ao fechar um canal, você deixa de usar outro canal, fechando indiretamente o outro canal.

Cenário 3: M receptores e N remetentes

Não podemos fazer com que o receptor ou o remetente fechem o canal usado para transmitir os dados, nem podemos fazer com que um dos múltiplos receptores feche um canal de sinalização adicional. Ambas as práticas violam o princípio do encerramento do canal.

Contudo, podemos apresentarUma função de mediador intermediário e fechamento de canais de sinalização adicionais para notificar todos os receptores e remetentes do fim do trabalho。

Exemplo de código:


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 + "终止了")
 
}

Compartilhamento de tecnologia