حالات استخدام القنوات

قبل قراءة هذا المقال، يُرجى قراءة مقالة القنوات في Go، التي تشرح أنواع القنوات وقيمها بالتفصيل. وقد يحتاج المبرمجون الجدد على Go إلى قراءة تلك المقالة وهذه المقالة عدة مرات لإتقان برمجة القنوات في Go.

سيعرض ما تبقّى من هذا المقال العديد من حالات استخدام القنوات. وآمل أن يُقنعك هذا المقال بأن

البرمجة غير المتزامنة والمتزامنة بقنوات Go سهلة وممتعة.
تقنية مزامنة القنوات لها نطاق استخدامات أوسع وتنوّعات أكثر من حلول المزامنة المستخدمة في بعض اللغات الأخرى، مثل نموذج الفاعل ونمط async/await.

يُرجى ملاحظة أن الهدف من هذا المقال هو عرض أكبر عدد ممكن من حالات استخدام القنوات. وينبغي أن نعلم أن القناة ليست تقنية مزامنة التزامن الوحيدة المدعومة في Go، وقد لا يكون أسلوب القناة أفضل حل في بعض الحالات. يُرجى قراءة العمليات الذرية وبعض تقنيات المزامنة الأخرى لمزيد من تقنيات مزامنة التزامن في Go.

استخدام القنوات كـFutures/Promises

تُستخدم الـFutures والـPromises في العديد من اللغات الشائعة الأخرى. وغالباً ما ترتبط بالطلبات والاستجابات.

إرجاع قنوات استقبال فقط كنتائج

في المثال التالي، تُطلَب قيمتا معطَي استدعاء الدالة sumSquares بشكل متزامن. وستحجب كل عملية استقبال من القناتين حتى تُنفَّذ عملية إرسال على القناة المقابلة. ويستغرق إرجاع النتيجة النهائية نحو ثلاث ثوانٍ بدلاً من ست ثوانٍ.


package main

import (
	"fmt"
	"math/rand"
	"time"
)

func longTimeRequest() <-chan int32 {
	r := make(chan int32)

	go func() {
		// محاكاة عبء عمل.
		time.Sleep(time.Second * 3)
		r <- rand.Int31n(100)
	}()

	return r
}

func sumSquares(a, b int32) int32 {
	return a*a + b*b
}

func main() {
	a, b := longTimeRequest(), longTimeRequest()
	fmt.Println(sumSquares(<-a, <-b))
}

تمرير قنوات إرسال فقط كمعطيات

مثل المثال السابق، في المثال التالي، تُطلَب قيمتا معطَي استدعاء الدالة sumSquares بشكل متزامن. وخلافاً للمثال السابق، تأخذ الدالة longTimeRequest قناة إرسال فقط كوسيط بدلاً من إرجاع قناة استقبال فقط كنتيجة.


package main

import (
	"fmt"
	"math/rand"
	"time"
)

func longTimeRequest(r chan<- int32)  {
	// محاكاة عبء عمل.
	time.Sleep(time.Second * 3)
	r <- rand.Int31n(100)
}

func sumSquares(a, b int32) int32 {
	return a*a + b*b
}

func main() {
	ra, rb := make(chan int32), make(chan int32)
	go longTimeRequest(ra)
	go longTimeRequest(rb)

	fmt.Println(sumSquares(<-ra, <-rb))
}

في الواقع، لا نحتاج في المثال المذكور أعلاه إلى قناتين لنقل النتائج؛ فاستخدام قناة واحدة كافٍ.


...

	// يمكن أن تكون القناة مخزّنة أو غير مخزّنة.
	results := make(chan int32, 2)
	go longTimeRequest(results)
	go longTimeRequest(results)

	fmt.Println(sumSquares(<-results, <-results))
}

هذا نوع من تجميع البيانات سيُقدَّم خصيصاً أدناه.

فوز أول استجابة

هذا تحسين للصيغة التي تستخدم قناة واحدة فقط في المثال السابق.

أحياناً، يمكن استقبال قطعة بيانات من عدة مصادر لتجنّب زمن استجابة مرتفع. ولأسباب كثيرة، قد تتفاوت مدد استجابة هذه المصادر كثيراً. وحتى بالنسبة إلى مصدر بعينه، فإن مدد استجابته ليست ثابتة أيضاً. ولجعل مدة الاستجابة أقصر ما يمكن، يمكننا إرسال طلب إلى كل مصدر في goroutine منفصلة. ولن تُستخدم إلا أول استجابة، وستُهمَل الاستجابات الأبطأ الأخرى.

لاحظ أنه إذا كان هناك N من المصادر، فيجب أن تكون سعة قناة التواصل N-1 على الأقل، لتجنّب بقاء الـgoroutines المقابلة للاستجابات المُهمَلة محجوبة إلى الأبد.


package main

import (
	"fmt"
	"math/rand"
	"time"
)

func source(c chan<- int32) {
	ra, rb := rand.Int31(), rand.Intn(3) + 1
	// النوم 1 ثانية/2 ثانية/3 ثوانٍ.
	time.Sleep(time.Duration(rb) * time.Second)
	c <- ra
}

func main() {
	startTime := time.Now()
	// يجب أن تكون c قناة مخزّنة.
	c := make(chan int32, 5)
	for i := 0; i < cap(c); i++ {
		go source(c)
	}
	// لن تُستخدم إلا أول استجابة.
	rnd := <- c
	fmt.Println(time.Since(startTime))
	fmt.Println(rnd)
}

هناك طرق أخرى لتنفيذ حالة استخدام فوز أول استجابة، باستخدام آلية select وقناة مخزّنة سعتها واحد. وستُقدَّم طرق أخرى أدناه.

استخدام القنوات للإشعارات

يمكن اعتبار الإشعارات طلبات/استجابات خاصة لا تكون فيها القيم المُستجاب بها مهمة. وعموماً، نستخدم نوع البنية الفارغة struct{} كنوع عنصر لقنوات الإشعار، لأن حجم النوع struct{} صفر، ومن ثمّ لا تستهلك قيم struct{} ذاكرة.

إشعار من 1 إلى 1 بإرسال قيمة إلى قناة

إذا لم تكن هناك قيم يمكن استقبالها من قناة، فستحجب عملية الاستقبال التالية على القناة حتى ترسل goroutine أخرى قيمة إلى القناة. لذا يمكننا إرسال قيمة إلى قناة لإشعار goroutine أخرى تنتظر استقبال قيمة من القناة نفسها.

في المثال التالي، تُستخدم القناة done كقناة إشارة لتنفيذ الإشعارات.


package main

import (
	"crypto/rand"
	"fmt"
	"os"
	"sort"
)

