Обмен технологиями

Комплексные вопросы применения сопрограмм и каналов Go.

2024-07-12

한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina

1. Кратко разобраться, что такое сопрограммы и каналы

Что такое сопрограмма

Сопрограмма — это легкий поток пользовательского уровня, который имеет независимое пространство стека и разделяет пространство кучи программы.

Это микропоток, реализованный с помощью алгоритмов на основе однопоточного программирования. По сравнению с многопоточным программированием он имеет следующие преимущества:

  • Переключение контекста сопрограммы определяется пользователем без необходимости переключения контекста ядра системы, что снижает накладные расходы.
  • По умолчанию сопрограммы полностью защищены от прерываний.Не требуется атомная блокировка операций
  • Один поток также может обеспечить высокий уровень параллелизма, и даже одноядерный процессор может поддерживать десятки тысяч сопрограмм.

что такое канал

Канал — это структура данных, используемая для связи между сопрограммами. Подобно очереди, один конец — отправитель, а другой — получатель. Использование каналов может обеспечить синхронизацию и порядок данных.

Каналы делятся на буферизованные и небуферизованные каналы, которые объявляются следующим образом:

  • Есть буферный канал
intChan := make(chan int,<缓冲容量>)
  • небуферизованный канал
intChan := make(chan int)

Разница между буферизованными и небуферизованными каналами:

  • Блокировка: отправитель небуферизованного канала будет блокироваться до тех пор, пока не будут получены данные; отправитель буферизованного канала будет блокироваться до тех пор, пока буфер не заполнится, а получатель будет блокироваться до тех пор, пока буфер не станет пустым.
  • Синхронизация и последовательность данных. Небуферизованные каналы гарантируют синхронизацию и последовательность данных, а буферизованные каналы не гарантируют синхронизацию и последовательность данных.
  • Сценарии применения: небуферизованные каналы требуют строгой синхронизации, а последовательность буферизованных каналов может взаимодействовать асинхронно и повышать пропускную способность;

Что необходимо отметить при реализации небуферизованных каналов, так это то, что на обоих концах канала должны быть отправитель и получатель, иначе произойдет взаимоблокировка.

2. Случай параллельного программирования каналов сопрограммы

(1) Печатайте буквы и цифры попеременно.

Значение вопроса: используйте канал сопрограммы для поочередной печати цифр 1-10 и букв AJ.

Код:

  1. package main
  2.  
  3. import (
  4. 	"fmt"
  5. 	"sync"
  6. )
  7.  
  8. /*
  9. 无缓冲chanel：需要在写入chanel的时候要保证有另外一个协程在读取chanel。否则会导致写端阻塞，发生死锁
  10. 解决办法：
  11. 避免死锁的发生：
  12. 当i循环到10时，printAlp协程已然结束，所以此时不必再写入alp通道
  13. */
  14.  
  15. func printNum(wg *sync.WaitGroup, numCh chan struct{}, alpCh chan struct{}) {
  16. 	defer wg.Done()
  17.  
  18. 	for i := 1; i <= 10; i++ {
  19. 		<-alpCh // 等待字母goroutine发信号
  20. 		fmt.Print(i, " ")
  21. 		//避免死锁发生
  22. 		if i < 10 {
  23. 			numCh <- struct{}{} // 发信号给字母goroutine
  24. 		}
  25. 		if i == 10 {
  26. 			close(numCh)
  27. 		}
  28. 	}
  29.  
  30. }
  31.  
  32. func printAlp(wg *sync.WaitGroup, numCh chan struct{}, alpCh chan struct{}) {
  33. 	defer wg.Done()
  34.  
  35. 	for i := 'A'; i <= 'J'; i++ {
  36. 		<-numCh // 等待数字goroutine发信号
  37. 		fmt.Printf("%c", i)
  38. 		alpCh <- struct{}{} // 发信号给数字goroutine
  39. 	}
  40. 	close(alpCh)
  41. }
  42.  
  43. func main() {
  44. 	numCh := make(chan struct{}) // 用于数字goroutine的信号通道
  45. 	alpCh := make(chan struct{}) // 用于字母goroutine的信号通道
  46. 	var wg sync.WaitGroup
  47.  
  48. 	wg.Add(2)
  49.  
  50. 	go printAlp(&wg, numCh, alpCh)
  51. 	go printNum(&wg, numCh, alpCh)
  52.  
  53. 	// 启动时先给数字goroutine发送一个信号
  54. 	numCh <- struct{}{}
  55.  
  56. 	wg.Wait()
  57.  
  58. }

