技術共有

Go コルーチンとチャネルの包括的なアプリケーションの問題

2024-07-12

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

1. コルーチンとチャネルとは何かを簡単に理解する

コルーチンとは

コルーチンは、独立したスタック領域を持ち、プログラムのヒープ領域を共有するユーザーレベルの軽量スレッドです。

シングルスレッドをベースにしたアルゴリズムで実装されたマイクロスレッドであり、マルチスレッドプログラミングと比較して次のような利点があります。

  • コルーチンのコンテキスト切り替えはユーザーによって決定されるため、システム カーネルのコンテキスト切り替えは必要なく、オーバーヘッドが削減されます。
  • デフォルトでは、コルーチンは中断を防ぐために完全に保護されています。アトミック操作ロックは必要ありません
  • 単一スレッドでも高い同時実行性を実現でき、シングルコア CPU でも数万のコルーチンをサポートできます。

チャンネルとは何ですか

チャネルは、コルーチン間の通信に使用されるデータ構造です。キューと同様に、一方の端が送信者で、もう一方の端が受信者です。チャネルを使用すると、データの同期と順序を確保できます。

チャネルはバッファ付きチャネルとバッファなしチャネルに分けられ、次のように宣言されます。

  • バッファチャネルがあります
intChan := make(chan int,<缓冲容量>)
  • バッファリングされていないチャネル
intChan := make(chan int)

バッファー付きチャネルとバッファーなしチャネルの違い:

  • ブロッキング: バッファなしチャネルの送信側はデータが受信されるまでブロックし、バッファ付きチャネルの送信側はバッファがいっぱいになるまでブロックし、受信側はバッファが空でなくなるまでブロックします。
  • データの同期と順序: バッファなしチャネルはデータの同期と順序を保証しますが、バッファ付きパイプはデータの同期と順序を保証しません。
  • アプリケーション シナリオ: バッファなしチャネルには厳密な同期とシーケンスが必要ですが、バッファ付きチャネルは非同期通信を行うことができ、スループットを向上させることができます。

バッファなしチャネルの実装で注意する必要があるのは、チャネルの両端に送信者と受信者が存在する必要があることです。そうでないとデッドロックが発生します。

2.コルーチンチャネル同時プログラミングのケース

(1) 文字と数字を交互に印刷する

質問の意味: coroutine-channel を使用して、数字 1 ～ 10 と文字 A～J を交互に出力します。

コード：

  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. }

トピック分析:

この質問では、文字と数字を交互に出力する必要があるため、バッファなしチャネルのアプリケーション シナリオと一致する 2 つのコルーチンの厳密な順序を確保する必要があります。数字と文字をそれぞれ保存する 2 つのチャネルを設定します。数字と文字を出力する 2 つのコルーチンは、それぞれ 2 つのチャネルの送信側と受信側として機能します。ループ内で 1 回印刷し、別のコルーチンに印刷を促す信号を 1 回送信します。

最後の文字「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 個の受信者と 1 人の送信者

最も対処しやすい状況の 1 つです。送信者が送信を終了する必要がある場合は、チャネルを閉じてください。上記の 2 つのプログラミング例がこれに当てはまります。

シナリオ 2: 1 人の受信者と 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 を閉じません。チャネルがコルーチンによって使用されなくなった場合、チャネルが閉じられているかどうかに関係なく、徐々にガベージ コレクションが行われます。

この方法の優れた点は、1 つのチャネルを閉じることで別のチャネルの使用を停止し、それによって間接的に他のチャネルも閉じることになることです。

シナリオ 3: M 個の受信者と N 個の送信者

受信者または送信者のいずれかがデータの送信に使用されるチャネルを閉じることはできません。また、複数の受信者の 1 つが追加のシグナリング チャネルを閉じることもできません。これらの実践はどちらもチャネル閉鎖原則に違反します。

ただし、ご紹介できるのは、中間調停者の役割と、すべての受信者と送信者に作業の終了を通知する追加の信号チャネルを閉じさせる。

コード例:

  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 などのさまざまな言語に堪能であり、オープンソース分野で多くの貢献を行っています。将来の参考のために技術開発におけるいくつかの問題を共有する開発者ドキュメント ステーション。

プライベートな連絡先の最初の情報

送料