Berbagi teknologi

Masalah aplikasi coroutine dan saluran Go yang komprehensif

2024-07-12

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

1. Pahami secara singkat apa itu coroutine dan saluran

Apa itu coroutine

Coroutine adalah thread ringan tingkat pengguna yang memiliki ruang tumpukan independen dan berbagi ruang tumpukan program.

Ini adalah mikro-thread yang diimplementasikan melalui algoritma berdasarkan single thread, dibandingkan dengan pemrograman multi-thread, ia memiliki keuntungan sebagai berikut:

  • Peralihan konteks coroutine ditentukan oleh pengguna, tanpa memerlukan peralihan konteks kernel sistem, sehingga mengurangi overhead.
  • Secara default, coroutine dilindungi sepenuhnya untuk mencegah interupsi.Tidak diperlukan kunci operasi atom
  • Satu thread juga dapat mencapai konkurensi tinggi, dan bahkan CPU inti tunggal dapat mendukung puluhan ribu coroutine.

apa itu saluran

Saluran adalah struktur data yang digunakan untuk komunikasi antar coroutine. Mirip dengan antrian, salah satu ujungnya adalah pengirim dan ujung lainnya adalah penerima. Menggunakan saluran dapat memastikan sinkronisasi dan urutan data.

Saluran dibagi menjadi saluran buffered dan saluran unbuffered, yang dinyatakan sebagai berikut:

  • Ada saluran penyangga
intChan := make(chan int,<缓冲容量>)
  • saluran tanpa buffer
intChan := make(chan int)

Perbedaan antara saluran yang di-buffer dan saluran yang tidak di-buffer:

  • Pemblokiran: Pengirim saluran yang tidak di-buffer akan memblokir hingga data diterima; pengirim saluran yang di-buffer akan memblokir hingga buffer penuh, dan penerima akan memblokir hingga buffer tidak kosong.
  • Sinkronisasi dan urutan data: Saluran tanpa buffer menjamin sinkronisasi dan urutan data; pipa dengan buffer tidak menjamin sinkronisasi dan urutan data.
  • Skenario aplikasi: Saluran tanpa buffer memerlukan sinkronisasi dan urutan yang ketat; saluran dengan buffer dapat berkomunikasi secara asinkron dan meningkatkan throughput.

Yang perlu diperhatikan dalam penerapan saluran unbuffered adalah harus ada pengirim dan penerima di kedua ujung saluran, jika tidak maka akan terjadi kebuntuan.

2. Kasus pemrograman bersamaan saluran coroutine

(1) Cetak huruf dan angka secara bergantian

Arti pertanyaan: Gunakan saluran coroutine untuk mencetak angka 1-10 dan huruf AJ secara bergantian.

Kode:

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

Analisis topik:

Pertanyaannya mengharuskan kita mencetak huruf dan angka secara bergantian, jadi kita perlu memastikan urutan kedua coroutine yang ketat, yang konsisten dengan skenario penerapan saluran tanpa buffer. Siapkan dua saluran untuk menyimpan angka dan huruf masing-masing. Dua coroutine yang mencetak angka dan huruf masing-masing berfungsi sebagai pengirim dan penerima kedua saluran. Cetak sekali dalam satu putaran dan kirim sinyal sekali untuk mengingatkan coroutine lain untuk mencetak.

Perlu dicatat bahwa ketika karakter terakhir '10' dicetak, coroutine untuk mencetak huruf telah berakhir, dan saluran numCh tidak memiliki penerima. Saat ini, kondisi implementasi saluran tanpa buffer tidak lagi terpenuhi - harus ada pengirim dan penerima. Jika tidak, pengiriman sinyal lagi akan menyebabkan kebuntuan pemblokiran. Jadi tidak perlu mengirimkan sinyal lagi untuk yang ke 10 kalinya.

(2) Rancang penjadwal tugas

Judul: Merancang penjadwal tugas yang menggunakan model pemrograman multi-coroutine + saluran untuk mengimplementasikan skenario bisnis pemrosesan multi-tugas secara bersamaan, dan memerlukan urutan penjadwalan sesuai urutan penambahan tugas.

Kode:

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

analisa masalah:

Karena tugas yang ditambahkan bersifat multitugas, maka ada lebih dari satu tugas, dan pemrosesan asinkron diperlukan untuk menjalankan tugas ini. Mematuhi saluran buffered memerlukan peningkatan throughput dan pemrosesan asinkron.

Kemudian, kita perlu memasukkan tugas ke dalam saluran, dan beberapa penerima dapat mengambil tugas dari saluran tersebut secara berurutan dan menjalankannya.