Анализ темы:

Вопрос требует, чтобы мы поочередно выводили буквы и цифры, поэтому нам необходимо обеспечить строгий порядок двух сопрограмм, который соответствует сценарию применения небуферизованных каналов. Настройте два канала для хранения цифр и букв соответственно. Две сопрограммы, которые печатают цифры и буквы, служат отправителем и получателем двух каналов соответственно. Напечатайте один раз в цикле и один раз отправьте сигнал, чтобы напомнить другой сопрограмме о необходимости печати.

Следует отметить, что при печати последнего символа «10» сопрограмма печати букв завершилась, а у канала numCh нет приемника. В это время условия реализации небуферизованного канала уже не выполняются — они должны быть. отправитель и получатель. В противном случае повторная отправка сигнала приведет к взаимоблокировке блокировки. Таким образом, нет необходимости отправлять сигнал еще раз в 10-й раз.

(2) Создайте планировщик задач.

Название: Разработка планировщика задач, который использует модель многосопрограммного + канального программирования для реализации бизнес-сценариев для одновременной обработки нескольких задач, при этом порядок планирования должен соответствовать порядку добавления задач.

Код:

  1. type scheduler struct {
  2. 	taskChan chan func()
  3. 	wt       sync.WaitGroup
  4. }
  5.  
  6. func (td *scheduler) AddTask(task func()) {
  7. 	td.taskChan <- task
  8. }
  9.  
  10. func (td *scheduler) Executer() {
  11. 	defer td.wt.Done()
  12. 	for {
  13. 		task, ok := <-td.taskChan
  14. 		task()
  15. 		if ok && len(td.taskChan) == 0 {
  16. 			break
  17. 		}
  18. 	}
  19. }
  20.  
  21. func (td *scheduler) Start() {
  22. 	td.wt.Add(4)
  23. 	//假设四个消费者
  24. 	for i := 0; i < 4; i++ {
  25. 		go td.Executer()
  26. 	}
  27.  
  28. 	td.wt.Wait()
  29. }
  30.  
  31. func main() {
  32. 	sd := scheduler{
  33. 		taskChan: make(chan func(), 5),
  34. 	}
  35.  
  36. 	go func() {
  37. 		sd.AddTask(func() {
  38. 			fmt.Println("任务1")
  39. 		})
  40. 		sd.AddTask(func() {
  41. 			fmt.Println("任务2")
  42. 		})
  43. 		sd.AddTask(func() {
  44. 			fmt.Println("任务3")
  45. 		})
  46. 		sd.AddTask(func() {
  47. 			fmt.Println("任务4")
  48. 		})
  49. 		sd.AddTask(func() {
  50. 			fmt.Println("任务5")
  51. 		})
  52. 		sd.AddTask(func() {
  53. 			fmt.Println("任务6")
  54. 		})
  55. 		close(sd.taskChan)
  56. 	}()
  57.  
  58. 	sd.Start()
  59.  
  60. }

анализ проблемы:

Поскольку добавленные задачи являются многозадачными, их несколько, и для выполнения этих задач требуется асинхронная обработка. Соответствие буферизованным каналам требует повышения пропускной способности и асинхронной обработки.

Затем нам нужно поместить задачу в канал, и несколько получателей смогут по порядку брать задачи из канала и выполнять их.

Следует отметить, что если количество добавленных задач превышает буфер канала, это приведет к блокировке при добавлении задач. Чтобы не влиять на нормальный запуск потребителя, ему нужно открыть отдельную сопрограмму для добавления задач.

