Problemas completos de aplicación de corrutinas y canales de Go

2024-07-12

1. Comprenda brevemente qué son las corrutinas y los canales.

¿Qué es la corrutina?

Una corrutina es un subproceso liviano a nivel de usuario que tiene un espacio de pila independiente y comparte el espacio de pila del programa.

Es un microhilo implementado mediante algoritmos basados en un solo subproceso. En comparación con la programación multiproceso, tiene las siguientes ventajas:

El cambio de contexto de la corrutina lo decide el usuario, sin necesidad de cambiar de contexto del núcleo del sistema, lo que reduce la sobrecarga.
De forma predeterminada, las corrutinas están completamente protegidas para evitar interrupciones.No se requiere bloqueo de operación atómica
Un solo subproceso también puede lograr una alta concurrencia, e incluso una CPU de un solo núcleo puede admitir decenas de miles de corrutinas.

¿Qué es el canal?

El canal es una estructura de datos utilizada para la comunicación entre corrutinas. Similar a una cola, un extremo es el remitente y el otro extremo es el receptor. El uso de canales puede garantizar la sincronización y el orden de los datos.

Los canales se dividen en canales con búfer y canales sin búfer, que se declaran de la siguiente manera:

Hay un canal de buffer

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

canal sin búfer

intChan := make(chan int)

La diferencia entre canales con búfer y canales sin búfer:

Bloqueo: el remitente de un canal sin búfer se bloqueará hasta que se reciban los datos; el remitente de un canal con búfer se bloqueará hasta que el búfer esté lleno y el receptor se bloqueará hasta que el búfer no esté vacío.
Sincronización y secuencia de datos: los canales sin búfer garantizan la sincronización y secuencia de datos; las tuberías con búfer no garantizan la sincronización y secuencia de datos.
Escenarios de aplicación: los canales sin búfer requieren una sincronización estricta y los canales con búfer pueden comunicarse de forma asíncrona y mejorar el rendimiento;

Lo que hay que tener en cuenta en la implementación de canales sin búfer es que debe haber un remitente y un receptor en ambos extremos del canal; de lo contrario, se producirá un punto muerto.

2. Caso de programación concurrente de canal de rutina

(1) Imprima letras y números alternativamente

Significado de la pregunta: utilice el canal de rutina para imprimir alternativamente los números del 1 al 10 y las 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álisis del tema:

La pregunta requiere que imprimamos letras y números alternativamente, por lo que debemos garantizar el orden estricto de las dos corrutinas, que es consistente con el escenario de aplicación de canales sin búfer. Configure dos canales para almacenar números y letras respectivamente. Dos corrutinas que imprimen números y letras sirven como remitente y receptor de los dos canales respectivamente. Imprima una vez en un bucle y envíe una señal una vez para recordarle a otra rutina que imprima.

Cabe señalar que cuando se imprime el último carácter '10', la rutina para imprimir letras ha finalizado y el canal numCh no tiene receptor. En este momento, las condiciones de implementación del canal sin búfer ya no se cumplen; debe haberlas. un remitente y un receptor, de lo contrario, enviar la señal nuevamente provocará un punto muerto de bloqueo. Por lo tanto, no es necesario volver a enviar la señal por décima vez.

(2) Diseñar un programador de tareas

Título: Diseñe un programador de tareas que utilice el modelo de programación de múltiples corrutinas + canales para implementar escenarios comerciales para el procesamiento simultáneo de múltiples tareas, y el orden de programación debe estar en el orden en que se agregan las tareas.

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

análisis del problema:

Dado que las tareas agregadas son múltiples, hay más de una y se requiere procesamiento asincrónico para ejecutar estas tareas. Cumplir con los canales almacenados en búfer requiere un rendimiento mejorado y un procesamiento asincrónico.

Luego, debemos colocar la tarea en el canal, y varios receptores pueden tomar las tareas del canal en orden y ejecutarlas.

Una cosa a tener en cuenta es que si la cantidad de tareas agregadas es mayor que el búfer del canal, se bloqueará al agregar tareas. Para no afectar el inicio normal del consumidor, es necesario abrir una rutina separada para agregar tareas.

De esta forma, cuando el consumidor consuma, el productor bloqueador se despertará para continuar agregando tareas.

3. Resumen

Después de estudiar el modelo de programación de canales corrutina +, además de lo que se acaba de mencionar en el título, también debemos prestar atención a las siguientes cuestiones:

1. ¿Por qué debería cerrarse el canal una vez agotado? ¿Cuáles son los riesgos de no cerrarlo?

Para evitar un punto muerto. Cerrar el canal también le dice al receptor que no hay más datos para enviar desde el remitente y que no hay necesidad de seguir esperando datos. Después de recibir la información de cierre del canal, el receptor deja de recibir datos; si el canal no está cerrado, el receptor permanecerá bloqueado y existe riesgo de bloqueo.
Libere recursos y evite fugas de recursos. Después de cerrar el canal, el sistema liberará los recursos correspondientes. Cerrar el canal a tiempo puede evitar el desperdicio y la fuga de recursos.

2. ¿Cómo cerrar el canal con gracia?

En primer lugar, el principio más básico para cerrar canales es no cerrar canales que ya se han cerrado. En segundo lugar, existe otro principio para utilizar canales Go:No cierre el canal en el receptor de datos o cuando haya varios remitentes.en otras palabras，Sólo debemos permitir que el único remitente de un canal cierre este canal.

Una forma grosera es cerrar el canal mediante la recuperación de excepciones, pero esto obviamente viola los principios anteriores y puede causar una carrera de datos. Otra forma es cerrar el canal con sync.Once o sync.Mutex, lo cual no está garantizado que suceda. Las operaciones y las operaciones de envío en un canal no crean carreras de datos. Ambos métodos tienen ciertos problemas, por lo que no los presentaré en detalle. Aquí hay un método sobre cómo cerrar el canal correctamente.

Escenario 1: M receptores y un remitente

Una de las situaciones más fáciles de afrontar. Cuando el remitente necesite terminar de enviar, simplemente déjelo cerrar el canal. Este es el caso de los dos ejemplos de programación anteriores.

Escenario 2: un receptor y N remitentes

De acuerdo con los principios básicos de los canales Go, solo podemos cerrar el canal en el único remitente del canal. Entonces, en este caso, no podemos cerrar el canal directamente en alguna parte.Pero podemos dejar que el receptor cierre un canal de señal adicional para indicarle al remitente que no envíe más datos.。


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

En este método, agregamos un canal de señal adicional stopCh, que el receptor usa para decirle al remitente que ya no necesita recibir datos. Además, este método no cierra dataCh. Cuando una rutina ya no utiliza un canal, se recolectará basura gradualmente, independientemente de si se ha cerrado.

La elegancia de este método es que al cerrar un canal, dejas de usar otro canal, cerrando así indirectamente el otro canal.

Escenario 3: M receptores y N remitentes

No podemos hacer que ni el receptor ni el emisor cierren el canal utilizado para transmitir los datos, ni podemos hacer que uno de los múltiples receptores cierre un canal de señalización adicional. Ambas prácticas violan el principio de cierre de canales.

Sin embargo, podemos introducirUn papel de mediador intermedio y que cierre canales de señalización adicionales para avisar a todos los receptores y remitentes del final del trabajo.。

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

Compartir tecnología