Channel o'z-o'zicha shunchaki tipli quvur: biror narsa bir uchidan kiradi, ikkinchisidan chiqadi, va runtime topshirishni xavfsiz bajaradi. Channel'larni qiziqarli qiladigan narsa — ularni bir nechtasini bir-biriga ulaganingizda sodir bo'ladigan holatlar. Bir necha shakl Go dasturlarida shunchalik tez-tez takrorlanadiki — konveyerlar (pipeline), fan-out, fan-in va cheklangan worker pool'lar — ularni har safar concurrent ishlov kerak bo'lganda qayta ixtiro qilmasdan, nomlangan andozalar sifatida o'rganish foydali.
Konveyer — channel'lar bilan bog'langan bosqichlar ketma-ketligi, bunda har bir bosqich bir xil funksiyani bajaradigan goroutine'lar guruhi. Har bir bosqich kiruvchi channel'dan qiymatlarni oladi, biroz ish qiladi va natijalarni chiquvchi channel'ga yuboradi. Bir bosqichning chiquvchi channel'i keyingisining kiruvchi channel'iga aylanadi, shuning uchun ma'lumot konveyer bo'ylab suv bir qator quvurlardan oqib o'tgandek oqadi — nomi ham shundan.
package main
import "fmt"
// generate konveyerni qat'iy qiymatlar to'plami bilan boshlaydi, so'ng channel'ni yopadi.
func generate(nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
out <- n
}
}()
return out
}
// square bitta channel'ni iste'mol qiladi va boshqasini hosil qiladi, yo'l-yo'lakay har bir qiymatni kvadratga oshiradi.
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
out <- n * n
}
}()
return out
}
func main() {
numbers := generate(2, 3, 4)
squares := square(numbers)
for n := range squares {
fmt.Println(n) // 4, 9, 16
}
}
Shaklga e'tibor bering: har bir bosqich aynan bitta chiquvchi channel'ga egalik qiladi va yuborishni tugatgach uni yopadi, iste'mol qiluvchi bosqich esa shu yopilish sodir bo'lguncha o'zining kiruvchi channel'i ustida range qiladi. Bu kelishuv — "yuboruvchi yopadi, receiver range qiladi" — konveyerlar hech bir bosqich nechta qiymat kelishini bilishga majbur bo'lmasdan toza tugashiga imkon beradi. main aniq sonli marta aylanmaydi; u konveyer unga boshqa hech narsa qolmaganini aytguncha squaresni shunchaki bo'shatib turadi.
Konveyerlar yaxshi kompozitsiya qilinadi, chunki har bir bosqichning signaturasi bir xil ko'rinadi: <-chan T qabul qiladi, <-chan U qaytaradi. Bosqichlarni qo'shish, olib tashlash yoki qayta tartiblash uchun ikkala tomondagilarga tegish shart emas. io.Reader/io.Writer kompozitsiyasi oqim ma'lumotlari uchun shunchalik yaxshi ishlashining sababi ham shu: sizga moslashuvchanlikni aniq implementatsiya emas, ulashilgan interface beradi.
Fan-out — bir nechta goroutine'ni ishga tushirish, ularning hammasi bir xil kirish channel'idan o'qiydi, shunda ish ular orasida avtomatik taqsimlanadi: qaysi goroutine bo'sh bo'lsa, keyingi qiymatni oladi. Har bir element uchun CPU- yoki I/O-bound ish bajaradigan konveyer bosqichini aniq yuk balanslash mantig'ini yozmasdan parallellashtirish shunday amalga oshiriladi; channel'ning o'zi ish navbati vazifasini bajaradi.
package main
import "sync"
func process(n int) int {
return n * 2 // haqiqiy, ehtimol qimmat ish o'rnida
}
// fanOut `workers` ta goroutine ishga tushiradi, ularning hammasi bir xil `in` channel'idan oladi
// va ularning chiqishini bitta channel'ga birlashtiradi.
func fanOut(in <-chan int, workers int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
wg.Add(workers)
for i := 0; i < workers; i++ {
go func() {
defer wg.Done()
for n := range in {
out <- process(n)
}
}()
}
go func() {
wg.Wait()
close(out)
}()
return out
}
Ikkita tafsilot bu versiyani shunchaki ehtimoliy emas, to'g'ri qiladi. Birinchidan, har bir worker bir xil in channel'ini ulashadi — aynan shu uni "bir nechta mustaqil konveyer" emas, fan-out qiladi. Ikkinchidan, faqat bitta goroutine close(out)ni chaqiradi, va u buni faqat wg.Wait() har bir worker yuborishni tugatganini tasdiqlagandan keyin qiladi. Agar har bir worker o'z chiquvchi channel'ini alohida yopsa, ularni baribir birlashtirish uchun fan-in bosqichi kerak bo'lardi; shuning uchun barcha worker'lar bitta ulashilgan outga yozib, uni maxsus goroutine'dan aynan bir marta yopgani oddiyroq.
Fan-in — buning ko'zgudagi aksi: sizda mustaqil ravishda qiymat ishlab chiqaruvchi bir nechta channel bor va ularni bittaga birlashtirmoqchisiz. Bu konveyerning avvalgi qismida fan-out qilib, oqimlarni qayta birlashtirish kerak bo'lganda yoki aloqasiz manbalardan — masalan, bir nechta mustaqil API chaqiruvidan — natijalarni chaqiruvchi bir joyda iste'mol qilishi uchun yig'ayotganingizda paydo bo'ladi.
package main
import "sync"
func fanIn(inputs ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
wg.Add(len(inputs))
for _, in := range inputs {
go func(ch <-chan int) {
defer wg.Done()
for n := range ch {
out <- n
}
}(in)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
Bu yerdagi sync.WaitGroup fanOutdagi bilan bir xil ishni bajaradi: outni hamon nechta goroutine to'ldirayotganini kuzatib boradi, shunda yopuvchi goroutine channel'ni qachon yopish xavfsiz ekanini biladi. Bu "WaitGroup va maxsus yopuvchi goroutine" idiomasini esda tutish kerak, chunki u bir nechta goroutine yozadigan channel'ni yopishning standart usuli: bir nechta goroutine'dan yopish yoki yuborish hamon davom etayotgan paytda yopish ish vaqtida panic qiladi.
Yuqoridagi konveyerlar boshdan-oxirigacha buferlanmagan channel'lardan foydalanadi, va bu unutish emas, ataylab qilingan standart tanlov. Buferlanmagan channel yuborishni receiver tayyor bo'lguncha bloklaydi, ya'ni yuboruvchi va receiver uchrashadi: yuborish faqat receiver qiymatni haqiqatan olgandan keyin tugaydi. Bu sizga kuchli, foydali kafolat beradi: tugagan yuborish yetkazib berishni isbotlaydi, va bu tabiiy bloklanish xatti-harakati oqim nazorati vazifasini ham bajaradi, chunki sekin iste'molchi tez ishlab chiqaruvchini avtomatik ushlab turadi.
Buferli channel buni yumshatadi: make(chan int, n) receiver hali tayyor bo'lmasa ham n tagacha yuborish tugashiga imkon beradi. Bu ishlab chiqaruvchi ba'zan iste'molchidan tezroq ishlab qolishini tekislamoqchi bo'lganingizda yoki aynan nechta qiymat yuborishingizni bilib, keraksiz goroutine to'xtatilishidan qochmoqchi bo'lganingizda to'g'ri vosita. Masalan, yuqoridagi generate funksiyasi len(nums)ga teng buferdan foydalanib, square iste'mol qilishni boshlashidan oldin ham barcha yuborishlar darhol tugashiga imkon berishi mumkin.
Yodda tutish kerak bo'lgan kelishuv shuki, bufer yetkazib berish kafolatini zaiflashtiradi: tugagan yuborish faqat "qiymat buferda turibdi" degani, "kimdir uni oldi" degani emas. Agar dasturingiz to'g'riligi receiver qiymatni aslida qachon qayta ishlaganini bilishga bog'liq bo'lsa, bu talabni bufer bilan yashirmang — buferlanmagan channel yoki aniq tasdiqdan foydalaning. Buferlashni unumdorlikni yaxshilash uchun ishlating, haqiqiy sinxronizatsiya o'rnini bosuvchi sifatida emas.
Haqiqiy tizimlar channel'da abadiy kuta olmaydi: hamkor osilib qolishi mumkin, goroutine to'xtab qolishi mumkin, va kodingizga voz kechish usuli kerak. select operatori time.After yoki context deadline bilan birikkanda Goda "buni kut, lekin cheksiz emas" shunday ifodalanadi:
func readWithTimeout(ch <-chan int, timeout time.Duration) (int, error) {
select {
case v := <-ch:
return v, nil
case <-time.After(timeout):
return 0, fmt.Errorf("timed out after %s", timeout)
}
}
Lekin ko'p marta ishlaydigan siklda time.Afterni chaqirishda ehtiyot bo'ling: har bir chaqiruv u ishga tushmaguncha tozalanmaydigan yangi timer ajratadi, shuning uchun qisqa sikl kutilmagan darajada ko'p tirik timer to'plashi mumkin. Qizg'in siklda time.NewTimer bilan bitta time.Timer yarating va uni aniq reset qiling, yoki yaxshisi, deadline'li context.Contextdan foydalaning (keyingi bobda yoritiladi): shunda bekor qilish signali har iteratsiyada qayta qurilmaydi, balki ulashilgan va aniq bo'ladi.
Fan-out har bir elementga mavjud worker orqali navbat beradi, lekin ba'zan teskari nuqtai nazar kerak bo'ladi: worker'larning qat'iy o'lchamli puli, job'lar navbati va natijalar channel'i, hammasi aniq bir-biriga ulangan. Haqiqiy backend'larda uchraydigan shakl shu: cheklangan sondagi goroutine cheklangan sondagi job'larni qayta ishlaydi, shunda so'rovlar portlashi hech qachon bir xil quyi oqim resursi uchun raqobatlashuvchi minglab goroutine yaratmaydi.
type Job struct {
ID int
Data string
}
type Result struct {
JobID int
Output string
}
func workerPool(jobs <-chan Job, numWorkers int) <-chan Result {
results := make(chan Result)
var wg sync.WaitGroup
wg.Add(numWorkers)
for i := 0; i < numWorkers; i++ {
go func() {
defer wg.Done()
for job := range jobs {
results <- Result{JobID: job.ID, Output: handle(job.Data)}
}
}()
}
go func() {
wg.Wait()
close(results)
}()
return results
}
Bu aslida nomlangan tiplar va tasodifan uzatib yuborishingiz mumkin bo'lgan parametr o'rniga tanlangan, qat'iy numWorkers bilan fanOutning o'zi — nomlash farqi kodda emas, tushunchada ko'proq muhim. numWorkersni tanlashning o'zi dizayn qarori: juda kam bo'lsa, throughput'ni qo'ldan boy berasiz; juda ko'p bo'lsa, worker'lar oxir-oqibat chaqiradigan narsani — ma'lumotlar bazasi, tashqi API, disk — bosib ketishingiz mumkin. Keng tarqalgan yondashuv: CPU-bound ish uchun runtime.GOMAXPROCS(0) bilan boshlash yoki I/O-bound ish uchun quyi oqim tizimining haqiqiy sig'imiga nisbatan tajriba yo'li bilan sozlangan son bilan boshlash va uni qattiq yozib qo'yish o'rniga sozlanadigan qilish.
Ochiq aytish kerak: har bir concurrent muammo konveyer talab qilmaydi. Agar sizga faqat bitta ulashilgan holatni — hisoblagich, kesh, map'ni — himoyalash kerak bo'lsa, sync.Mutex holatga egalik qiladigan va channel'lar orqali muloqot qiladigan goroutine sifatida modellashtirishdan oddiyroq va aniqroq. Channel'lar murakkabligini mustaqil ish bosqichlari orasida haqiqiy ma'lumotlar oqimi bo'lganda oqlaydi. Ularni muammoingiz shakli mos kelgani uchun tanlang, "channel'lardan foydalan" hamma joyda qo'llash kerak bo'lgan qoida bo'lgani uchun emas. Keyingi bob shu qarorning teskari tomoniga qaraydi — sync.Mutex, sync.RWMutex va atomik amallarga — va ularning har biri Go dasturida o'z o'rnini qachon oqlashini aniqroq his qilishga yordam beradi.