Masalah yang perlu diperhatikan adalah jika jumlah tugas yang ditambahkan lebih besar dari buffer saluran maka akan menyebabkan pemblokiran saat penambahan tugas. Agar tidak mempengaruhi startup normal konsumen, ia perlu membuka coroutine terpisah untuk menambahkan tugas.

Dengan cara ini, ketika konsumen melakukan konsumsi, produsen yang menghalangi akan dibangunkan untuk terus menambah tugas.

3. Ringkasan

Setelah mempelajari model pemrograman coroutine + channel, selain yang baru saja disebutkan pada judul, kita juga harus memperhatikan hal-hal berikut:

1. Mengapa saluran harus ditutup setelah habis?

  • Untuk menghindari kebuntuan. Menutup saluran juga memberi tahu penerima bahwa tidak ada lagi data yang dikirim dari pengirim dan tidak perlu terus menunggu data. Setelah menerima informasi penutupan saluran, penerima berhenti menerima data; jika saluran tidak ditutup, penerima akan tetap diblokir dan ada risiko kebuntuan.
  • Lepaskan sumber daya dan hindari kebocoran sumber daya. Setelah menutup saluran, sistem akan melepaskan sumber daya yang sesuai. Menutup saluran tepat waktu dapat menghindari pemborosan dan kebocoran sumber daya.

2. Bagaimana cara menutup saluran dengan baik?

Pertama-tama, prinsip paling dasar dalam menutup saluran adalah jangan menutup saluran yang sudah ditutup. Kedua, ada prinsip lain dalam menggunakan saluran Go:Jangan menutup saluran pada penerima data atau ketika ada banyak pengirim.dengan kata lain，Kita hanya boleh membiarkan satu-satunya pengirim saluran menutup saluran ini.

Cara yang kasar adalah dengan menutup saluran melalui pemulihan pengecualian, tetapi ini jelas melanggar prinsip di atas dan dapat menyebabkan persaingan data. Cara lain adalah menutup saluran dengan sync.Once atau sync.Mutex, yang tidak dijamin akan terjadi secara bersamaan operasi dan operasi pengiriman pada saluran tidak menyebabkan perlombaan data. Kedua metode tersebut memiliki masalah tertentu, jadi saya tidak akan memperkenalkannya secara detail. Berikut adalah metode cara menutup saluran dengan baik.

Skenario 1: M penerima dan satu pengirim

Salah satu situasi termudah untuk dihadapi. Ketika pengirim harus menyelesaikan pengiriman, biarkan saja ia menutup salurannya. Hal ini terjadi pada dua contoh pemrograman di atas.

Skenario 2: Satu penerima dan N pengirim

Menurut prinsip dasar saluran Go, kita hanya dapat menutup saluran pada satu-satunya pengirim saluran tersebut. Jadi, dalam hal ini, kita tidak bisa langsung menutup saluran tersebut di suatu tempat.Namun kita dapat membiarkan penerima menutup saluran sinyal tambahan untuk memberitahu pengirim agar tidak mengirim data lagi.。

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

Dalam metode ini, kami menambahkan saluran sinyal tambahan stopCh, yang digunakan penerima untuk memberi tahu pengirim bahwa ia tidak perlu lagi menerima data. Selain itu, metode ini tidak menutup dataCh. Ketika saluran tidak lagi digunakan oleh coroutine apa pun, saluran tersebut akan dikumpulkan secara bertahap, terlepas dari apakah saluran tersebut telah ditutup.

Keanggunan metode ini adalah dengan menutup satu saluran, Anda berhenti menggunakan saluran lain, sehingga secara tidak langsung menutup saluran lainnya.

Skenario 3: M penerima dan N pengirim

Kita tidak bisa meminta penerima atau pengirim menutup saluran yang digunakan untuk mengirimkan data, kita juga tidak bisa meminta salah satu dari beberapa penerima menutup saluran sinyal tambahan. Kedua praktik ini melanggar prinsip penutupan saluran.

Namun, kami bisa memperkenalkannyaPeran mediator perantara dan menutup saluran sinyal tambahan untuk memberi tahu semua penerima dan pengirim tentang akhir pekerjaan。

Contoh kode:

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

Profil pribadi

Ia telah mengabdikan dirinya untuk meneliti teknologi selama lebih dari 30 tahun, dan mahir dalam berbagai bahasa seperti java, linux, javascript, php, css, dll. Saya telah memberikan banyak kontribusi di bidang dokumentasi pengembang open source untuk berbagi beberapa masalah dalam pengembangan teknologi untuk referensi di masa mendatang.

informasi kontak saya

Surat