ابتدا ببینیم Concurrency یا هم‌زمانی چیست؟

هم‌زمانی یعنی توانایی یک برنامه برای انجام چند کار مختلف به طور هم‌زمان. این ویژگی در نرم‌افزارهای مدرن خیلی مهم است، چون نیاز داریم بخش‌های مستقلی از کد را با سرعت زیاد اجرا کنیم، بدون اینکه جریان اصلی برنامه مختل شود.

به‌عبارت ساده‌تر، هم‌زمانی یعنی این که توابع بتوانند به‌طور مستقل و جدا از هم اجرا شوند. اما Parallelism (موازی‌سازی) ویژگی زمان اجرا است که در آن دو یا چند کار دقیقاً به‌طور هم‌زمان اجرا می‌شوند. با هم‌زمانی، هدف این است که ساختار مناسبی برای برنامه تعبیه کنیم.

منبع تصویر: blog.knoldus

برای درک الگوی Fan-In Fan-Out و استفاده درست از آن، باید با مفاهیمی مثل goroutines، WaitGroups، channels و... آشنا باشید. اگر هنوز اطلاعات کافی در مورد آن‌ها ندارید، می‌توانید مقاله‌ی من را در این زمینه مطالعه کنید: شروع با Go

---

Fan Out

Fan out زمانی کاربرد دارد که چندین تابع (یا goroutine) از یک کانال مشترک داده بخوانند. خواندن تا زمانی ادامه دارد که کانال بسته شود. این ویژگی به توزیع کار بین چند worker یا کارگر کمک می‌کند تا CPU و I/O به صورت موازی استفاده شود.

مثال زیر را ببینید:

1package main  
2
3import (  
4 "fmt"  
5 "sync"  
6)  
7
8func generator(nums ...int) <-chan int {  
9 myChannel := make(chan int) // تعریف کانال  
10
11 go func() {  
12  // داده‌ها را به کانال ارسال می‌کنیم  
13  for _, val := range nums {  
14   myChannel <- val  
15  }  
16  close(myChannel)  
17 }()  
18
19 return myChannel  
20}  
21
22func main() {  
23 data1 := []int{1, 2, 3, 4, 5}  
24 data2 := []int{10, 20, 30, 40, 50}  
25 var wg sync.WaitGroup  
26
27 // کانال‌های فقط خواندنی تولید می‌شوند  
28 ch1 := generator(data1...)  
29 ch2 := generator(data2...)  
30 wg.Add(2)  
31
32 // همه داده‌ها از هر دو کانال خوانده می‌شوند  
33 go func() {  
34  for val := range ch1 {  
35   fmt.Printf("Channel1 data: %v\n", val)  
36  }  
37  wg.Done()  
38 }()  
39
40 go func() {  
41  for val := range ch2 {  
42   fmt.Printf("Channel2 data: %v\n", val)  
43  }  
44  wg.Done()  
45 }()  
46
47 wg.Wait() // منتظر می‌مانیم تا همه goroutine‌ها تمام شوند  
48}

این کد را می‌توانید به صورت زنده در Go playground اجرا کنید.

شرح کد:

• چند goroutine داریم که از کانال‌های برگشتی از تابع generator می‌خوانند.
• تابع generator یک کانال می‌سازد، داده‌ها را ارسال می‌کند و پس از اتمام کار، کانال را می‌بندد.

---

Fan In

Fan in زمانی استفاده می‌شود که یک تابع بخواهد از چند ورودی (کانال) بخواند و تا زمانی که همه بسته شوند ادامه دهد. این کار با multiplexing (ترکیب چند ورودی به یک کانال خروجی) انجام می‌شود.

فرض کنید دو فایل داریم به نام‌های text1.txt و text2.txt در همان محل کد:

1package main  
2
3import (  
4 "bufio"  
5 "fmt"  
6 "log"  
7 "os"  
8 "sync"  
9)  
10
11func readData(file string) <-chan string {  
12 f, err := os.Open(file) // باز کردن فایل  
13 if err != nil {  
14  log.Fatal(err)  
15 }  
16
17 out := make(chan string) // تعریف کانال  
18
19 fileScanner := bufio.NewScanner(f)  
20 fileScanner.Split(bufio.ScanLines) // خواندن خط به خط  
21
22 go func() {  
23  for fileScanner.Scan() {  
24   val := fileScanner.Text()  // خواندن هر خط  
25   out <- val                 // ارسال به کانال  
26  }  
27
28  close(out) // کانال را می‌بندیم بعد از خواندن کل محتوا  
29
30  err := f.Close()  
31  if err != nil {  
32   fmt.Printf("Unable to close an opened file: %v\n", err.Error())  
33   return  
34  }  
35 }()  
36
37 return out  
38}  
39
40func fanInMergeData(ch1, ch2 <-chan string) chan string {  
41 chRes := make(chan string)  
42 var wg sync.WaitGroup  
43 wg.Add(2)  
44
45 // خواندن از کانال اول  
46 go func() {  
47  for val := range ch1 {  
48   chRes <- val  
49  }  
50  wg.Done()  
51 }()  
52
53 // خواندن از کانال دوم  
54 go func() {  
55  for val := range ch2 {  
56   chRes <- val  
57  }  
58  wg.Done()  
59 }()  
60
61 go func() {  
62  wg.Wait()    // منتظر تمام شدن goroutine‌ها  
63  close(chRes) // بستن کانال خروجی  
64 }()  
65
66 return chRes  
67}  
68
69func main() {  
70 ch1 := readData("text1.txt")  
71 ch2 := readData("text2.txt")  
72
73 // داده‌ها را از چند کانال می‌خوانیم و در یک کانال مشترک می‌ریزیم — FanIn  
74 chRes := fanInMergeData(ch1, ch2)  
75
76 // پردازش داده‌های دریافتی  
77 for val := range chRes {  
78  fmt.Println(val)  
79 }  
80}

توضیح:

• داده‌ها از چند فایل به صورت جداگانه خوانده می‌شوند (هر فایل یک کانال می‌سازد).
• سپس همه داده‌ها را در یک کانال مشترک (chRes) ادغام می‌کنیم.
• از bufio.Scanner برای خواندن خط به خط فایل استفاده شده است.
• تا زمانی که همه کانال‌ها بسته شوند، حلقه خواندن ادامه دارد.

---

برای مطالعه بیشتر می‌توانید به بلاگ رسمی Go مراجعه کنید: Go blog — Pipelines و قسمت Fan-out, fan-in را بخوانید.

همچنین یک مقاله بسیار خوب دیگر در این زمینه:
GO: A BETTER FAN-OUT, FAN-IN EXAMPLE
که نمونه‌های بهتر و عملی‌تری ارائه می‌دهد.