func main() {
	values := make([]byte, 32 * 1024 * 1024)
	if _, err := rand.Read(values); err != nil {
		fmt.Println(err)
		os.Exit(1)
	}

	done := make(chan struct{}) // يمكن أن تكون مخزّنة أو غير مخزّنة

	// الـgoroutine الخاصة بالترتيب
	go func() {
		sort.Slice(values, func(i, j int) bool {
			return values[i] < values[j]
		})
		// الإشعار بأن الترتيب انتهى.
		done <- struct{}{}
	}()

	// القيام بأشياء أخرى ...

	<- done // الانتظار هنا حتى الإشعار
	fmt.Println(values[0], values[len(values)-1])
}

إشعار من 1 إلى 1 باستقبال قيمة من قناة

إذا كان طابور مخزن القيم لقناة ممتلئاً (وطابور مخزن القيم لقناة غير مخزّنة ممتلئ دائماً)، فستحجب عملية إرسال على القناة حتى تستقبل goroutine أخرى قيمة من القناة. لذا يمكننا استقبال قيمة من قناة لإشعار goroutine أخرى تنتظر إرسال قيمة إلى القناة نفسها. وعموماً، ينبغي أن تكون القناة غير مخزّنة.

وتُستخدم طريقة الإشعار هذه أقل بكثير من الطريقة المُقدَّمة في المثال السابق.


package main

import (
	"fmt"
	"time"
)

func main() {
	done := make(chan struct{})
		// يمكن أن تكون سعة قناة الإشارة
		// واحداً أيضاً. وإذا كان الأمر كذلك،
		// فيجب إرسال قيمة إلى القناة قبل
		// إنشاء الـgoroutine التالية.

	go func() {
		fmt.Print("Hello")
		// محاكاة عبء عمل.
		time.Sleep(time.Second * 2)

		// استقبل قيمة من القناة done،
		// لفكّ حجب الإرسال الثاني في
		// الـgoroutine main.
		<- done
	}()

	// محجوب هنا، في انتظار إشعار.
	done <- struct{}{}
	fmt.Println(" world!")
}

في الواقع، لا فروق جوهرية بين استقبال القيم أو إرسالها لتنفيذ الإشعارات؛ ويمكن تلخيص كلتيهما بأن الأسرع يتلقّى الإشعار من الأبطأ.

إشعارات من N إلى 1 ومن 1 إلى N

بتوسيع حالتي الاستخدام أعلاه قليلاً، يسهل تنفيذ إشعارات من N إلى 1 ومن 1 إلى N.


package main

import "log"
import "time"

type T = struct{}

func worker(id int, ready <-chan T, done chan<- T) {
	<-ready // احجب هنا وانتظر إشعاراً
	log.Print("Worker#", id, " starts.")
	// محاكاة عبء عمل.
	time.Sleep(time.Second * time.Duration(id+1))
	log.Print("Worker#", id, " job done.")
	// إشعار الـgoroutine main (من N إلى 1)،
	done <- T{}
}

func main() {
	log.SetFlags(0)

	ready, done := make(chan T), make(chan T)
	go worker(0, ready, done)
	go worker(1, ready, done)
	go worker(2, ready, done)

	// محاكاة مرحلة تهيئة.
	time.Sleep(time.Second * 3 / 2)
	// إشعارات من 1 إلى N.
	ready <- T{}; ready <- T{}; ready <- T{}
	// تلقّي إشعار من N إلى 1.
	<-done; <-done; <-done
}

في الواقع، لا تُستخدم طرق تنفيذ الإشعارات من 1 إلى N ومن N إلى 1 المُقدَّمة في هذا القسم الفرعي كثيراً في الممارسة. ففي الممارسة، نستخدم غالباً sync.WaitGroup لتنفيذ إشعارات من N إلى 1، وننفّذ إشعارات من 1 إلى N بإغلاق القنوات. يُرجى قراءة القسم الفرعي التالي للتفاصيل.

إشعارات البثّ (من 1 إلى N) بإغلاق قناة

نادراً ما تُستخدم طريقة تنفيذ الإشعارات من 1 إلى N المعروضة في القسم الفرعي السابق في الممارسة، لوجود طريقة أفضل. فبالاستفادة من ميزة إمكانية استقبال عدد لا نهائي من القيم من قناة مغلقة، يمكننا إغلاق قناة لبثّ الإشعارات.

بالمثال الوارد في القسم الفرعي السابق، يمكننا استبدال عمليات إرسال القناة الثلاث ready <- struct{}{} في المثال السابق بعملية إغلاق قناة واحدة close(ready) لتنفيذ إشعارات من 1 إلى N.


...
	close(ready) // بثّ الإشعارات
...

بالطبع، يمكننا أيضاً إغلاق قناة لتنفيذ إشعار من 1 إلى 1. وفي الواقع، هذه هي طريقة الإشعار الأكثر استخداماً في Go.

ستُستغلّ ميزة إمكانية استقبال عدد لا نهائي من القيم من قناة مغلقة في العديد من حالات الاستخدام الأخرى المُقدَّمة أدناه. وفي الواقع، تُستخدم هذه الميزة على نطاق واسع في الحزم القياسية. فمثلاً، تستخدم حزمة context هذه الميزة لتأكيد الإلغاءات.

المؤقّت: إشعار مجدول

يسهل استخدام القنوات لتنفيذ مؤقّتات تُستخدم مرة واحدة.

تنفيذ مخصّص لمؤقّت يُستخدم مرة واحدة:


package main

import (
	"fmt"
	"time"
)

func AfterDuration(d time.Duration) <- chan struct{} {
	c := make(chan struct{}, 1)
	go func() {
		time.Sleep(d)
		c <- struct{}{}
	}()
	return c
}

func main() {
	fmt.Println("Hi!")
	<- AfterDuration(time.Second)
	fmt.Println("Hello!")
	<- AfterDuration(time.Second)
	fmt.Println("Bye!")
}

في الواقع، توفّر الدالة After في الحزمة القياسية time الوظيفة نفسها بتنفيذ أعلى كفاءة بكثير. وينبغي استخدام تلك الدالة بدلاً منها لتبدو الشيفرة نظيفة.

يُرجى ملاحظة أن <-time.After(aDuration) سيجعل الـgoroutine الحالية تدخل حالة الحجب، بينما استدعاء الدالة time.Sleep(aDuration) لن يفعل ذلك.

غالباً ما يُستخدم <-time.After(aDuration) في آلية المهلة الزمنية التي ستُقدَّم أدناه.

استخدام القنوات كأقفال mutex

ذكر أحد الأمثلة أعلاه أنه يمكن استخدام قنوات مخزّنة بسعة واحد كـسيمافور ثنائي يُستخدم مرة واحدة. وفي الواقع، يمكن استخدام مثل هذه القنوات أيضاً كسيمافورات ثنائية تُستخدم عدة مرات، أي أقفال mutex، رغم أن أقفال mutex هذه ليست بكفاءة أقفال mutex الموفَّرة في الحزمة القياسية sync.

هناك طريقتان لاستخدام قنوات مخزّنة بسعة واحد كأقفال mutex.