Таким образом, когда потребитель потребляет, производитель блокировки будет пробужден для продолжения добавления задач.

3. Резюме

После изучения модели программирования сопрограмма + канал, помимо того, что было только что упомянуто в заголовке, нам следует также обратить внимание на следующие вопросы:

1. Почему канал должен быть закрыт после того, как он исчерпан? Какие риски, если его не закрыть?

  • Чтобы избежать тупика. Закрытие канала также сообщает получателю, что от отправителя больше нет данных для отправки и нет необходимости продолжать ждать данных. После получения информации о закрытии канала приемник прекращает прием данных, если канал не закрыт, приемник останется заблокированным и существует риск тупика;
  • Освободите ресурсы и избегайте утечек ресурсов. После закрытия канала система высвободит соответствующие ресурсы. Своевременное закрытие канала позволяет избежать нерациональной траты и утечки ресурсов.

2. Как корректно закрыть канал?

Прежде всего, самый основной принцип закрытия каналов – не закрывать каналы, которые были закрыты. Во-вторых, есть еще один принцип использования каналов Go:Не закрывайте канал на приемнике данных или при наличии нескольких отправителей.другими словами，Нам следует позволить единственному отправителю канала закрыть этот канал.

Грубый способ — закрыть канал через восстановление исключений, но это явно нарушает вышеуказанные принципы и может вызвать гонку данных; другой способ — закрыть канал с помощью sync.Once или sync.Mutex, что не гарантирует, что произойдет одновременное закрытие. операции и операции отправки на канале не создают гонок за данными. Оба метода имеют определенные проблемы, поэтому я не буду описывать их подробно. Вот способ корректного закрытия канала.

Сценарий 1: M получателей и один отправитель

Одна из самых простых ситуаций. Когда отправителю необходимо завершить отправку, просто позвольте ему закрыть канал. Именно так обстоит дело в двух приведенных выше примерах программирования.

Сценарий 2: Один получатель и N отправителей.

Согласно основным принципам каналов Go, мы можем закрыть канал только у единственного отправителя канала. Значит, в данном случае мы не можем напрямую где-то закрыть канал.Но мы можем позволить получателю закрыть дополнительный канал сигнала, чтобы сообщить отправителю больше не отправлять данные.。

  1. package main
  2.  
  3. import (
  4. 	"log"
  5. 	"sync"
  6. )
  7.  
  8. func main() {
  9.  
  10. 	cosnt N := 5
  11. 	cosnt Max := 60000
  12. 	count := 0
  13.  
  14. 	dataCh := make(chan int)
  15. 	stopCh := make(chan bool)
  16.  
  17. 	var wt sync.WaitGroup
  18. 	wt.Add(1)
  19.  
  20. 	//发送者
  21. 	for i := 0; i < N; i++ {
  22. 		go func() {
  23. 			for {
  24. 				select {
  25. 				case <-stopCh:
  26. 					return
  27. 				default:
  28. 					count += 1
  29. 					dataCh <- count
  30. 				}
  31. 			}
  32. 		}()
  33. 	}
  34.  
  35. 	//接收者
  36. 	go func() {
  37. 		defer wt.Done()
  38. 		for value := range dataCh {
  39. 			if value == Max {
  40. 				// 此唯一的接收者同时也是stopCh通道的
  41. 				// 唯一发送者。尽管它不能安全地关闭dataCh数
  42. 				// 据通道，但它可以安全地关闭stopCh通道。
  43. 				close(stopCh)
  44. 				return
  45. 			}
  46. 			log.Println(value)
  47. 		}
  48. 	}()
  49.  
  50. 	wt.Wait()
  51. }

В этом методе мы добавляем дополнительный канал сигнала stopCh, который получатель использует, чтобы сообщить отправителю, что ему больше не нужно получать данные. Более того, этот метод не закрывает dataCh. Когда канал больше не используется какой-либо сопрограммой, он постепенно будет очищен от мусора, независимо от того, был ли он закрыт.