القفل عبر إرسال، وفتح القفل عبر استقبال.
القفل عبر استقبال، وفتح القفل عبر إرسال.

في ما يلي مثال على القفل عبر الإرسال.


package main

import "fmt"

func main() {
	// يجب أن تكون السعة واحداً.
	mutex := make(chan struct{}, 1)

	counter := 0
	increase := func() {
		mutex <- struct{}{} // قفل
		counter++
		<-mutex // فتح القفل
	}

	increase1000 := func(done chan<- struct{}) {
		for i := 0; i < 1000; i++ {
			increase()
		}
		done <- struct{}{}
	}

	done := make(chan struct{})
	go increase1000(done)
	go increase1000(done)
	<-done; <-done
	fmt.Println(counter) // 2000
}

في ما يلي مثال على القفل عبر الاستقبال. وهو يعرض فقط الجزء المعدَّل بناءً على مثال القفل عبر الإرسال أعلاه.


...
func main() {
	mutex := make(chan struct{}, 1)
	mutex <- struct{}{} // هذا السطر مطلوب.

	counter := 0
	increase := func() {
		<-mutex // قفل
		counter++
		mutex <- struct{}{} // فتح القفل
	}
...

استخدام القنوات كسيمافورات عدّية

يمكن استخدام القنوات المخزّنة كـسيمافورات عدّية. ويمكن اعتبار السيمافورات العدّية أقفالاً متعددة المالكين. فإذا كانت سعة قناة هي N، فيمكن اعتبارها قفلاً يمكن أن يكون له N من المالكين على الأكثر في أي وقت. والسيمافورات الثنائية (أقفال mutex) سيمافورات عدّية خاصة، إذ يمكن أن يكون لكل سيمافور ثنائي مالك واحد على الأكثر في أي وقت.

غالباً ما تُستخدم السيمافورات العدّية لفرض حدّ أقصى لعدد الطلبات المتزامنة.

ومثل استخدام القنوات كأقفال mutex، هناك أيضاً طريقتان لاكتساب حصة ملكية واحدة من سيمافور قناة.

اكتساب الملكية عبر إرسال، وتحريرها عبر استقبال.
اكتساب الملكية عبر استقبال، وتحريرها عبر إرسال.

مثال على اكتساب الملكية عبر استقبال قيم من قناة.


package main

import (
	"log"
	"math/rand"
	"time"
)

type Seat int
type Bar chan Seat

func (bar Bar) ServeCustomer(c int) {
	log.Print("customer#", c, " enters the bar")
	seat := <- bar // يحتاج مقعداً ليشرب
	log.Print("++ customer#", c, " drinks at seat#", seat)
	time.Sleep(time.Second * time.Duration(2 + rand.Intn(6)))
	log.Print("-- customer#", c, " frees seat#", seat)
	bar <- seat // حرّر المقعد واترك الحانة
}

func main() {
	// الحانة فيها 10 مقاعد.
	bar24x7 := make(Bar, 10)
	// ضع المقاعد في الحانة.
	for seatId := 0; seatId < cap(bar24x7); seatId++ {
		// لن تحجب أي من عمليات الإرسال.
		bar24x7 <- Seat(seatId)
	}

	for customerId := 0; ; customerId++ {
		time.Sleep(time.Second)
		go bar24x7.ServeCustomer(customerId)
	}

	// النوم != الحجب
	for {time.Sleep(time.Second)}
}

في المثال أعلاه، لا يمكن للعملاء الشرب إلا إذا حصل كل منهم على مقعد. لذا سيكون هناك عشرة عملاء على الأكثر يشربون في أي وقت معيّن.

الحلقة for الأخيرة في الدالة main موجودة لتجنّب خروج البرنامج. وهناك طريقة أفضل، ستُقدَّم أدناه، للقيام بالمهمة.

في المثال أعلاه، رغم أنه سيكون هناك عشرة عملاء على الأكثر يشربون في أي وقت معيّن، فقد يُخدَم أكثر من عشرة عملاء في الحانة في الوقت نفسه. فبعض العملاء ينتظرون مقاعد شاغرة. ومع أن كل goroutine خاصة بعميل تستهلك موارد أقل بكثير من خيط نظام، فإن الموارد الإجمالية التي تستهلكها أعداد كبيرة من الـgoroutines ليست مهملة. لذا من الأفضل إنشاء goroutine لعميل فقط عند توفّر مقعد شاغر.


... // نفس شيفرة المثال أعلاه

func (bar Bar) ServeCustomerAtSeat(c int, seat Seat) {
	log.Print("++ customer#", c, " drinks at seat#", seat)
	time.Sleep(time.Second * time.Duration(2 + rand.Intn(6)))
	log.Print("-- customer#", c, " frees seat#", seat)
	bar <- seat // حرّر المقعد واترك الحانة
}

func main() {
	bar24x7 := make(Bar, 10)
	for seatId := 0; seatId < cap(bar24x7); seatId++ {
		bar24x7 <- Seat(seatId)
	}

	for customerId := 0; ; customerId++ {
		time.Sleep(time.Second)
		// نحتاج مقعداً لخدمة العميل التالي.
		seat := <- bar24x7
		go bar24x7.ServeCustomerAtSeat(customerId, seat)
	}
	for {time.Sleep(time.Second)}
}

سيكون هناك نحو عشرة goroutines حيّة للعملاء على الأكثر متعايشة في النسخة المحسَّنة أعلاه (لكن سيظل هناك الكثير من الـgoroutines الخاصة بالعملاء تُنشأ خلال عمر البرنامج).

في التنفيذ الأكثر كفاءة المعروض أدناه، ستُنشأ عشرة goroutines لخدمة العملاء على الأكثر خلال عمر البرنامج.


... // نفس شيفرة المثال أعلاه

func (bar Bar) ServeCustomerAtSeat(consumers chan int) {
	for c := range consumers {
		seatId := <- bar
		log.Print("++ customer#", c, " drinks at seat#", seatId)
		time.Sleep(time.Second * time.Duration(2 + rand.Intn(6)))
		log.Print("-- customer#", c, " frees seat#", seatId)
		bar <- seatId // حرّر المقعد واترك الحانة
	}
}

func main() {
	bar24x7 := make(Bar, 10)
	for seatId := 0; seatId < cap(bar24x7); seatId++ {
		bar24x7 <- Seat(seatId)
	}

	consumers := make(chan int)
	for i := 0; i < cap(bar24x7); i++ {
		go bar24x7.ServeCustomerAtSeat(consumers)
	}

	for customerId := 0; ; customerId++ {
		time.Sleep(time.Second)
		consumers <- customerId
	}
}

خارج الموضوع: بالطبع، إذا لم نكن نهتم بمعرّفات المقاعد (وهو أمر شائع في الممارسة)، فإن سيمافور bar24x7 ليس ضرورياً على الإطلاق:


... // نفس شيفرة المثال أعلاه

func ServeCustomer(consumers chan int) {
	for c := range consumers {
		log.Print("++ customer#", c, " drinks at the bar")
		time.Sleep(time.Second * time.Duration(2 + rand.Intn(6)))
		log.Print("-- customer#", c, " leaves the bar")
	}
}

func main() {
	const BarSeatCount = 10
	consumers := make(chan int)
	for i := 0; i < BarSeatCount; i++ {
		go ServeCustomer(consumers)
	}

	for customerId := 0; ; customerId++ {
		time.Sleep(time.Second)
		consumers <- customerId
	}
}

طريقة اكتساب ملكية السيمافور عبر الإرسال أبسط نسبياً، ولا حاجة فيها إلى خطوة وضع المقاعد.


package main

import (
	"log"
	"math/rand"
	"time"
)

type Customer struct{id int}
type Bar chan Customer

func (bar Bar) ServeCustomer(c Customer) {
	log.Print("++ customer#", c.id, " starts drinking")
	time.Sleep(time.Second * time.Duration(3 + rand.Intn(16)))
	log.Print("-- customer#", c.id, " leaves the bar")
	<- bar // يترك الحانة ويوفّر مكاناً
}

func main() {
	// يمكن للحانة خدمة 10 عملاء
	// على الأكثر في الوقت نفسه.
	bar24x7 := make(Bar, 10)
	for customerId := 0; ; customerId++ {
		time.Sleep(time.Second * 2)
		customer := Customer{customerId}
		// انتظر الدخول إلى الحانة.
		bar24x7 <- customer
		go bar24x7.ServeCustomer(customer)
	}
	for {time.Sleep(time.Second)}
}

الحوار (Ping-Pong)

يمكن لـgoroutinتين التحاور عبر قناة. وفي ما يلي مثال سيطبع سلسلة من أعداد فيبوناتشي.


package main

import "fmt"
import "time"
import "os"

type Ball uint64

func Play(playerName string, table chan Ball) {
	var lastValue Ball = 1
	for {
		ball := <- table // استلم الكرة
		fmt.Println(playerName, ball)
		ball += lastValue
		if ball < lastValue { // تجاوز سعة
			os.Exit(0)
		}
		lastValue = ball
		table <- ball // أعد ضرب الكرة
		time.Sleep(time.Second)
	}
}

func main() {
	table := make(chan Ball)
	go func() {
		table <- 1 // ارمِ الكرة على الطاولة
	}()
	go Play("A:", table)
	Play("B:", table)
}

قناة مغلَّفة داخل قناة

أحياناً، يمكننا استخدام نوع قناة كنوع عنصر نوع قناة آخر. وفي المثال التالي، chan chan<- int نوع قناة نوع عنصره نوع قناة إرسال فقط chan<- int.


package main

import "fmt"

var counter = func (n int) chan<- chan<- int {
	requests := make(chan chan<- int)
	go func() {
		for request := range requests {
			if request == nil {
				n++ // زيادة
			} else {
				request <- n // أخذ
			}
		}
	}()

	// يُحوَّل ضمنياً إلى chan<- (chan<- int)
	return requests
}(0)

func main() {
	increase1000 := func(done chan<- struct{}) {
		for i := 0; i < 1000; i++ {
			counter <- nil
		}
		done <- struct{}{}
	}

	done := make(chan struct{})
	go increase1000(done)
	go increase1000(done)
	<-done; <-done

	request := make(chan int, 1)
	counter <- request
	fmt.Println(<-request) // 2000
}

مع أن تنفيذ التغليف هنا قد لا يكون الطريقة الأكثر كفاءة للمثال المذكور أعلاه، فقد تكون حالة الاستخدام مفيدة في بعض السيناريوهات الأخرى.

فحص أطوال القنوات وسعاتها

يمكننا استخدام الدالتين المدمجتين len وcap لفحص طول قناة وسعتها. لكننا نادراً ما نفعل ذلك في الممارسة. وسبب قلّة استخدامنا الدالة len لفحص طول قناة هو أن طول القناة قد يتغيّر بعد إرجاع استدعاء الدالة len. وسبب قلّة استخدامنا الدالة cap لفحص سعة قناة هو أن سعة القناة غالباً معروفة أو غير مهمة.

لكن هناك بالفعل بعض السيناريوهات التي نحتاج فيها إلى استخدام الدالتين. فمثلاً، أحياناً نريد استقبال جميع القيم المخزّنة في قناة غير مغلقة c لن يرسل إليها أحد قيماً بعد الآن، ويمكننا حينها استخدام الشيفرة التالية لاستقبال القيم المتبقية.


// افترض أن الـgoroutine الحالية هي الـgoroutine
// الوحيدة التي تحاول استقبال قيم من
// القناة c في الوقت الحالي.
for len(c) > 0 {
	value := <-c
	// استخدم القيمة ...
}

يمكننا أيضاً استخدام آلية الاستقبال المحاوِل المُقدَّمة أدناه للقيام بالمهمة نفسها. وكفاءة الطريقتين متقاربة تقريباً. وميزة آلية الاستقبال المحاوِل أن الـgoroutine الحالية غير مطلوب منها أن تكون الـgoroutine المستقبِلة الوحيدة.

أحياناً، قد تريد goroutine كتابة بعض القيم إلى قناة مخزّنة c حتى تمتلئ دون الدخول في حالة الحجب في النهاية، وكانت الـgoroutine هي المرسِلة الوحيدة للقناة، ويمكننا حينها استخدام الشيفرة التالية للقيام بهذه المهمة.


for len(c) < cap(c) {
	c <- aValue
}

بالطبع، يمكننا أيضاً استخدام آلية الإرسال المحاوِل المُقدَّمة أدناه للقيام بالمهمة نفسها.

حجب الـgoroutine الحالية إلى الأبد

آلية select ميزة فريدة في Go. وهي تجلب العديد من الأنماط والحِيَل للبرمجة المتزامنة. ولمعرفة قواعد تنفيذ الشيفرة في آلية select، يُرجى قراءة مقالة القنوات في Go.

يمكننا استخدام كتلة select فارغة select{} لحجب الـgoroutine الحالية إلى الأبد. وهذه أبسط حالة استخدام لآلية select. وفي الواقع، يمكن استبدال بعض استخدامات for {time.Sleep(time.Second)} في بعض الأمثلة أعلاه بـselect{}.

عموماً، يُستخدم select{} لمنع الـgoroutine main من الخروج، لأن البرنامج كله سيخرج أيضاً إذا خرجت الـgoroutine main.

مثال:


package main

import "runtime"

func DoSomething() {
	for {
		// قم بشيء ما ...

		runtime.Gosched() // تجنّب الجشع
	}
}

func main() {
	go DoSomething()
	go DoSomething()
	select{}
}

بالمناسبة، هناك بعض الطرق الأخرى لإبقاء goroutine في حالة الحجب إلى الأبد. لكن طريقة select{} هي الأبسط.

الإرسال المحاوِل والاستقبال المحاوِل

تُسمّى كتلة select ذات فرع default واحد وفرع case واحد فقط عملية إرسال محاوِل أو استقبال محاوِل، حسب ما إذا كانت عملية القناة التالية للكلمة المفتاحية case عملية إرسال إلى قناة أم عملية استقبال من قناة.

إذا كانت العملية التالية للكلمة المفتاحية case عملية إرسال، فتُسمّى كتلة select عملية إرسال محاوِل. وإذا كانت عملية الإرسال ستحجب، فسيُنفَّذ فرع default (فشل الإرسال)، وإلا ينجح الإرسال ويُنفَّذ فرع case الوحيد.
إذا كانت العملية التالية للكلمة المفتاحية case عملية استقبال، فتُسمّى كتلة select عملية استقبال محاوِل. وإذا كانت عملية الاستقبال ستحجب، فسيُنفَّذ فرع default (فشل الاستقبال)، وإلا ينجح الاستقبال ويُنفَّذ فرع case الوحيد.

لا تحجب عمليتا الإرسال المحاوِل والاستقبال المحاوِل أبداً.

يُجري مصرِّف Go القياسي تحسينات خاصة لكتل select الخاصة بالإرسال المحاوِل والاستقبال المحاوِل، وكفاءة تنفيذها أعلى بكثير من كتل select متعددة الفروع.

في ما يلي مثال يوضّح كيفية عمل الإرسال المحاوِل والاستقبال المحاوِل.


package main

import "fmt"

func main() {
	type Book struct{id int}
	bookshelf := make(chan Book, 3)

	for i := 0; i < cap(bookshelf) * 2; i++ {
		select {
		case bookshelf <- Book{id: i}:
			fmt.Println("succeeded to put book", i)
		default:
			fmt.Println("failed to put book")
		}
	}

	for i := 0; i < cap(bookshelf) * 2; i++ {
		select {
		case book := <-bookshelf:
			fmt.Println("succeeded to get book", book.id)
		default:
			fmt.Println("failed to get book")
		}
	}
}

ناتج البرنامج أعلاه:


succeed to put book 0
succeed to put book 1
succeed to put book 2
failed to put book
failed to put book
failed to put book
succeed to get book 0
succeed to get book 1
succeed to get book 2
failed to get book
failed to get book
failed to get book

ستعرض الأقسام الفرعية التالية المزيد من حالات استخدام الإرسال المحاوِل والاستقبال المحاوِل.

فحص ما إذا كانت قناة مغلقة دون حجب الـgoroutine الحالية

بافتراض ضمان عدم إرسال أي قيم إلى قناة قطّ (ولن تُرسل)، يمكننا استخدام الشيفرة التالية لفحص ما إذا كانت القناة مغلقة بالفعل (بشكل متزامن وآمن) دون حجب الـgoroutine الحالية، حيث T نوع عنصر نوع القناة المقابل.


func IsClosed(c chan T) bool {
	select {
	case <-c:
		return true
	default:
	}
	return false
}

تُستخدم طريقة فحص ما إذا كانت قناة مغلقة على نطاق واسع في برمجة Go المتزامنة للتحقق مما إذا كان إشعار قد وصل. ويُرسَل الإشعار بإغلاق القناة في goroutine أخرى.

تحديد الذروة/الاندفاع

يمكننا تنفيذ تحديد الذروة بالجمع بين استخدام القنوات كسيمافورات عدّية والإرسال/الاستقبال المحاوِل. وغالباً ما يُستخدم تحديد الذروة (أو تحديد الاندفاع) لتحديد عدد الطلبات المتزامنة دون حجب أي طلب.

في ما يلي نسخة معدَّلة من المثال الأخير في قسم استخدام القنوات كسيمافورات عدّية.


...
	// يمكن خدمة 10 عملاء على الأكثر في الوقت نفسه
	bar24x7 := make(Bar, 10)
	for customerId := 0; ; customerId++ {
		time.Sleep(time.Second)
		customer := Consumer{customerId}
		select {
		case bar24x7 <- customer: // حاول الدخول إلى الحانة
			go bar24x7.ServeConsumer(customer)
		default:
			log.Print("customer#", customerId, " goes elsewhere")
		}
	}
...

طريقة أخرى لتنفيذ حالة استخدام فوز أول استجابة

كما ذُكر أعلاه، يمكننا استخدام آلية select (الإرسال المحاوِل) مع قناة مخزّنة سعتها واحد (على الأقل) لتنفيذ حالة استخدام فوز أول استجابة. مثلاً،


package main

import (
	"fmt"
	"math/rand"
	"time"
)

func source(c chan<- int32) {
	ra, rb := rand.Int31(), rand.Intn(3)+1
	// النوم 1 أو 2 أو 3 ثوانٍ.
	time.Sleep(time.Duration(rb) * time.Second)
	select {
	case c <- ra:
	default:
	}
}

func main() {
	// يجب أن تكون السعة 1 على الأقل.
	c := make(chan int32, 1)
	for i := 0; i < 5; i++ {
		go source(c)
	}
	rnd := <-c // لا تُستخدم إلا أول استجابة
	fmt.Println(rnd)
}

يُرجى ملاحظة أن سعة القناة المستخدمة في المثال أعلاه يجب أن تكون واحداً على الأقل، حتى لا تُفوَّت أول عملية إرسال إذا لم تكن جهة الاستقبال/الطلب قد استعدّت في الوقت المناسب.

الطريقة الثالثة لتنفيذ حالة استخدام فوز أول استجابة

في حالة استخدام فوز أول استجابة، إذا كان عدد المصادر صغيراً، مثلاً مصدرين أو ثلاثة، فيمكننا استخدام كتلة شيفرة select لاستقبال استجابات المصادر في الوقت نفسه. مثلاً،


package main

import (
	"fmt"
	"math/rand"
	"time"
)

func source() <-chan int32 {
	// يجب أن تكون c قناة مخزّنة.
	c := make(chan int32, 1)
	go func() {
		ra, rb := rand.Int31(), rand.Intn(3)+1
		time.Sleep(time.Duration(rb) * time.Second)
		c <- ra
	}()
	return c
}

func main() {
	var rnd int32
	// الحجب هنا حتى تستجيب إحدى المصادر.
	select{
	case rnd = <-source():
	case rnd = <-source():
	case rnd = <-source():
	}
	fmt.Println(rnd)
}

ملاحظة: إذا كانت القناة المستخدمة في المثال أعلاه قناة غير مخزّنة، فسيبقى هناك goroutinتان معلّقتان إلى الأبد بعد تنفيذ كتلة شيفرة select. وهذه حالة تسرّب ذاكرة.

يمكن أيضاً استخدام الطريقتين المُقدَّمتين في هذا القسم الفرعي والقسم الفرعي السابق لتنفيذ إشعارات من N إلى 1.

المهلة الزمنية

في بعض سيناريوهات الطلب-الاستجابة، ولأسباب متنوعة، قد يحتاج الطلب إلى وقت طويل للاستجابة، وأحياناً لن يستجيب أبداً. وفي مثل هذه الحالات، ينبغي إرجاع رسالة خطأ إلى جهة العميل باستخدام حلّ مهلة زمنية. ويمكن تنفيذ حلّ المهلة الزمنية هذا بآلية select.

توضّح الشيفرة التالية كيفية إرسال طلب مع مهلة زمنية.


func requestWithTimeout(timeout time.Duration) (int, error) {
	c := make(chan int)
	// قد يستغرق الحصول على الاستجابة وقتاً طويلاً.
	go doRequest(c)

	select {
	case data := <-c:
		return data, nil
	case <-time.After(timeout):
		return 0, errors.New("timeout")
	}
}

المؤقّت الدوري

يمكننا استخدام آلية الإرسال المحاوِل لتنفيذ مؤقّت دوري.


package main

import "fmt"
import "time"

func Tick(d time.Duration) <-chan struct{} {
	// من الأفضل ضبط سعة c على واحد.
	c := make(chan struct{}, 1)
	go func() {
		for {
			time.Sleep(d)
			select {
			case c <- struct{}{}:
			default:
			}
		}
	}()
	return c
}

func main() {
	t := time.Now()
	for range Tick(time.Second) {
		fmt.Println(time.Since(t))
	}
}

في الواقع، هناك دالة Tick في الحزمة القياسية time توفّر الوظيفة نفسها بتنفيذ أعلى كفاءة بكثير. وينبغي استخدام تلك الدالة بدلاً منها لتبدو الشيفرة نظيفة وتعمل بكفاءة.

تحديد المعدل

عرض أحد الأقسام أعلاه كيفية استخدام الإرسال المحاوِل لتنفيذ تحديد الذروة. ويمكننا أيضاً استخدام الإرسال المحاوِل لتنفيذ تحديد المعدل (بمساعدة مؤقّت دوري). وفي الممارسة، غالباً ما يكون تحديد المعدل لتجنّب تجاوز الحصة واستنزاف الموارد.

يعرض ما يلي مثالاً من هذا القبيل مقتبساً من jh oh ويكي Go الرسمي. وفي هذا المثال، لن يتجاوز عدد الطلبات المُعالَجة في أي مدة دقيقة واحدة 200.


package main

import "fmt"
import "time"

type Request interface{}
func handle(r Request) {fmt.Println(r.(int))}

const RateLimitPeriod = time.Minute
const RateLimit = 200 // 200 طلب على الأكثر في دقيقة واحدة

func handleRequests(requests <-chan Request) {
	quotas := make(chan time.Time, RateLimit)

	go func() {
		tick := time.NewTicker(RateLimitPeriod / RateLimit)
		defer tick.Stop()
		for t := range tick.C {
			select {
			case quotas <- t:
			default:
			}
		}
	}()

	for r := range requests {
		<-quotas
		go handle(r)
	}
}

func main() {
	requests := make(chan Request)
	go handleRequests(requests)
	// time.Sleep(time.Minute)
	for i := 0; ; i++ {requests <- i}
}

في الممارسة، غالباً ما نستخدم تحديد المعدل وتحديد الذروة/الاندفاع معاً.

التبديلات

تعلّمنا من مقالة القنوات في Go أن إرسال قيمة إلى قناة nil أو استقبال قيمة منها عمليتان حاجبتان. وبالاستفادة من هذه الحقيقة، يمكننا تغيير القنوات المتضمنة في عمليات case في كتلة شيفرة select للتأثير في اختيار الفروع في كتلة شيفرة select.

في ما يلي مثال آخر لكرة الطاولة منفَّذ باستخدام آلية select. وفي هذا المثال، يكون أحد متغيرَي القناة المتضمنين في كتلة select هو nil. ولن يُختار فرع case المقابل للقناة nil بالتأكيد. ويمكننا اعتبار فروع case هذه في حالة إيقاف. وفي نهاية كل خطوة من الحلقة، تُبدَّل حالتا التشغيل/الإيقاف لفرعَي case.


package main

import "fmt"
import "time"
import "os"

type Ball uint8
func Play(playerName string, table chan Ball, serve bool) {
	var receive, send chan Ball
	if serve {
		receive, send = nil, table
	} else {
		receive, send = table, nil
	}
	var lastValue Ball = 1
	for {
		select {
		case send <- lastValue:
		case value := <- receive:
			fmt.Println(playerName, value)
			value += lastValue
			if value < lastValue { // تجاوز سعة
				os.Exit(0)
			}
			lastValue = value
		}
		// بدّل التشغيل/الإيقاف.
		receive, send = send, receive
		time.Sleep(time.Second)
	}
}

func main() {
	table := make(chan Ball)
	go Play("A:", table, false)
	Play("B:", table, true)
}

في ما يلي مثال آخر (غير متزامن) أبسط بكثير ويعرض أيضاً تأثير التبديل. وسيطبع هذا المثال 1212... عند تشغيله. وليس له فائدة كبيرة في الممارسة، وهو معروض هنا لغرض التعلّم فقط.


package main

import "fmt"
import "time"

func main() {
	for c := make(chan struct{}, 1); true; {
		select {
		case c <- struct{}{}:
			fmt.Print("1")
		case <-c:
			fmt.Print("2")
		}
		time.Sleep(time.Second)
	}
}

التحكم في أوزان احتمال تنفيذ الشيفرة

يمكننا تكرار فرع case في كتلة شيفرة select لزيادة وزن احتمال تنفيذ الشيفرة المقابلة.

مثال:


package main

import "fmt"

func main() {
	foo, bar := make(chan struct{}), make(chan struct{})
	close(foo); close(bar) // لغرض العرض
	x, y := 0.0, 0.0
	f := func(){x++}
	g := func(){y++}
	for i := 0; i < 100000; i++ {
		select {
		case <-foo: f()
		case <-foo: f()
		case <-bar: g()
		}
	}
	fmt.Println(x/y) // نحو 2
}

احتمال استدعاء الدالة f نحو ضعف احتمال استدعاء الدالة g.

select من عدد ديناميكي من الحالات

مع أن عدد الفروع في كتلة select ثابت، يمكننا استخدام الوظائف الموفَّرة في الحزمة القياسية reflect لبناء كتلة select في وقت التشغيل. ويمكن أن تحتوي كتلة select المُنشأة ديناميكياً على عدد اعتباطي من فروع case. لكن يُرجى ملاحظة أن طريقة الانعكاس أقل كفاءة من الطريقة الثابتة.

توفّر الحزمة القياسية reflect أيضاً الدالتين TrySend وTryRecv لتنفيذ كتل select ذات فرع case واحد بالإضافة إلى default.

التعامل مع تدفّقات البيانات

سيقدّم هذا القسم بعض حالات استخدام التعامل مع تدفّقات البيانات باستخدام القنوات.

عموماً، يتكوّن تطبيق تدفّق بيانات من وحدات كثيرة. وتؤدي الوحدات المختلفة مهامّ مختلفة. وقد تملك كل وحدة عاملاً واحداً أو عدة عمال (goroutines) يؤدون بشكل متزامن المهمة نفسها المحدّدة لتلك الوحدة. وفي ما يلي قائمة ببعض أمثلة مهامّ الوحدات في الممارسة:

توليد/جمع/تحميل البيانات.
تقديم/حفظ البيانات.
حساب/تحليل البيانات.
التحقق من البيانات/تصفيتها.
تجميع/تقسيم البيانات
تركيب/تفكيك البيانات.
تكرار/نشر البيانات.

قد يستقبل عامل في وحدة بيانات من عدة وحدات أخرى كمدخلات، ويرسل بيانات لخدمة وحدات أخرى كمخرجات. وبعبارة أخرى، يمكن أن تكون الوحدة مستهلكة بيانات ومنتجة بيانات معاً. وتُسمّى الوحدة التي ترسل بيانات فقط إلى بعض الوحدات الأخرى ولا تستقبل بيانات منها قطّ وحدة منتِجة فقط. وتُسمّى الوحدة التي تستقبل بيانات فقط من بعض الوحدات الأخرى ولا ترسل بيانات إليها قطّ وحدة مستهلكة فقط.

وتشكّل وحدات كثيرة معاً نظام تدفّق بيانات.

سيعرض ما يلي بعض تنفيذات عمال وحدات تدفّق البيانات. وهذه التنفيذات لغرض الشرح، لذا فهي بسيطة جداً وقد لا تكون فعّالة.

توليد/جمع/تحميل البيانات

هناك كل أنواع الوحدات المنتِجة فقط. وقد ينتج عامل وحدة منتِجة فقط تدفّق بيانات

بتحميل ملف، أو قراءة قاعدة بيانات، أو الزحف على الويب.
بجمع كل أنواع المقاييس من نظام برمجي أو كل أنواع العتاد.
بتوليد أرقام عشوائية.
إلخ.

هنا، نستخدم مولّد أرقام عشوائية كمثال. وتُرجع دالة التوليد نتيجة واحدة لكنها لا تأخذ أي وسائط.


import (
	"crypto/rand"
	"encoding/binary"
)

func RandomGenerator() <-chan uint64 {
	c := make(chan uint64)
	go func() {
		rnds := make([]byte, 8)
		for {
			_, err := rand.Read(rnds)
			if err != nil {
				close(c)
				break
			}
			c <- binary.BigEndian.Uint64(rnds)
		}
	}()
	return c
}

في الواقع، مولّد الأرقام العشوائية هو future/promise متعدد الإرجاع.

قد يغلق منتِج البيانات قناة تدفّق المخرجات في أي وقت لإنهاء توليد البيانات.

تجميع البيانات

يجمع عامل وحدة تجميع البيانات عدة تدفّقات بيانات من نوع البيانات نفسه في تدفّق واحد. وبافتراض أن نوع البيانات هو int64، ستجمع الدالة التالية عدداً اعتباطياً من تدفّقات البيانات في تدفّق واحد.


func Aggregator(inputs ...<-chan uint64) <-chan uint64 {
	out := make(chan uint64)
	for _, in := range inputs {
		go func(in <-chan uint64) {
			for {
				out <- <-in // <=> out <- (<-in)
			}
		}(in)
	}
	return out
}

ينبغي أن يراعي التنفيذ الأفضل ما إذا كان تدفّق مدخلات قد أُغلق أم لا. (وهذا صالح أيضاً لتنفيذات عمال الوحدات الأخرى التالية.)


import "sync"

func Aggregator(inputs ...<-chan uint64) <-chan uint64 {
	output := make(chan uint64)
	var wg sync.WaitGroup
	for _, in := range inputs {
		wg.Add(1)
		go func(int <-chan uint64) {
			defer wg.Done()
			// إذا أُغلقت in، فستنتهي
			// الحلقة في النهاية.
			for x := range in {
				output <- x
			}
		}(in)
	}
	go func() {
		wg.Wait()
		close(output)
	}()
	return output
}

إذا كان عدد تدفّقات البيانات المجمَّعة صغيراً جداً (اثنان أو ثلاثة)، فيمكننا استخدام كتلة select لتجميع تدفّقات البيانات هذه.


// افترض أن عدد تدفّقات المدخلات اثنان.
...
	output := make(chan uint64)
	go func() {
		inA, inB := inputs[0], inputs[1]
		for {
			select {
			case v := <- inA: output <- v
			case v := <- inB: output <- v
			}
		}
	}
...

تقسيم البيانات

يؤدي عامل وحدة تقسيم البيانات عكس ما يؤديه عامل وحدة تجميع البيانات. ويسهل تنفيذ عامل تقسيم، لكن عوامل التقسيم ليست مفيدة كثيراً ونادراً ما تُستخدم في الممارسة.


func Divisor(input <-chan uint64, outputs ...chan<- uint64) {
	for _, out := range outputs {
		go func(o chan<- uint64) {
			for {
				o <- <-input // <=> o <- (<-input)
			}
		}(out)
	}
}

تركيب البيانات

يدمج عامل تركيب البيانات عدة قطع بيانات من تدفّقات بيانات مدخلات مختلفة في قطعة بيانات واحدة.

في ما يلي مثال لعامل تركيب، تدمج فيه قيمتان من نوع uint64 من تدفّق واحد وقيمة واحدة من نوع uint64 من تدفّق آخر في قيمة uint64 جديدة. وبالطبع، أنواع عناصر قنوات هذه التدفّقات مختلفة عموماً في الممارسة.


func Composer(inA, inB <-chan uint64) <-chan uint64 {
	output := make(chan uint64)
	go func() {
		for {
			a1, b, a2 := <-inA, <-inB, <-inA
			output <- a1 ^ b & a2
		}
	}()
	return output
}

تفكيك البيانات

تفكيك البيانات هو العملية العكسية لتركيب البيانات. ويأخذ تنفيذ دالة عامل التفكيك وسيط تدفّق بيانات مدخلات واحداً ويُرجع عدة نتائج لتدفّقات بيانات. ولن تُعرض أمثلة على تفكيك البيانات هنا.

تكرار/نشر البيانات

يمكن اعتبار تكرار (نشر) البيانات تفكيكاً خاصاً للبيانات. إذ تُكرَّر قطعة بيانات وتُرسَل كل نسخة من البيانات المكرَّرة إلى تدفّق بيانات مخرجات مختلف.

مثال:


func Duplicator(in <-chan uint64) (<-chan uint64, <-chan uint64) {
	outA, outB := make(chan uint64), make(chan uint64)
	go func() {
		for x := range in {
			outA <- x
			outB <- x
		}
	}()
	return outA, outB
}

حساب/تحليل البيانات

تتفاوت وظائف وحدات حساب البيانات وتحليلها وكل منها محدّد جداً. وعموماً، تحوّل دالة عامل في مثل هذه الوحدات كل قطعة بيانات مدخلات إلى قطعة بيانات مخرجات أخرى.

لغرض عرض بسيط، يُعرض هنا مثال لعامل يقلب كل بت في كل قيمة uint64 منقولة.


func Calculator(in <-chan uint64, out chan uint64) (<-chan uint64) {
	if out == nil {
		out = make(chan uint64)
	}
	go func() {
		for x := range in {
			out <- ^x
		}
	}()
	return out
}

التحقق من البيانات/تصفيتها

تُهمل وحدة التحقق من البيانات أو تصفيتها بعض البيانات المنقولة في تدفّق. فمثلاً، تُهمل دالة العامل التالية جميع الأعداد غير الأولية.


import "math/big"

func Filter0(input <-chan uint64, output chan uint64) <-chan uint64 {
	if output == nil {
		output = make(chan uint64)
	}
	go func() {
		bigInt := big.NewInt(0)
		for x := range input {
			bigInt.SetUint64(x)
			if bigInt.ProbablyPrime(1) {
				output <- x
			}
		}
	}()
	return output
}

func Filter(input <-chan uint64) <-chan uint64 {
	return Filter0(input, nil)
}

يُرجى ملاحظة أن كلاً من التنفيذين يُستخدم في أحد المثالين الأخيرين المعروضين أدناه.

تقديم/حفظ البيانات

عموماً، تكون وحدة تقديم البيانات أو حفظها هي وحدة المخرجات الأخيرة في نظام تدفّق البيانات. ويُقدَّم هنا فقط عامل بسيط يطبع كل قطعة بيانات مستَقبَلة من تدفّق المدخلات.


import "fmt"

func Printer(input <-chan uint64) {
	for x := range input {
		fmt.Println(x)
	}
}

تجميع نظام تدفّق البيانات

الآن، لنستخدم دوال عمال الوحدات أعلاه لتجميع عدة أنظمة تدفّق بيانات. وتجميع نظام تدفّق بيانات هو مجرد إنشاء بعض عمال الوحدات المختلفة، وتحديد تدفّقات المدخلات لكل عامل.

مثال 1 لنظام تدفّق البيانات (خط أنابيب خطي):


package main

... // دوال العمال المعلنة أعلاه.

func main() {
	Printer(
		Filter(
			Calculator(
				RandomGenerator(), nil,
			),
		),
	)
}

يُصوَّر نظام تدفّق البيانات أعلاه في المخطّط التالي.

مثال 2 لنظام تدفّق البيانات (خط أنابيب برسم بياني موجّه لا دوري):


package main

... // دوال العمال المعلنة أعلاه.

func main() {
	filterA := Filter(RandomGenerator())
	filterB := Filter(RandomGenerator())
	filterC := Filter(RandomGenerator())
	filter := Aggregator(filterA, filterB, filterC)
	calculatorA := Calculator(filter, nil)
	calculatorB := Calculator(filter, nil)
	calculator := Aggregator(calculatorA, calculatorB)
	Printer(calculator)
}

يُصوَّر نظام تدفّق البيانات أعلاه في المخطّط التالي.

قد تكون طوبولوجيا تدفّق البيانات الأكثر تعقيداً رسوماً بيانية اعتباطية. فمثلاً، قد يكون لنظام تدفّق بيانات عدة مخرجات نهائية. لكن أنظمة تدفّق البيانات ذات الطوبولوجيا الرسومية الدورية نادراً ما تُستخدم في الواقع.

من المثالين أعلاه، يمكننا أن نجد أنه من السهل جداً والبديهي بناء أنظمة تدفّق بيانات بالقنوات.

ومن المثال الأخير، يمكننا أن نجد أنه بمساعدة المجمِّعات، يسهل تنفيذ التجميع والتوزيع (fan-in وfan-out) لعدد عمال وحدة معيّنة.

في الواقع، يمكننا استخدام قناة بسيطة لتحلّ محلّ دور المجمِّع. فمثلاً، يستبدل المثال التالي المجمِّعين بقناتين.


package main

... // دوال العمال المعلنة أعلاه.

func main() {
	c1 := make(chan uint64, 100)
	Filter0(RandomGenerator(), c1) // filterA
	Filter0(RandomGenerator(), c1) // filterB
	Filter0(RandomGenerator(), c1) // filterC
	c2 := make(chan uint64, 100)
	Calculator(c1, c2) // calculatorA
	Calculator(c1, c2) // calculatorB
	Printer(c2)
}

يُصوَّر نظام تدفّق البيانات المعدَّل في المخطّط التالي.

لا تراعي الشروح أعلاه لأنظمة تدفّق البيانات كثيراً كيفية إغلاق تدفّقات البيانات. يُرجى قراءة هذه المقالة لمعرفة كيفية إغلاق القنوات بأناقة.

حالات استخدام القنوات

استخدام القنوات كـFutures/Promises

إرجاع قنوات استقبال فقط كنتائج

تمرير قنوات إرسال فقط كمعطيات

فوز أول استجابة

المزيد من صيغ الطلب-الاستجابة

استخدام القنوات للإشعارات

إشعار من 1 إلى 1 بإرسال قيمة إلى قناة

إشعار من 1 إلى 1 باستقبال قيمة من قناة

إشعارات من N إلى 1 ومن 1 إلى N

إشعارات البثّ (من 1 إلى N) بإغلاق قناة

المؤقّت: إشعار مجدول

استخدام القنوات كأقفال mutex

استخدام القنوات كسيمافورات عدّية

الحوار (Ping-Pong)

قناة مغلَّفة داخل قناة

فحص أطوال القنوات وسعاتها

حجب الـgoroutine الحالية إلى الأبد

الإرسال المحاوِل والاستقبال المحاوِل

فحص ما إذا كانت قناة مغلقة دون حجب الـgoroutine الحالية

تحديد الذروة/الاندفاع

طريقة أخرى لتنفيذ حالة استخدام فوز أول استجابة

الطريقة الثالثة لتنفيذ حالة استخدام فوز أول استجابة

المهلة الزمنية

المؤقّت الدوري

تحديد المعدل

التبديلات

التحكم في أوزان احتمال تنفيذ الشيفرة

select من عدد ديناميكي من الحالات

التعامل مع تدفّقات البيانات

توليد/جمع/تحميل البيانات

تجميع البيانات

تقسيم البيانات

تركيب البيانات

تفكيك البيانات

تكرار/نشر البيانات

حساب/تحليل البيانات

التحقق من البيانات/تصفيتها

تقديم/حفظ البيانات

تجميع نظام تدفّق البيانات