Элегантность этого метода в том, что закрывая один канал, вы перестаете использовать другой канал, тем самым косвенно закрывая другой канал.

Сценарий 3: M получателей и N отправителей

Мы не можем заставить получателя или отправителя закрыть канал, используемый для передачи данных, а также мы не можем заставить один из нескольких получателей закрыть дополнительный сигнальный канал. Обе эти практики нарушают принцип закрытия канала.

Однако мы можем представитьРоль промежуточного посредника и закрытие им дополнительных сигнальных каналов для оповещения всех получателей и отправителей об окончании работы.。

Пример кода:

  1. package main
  2.  
  3. import (
  4. 	"log"
  5. 	"math/rand"
  6. 	"strconv"
  7. 	"sync"
  8. )
  9.  
  10. func main() {
  11.  
  12. 	const Max = 100000
  13. 	const NumReceivers = 10
  14. 	const NumSenders = 1000
  15.  
  16. 	var wt sync.WaitGroup
  17. 	wt.Add(NumReceivers)
  18.  
  19. 	dataCh := make(chan int)
  20. 	stopCh := make(chan struct{})
  21. 	// stopCh是一个额外的信号通道。它的发送
  22. 	// 者为中间调解者。它的接收者为dataCh
  23. 	// 数据通道的所有的发送者和接收者。
  24. 	toStop := make(chan string, 1)
  25. 	// toStop是一个用来通知中间调解者让其
  26. 	// 关闭信号通道stopCh的第二个信号通道。
  27. 	// 此第二个信号通道的发送者为dataCh数据
  28. 	// 通道的所有的发送者和接收者，它的接收者
  29. 	// 为中间调解者。它必须为一个缓冲通道。
  30.  
  31. 	var stoppedBy string
  32.  
  33. 	// 中间调解者
  34. 	go func() {
  35. 		stoppedBy = <-toStop
  36. 		close(stopCh)
  37. 	}()
  38.  
  39. 	// 发送者
  40. 	for i := 0; i < NumSenders; i++ {
  41. 		go func(id string) {
  42. 			for {
  43. 				value := rand.Intn(Max)
  44. 				if value == 0 {
  45. 					// 为了防止阻塞，这里使用了一个尝试
  46. 					// 发送操作来向中间调解者发送信号。
  47. 					select {
  48. 					case toStop <- "发送者#" + id:
  49. 					default:
  50. 					}
  51. 					return
  52. 				}
  53.  
  54. 				select {
  55. 				case <-stopCh:
  56. 					return
  57. 				case dataCh <- value:
  58. 				}
  59. 			}
  60. 		}(strconv.Itoa(i))
  61. 	}
  62.  
  63. 	// 接收者
  64. 	for i := 0; i < NumReceivers; i++ {
  65. 		go func(id string) {
  66. 			defer wt.Done()
  67.  
  68. 			for {
  69. 				select {
  70. 				case <-stopCh:
  71. 					return
  72. 				case value := <-dataCh:
  73. 					if value == Max {
  74. 						// 为了防止阻塞，这里使用了一个尝试
  75. 						// 发送操作来向中间调解者发送信号。
  76. 						select {
  77. 						case toStop <- "接收者:" + id:
  78. 						default:
  79. 						}
  80. 						return
  81. 					}
  82.  
  83. 					log.Println(value)
  84. 				}
  85. 			}
  86. 		}(strconv.Itoa(i))
  87. 	}
  88.  
  89. 	wt.Wait()
  90. 	log.Println("被" + stoppedBy + "终止了")
  91.  
  92. }

Личный профиль

Он посвятил себя исследованию технологий более 30 лет и владеет различными языками, такими как Java, Linux, Javascript, php, css и т. д. Он внес большой вклад в область открытого исходного кода. Станция документации для разработчиков, где можно поделиться некоторыми проблемами в разработке технологий для дальнейшего использования.

моя контактная информация

Почта