با استفاده از maxwell تغییرات اطلاعات رو از دیتابیس اصلی می خوانیم و بر روی RabbitMQ قرار می دهیم و در نهایت یک سرویس indexer ایجاد می کنیم تا مسیج هایی که روی RabbitMQ قرار گرفته اند را پردازش و اطلاعات را بر روی Typesense ایندکس کند.
با استفاده از CDC پایپلاین میتوانیم اطلاعات را به صورت به صورت آنی یا نزدیک به آنی, از یک ماشین به ماشین دیگری منتقل کنیم بدون اینکه ترافیکی روی دیتابیس ایجاد کنیم.
با استفاده از Debezium تغییرات اطلاعات رو از دیتابیس اصلی می خوانیم و بر روی RabbitMQ قرار می دهیم و در نهایت یک سرویس indexer ایجاد می کنیم تا مسیج هایی که روی RabbitMQ قرار گرفته اند را پردازش و اطلاعات را بر روی Typesense ایندکس کند.
با مفهوم همروندی و پردازش موازی آشنا میشیم و یاد میگیریم چطور با استفاده از گوروتین (goroutine) و چنل ها (channels) دستورات را به صورت همروند و یا موازی اجرا و کنترل کنیم.
در ساده ترین حالت, کارها میتوانند به صورت ترتیبی (sequential) اجرا شوند, بدین معنی که اگر سه کار (task) داشته باشیم CPU آنها را پشت سر هم اجرا میکند.
خیلی وقت ها, یک برنامه منتظر اتفاق افتادن رویدادی است و در این بازه زمانی از CPU استفاده نمیکند. برای مثال، برنامه ممکن است منتظر تکمیل عملیات دریافت یا ارسال دادهها بر روی دیسک (I/O) باشد. در چنین شرایطی، اجرای برنامه میتواند موقتاً متوقف شده و برنامهی دیگری روی CPU اجرا شود و پس از پایان عملیات انتظار، اجرای برنامهی اولیه از ادامه پیدا میکند.
نمونه هایی از عملیاتی که باعث ایجاد انتظار در برنامه ها میشوند به صورت زیر هستند:
ارسال یا دریافت اطلاعات بر روی شبکه
نوشتن یا خواندن اطلاعات بر روی دیسک
این کارها معمولا در دسته ی کارهای I/O قرار میگیرند.
برای اینکه زمان بیکاری CPU را کاهش دهیم میتوانیم برنامهها به صورت همروند (concurrent) اجرا کنیم. بنابراین همروندی این قابلیت را به ما میدهد تا چندین برنامه که دارای وقفه هستند را بر روی یک CPU زمانبندی کنیم.
به عنوان مثال فکر کنید سه برنامهی A و B و C داریم که هرکدام کارهای زیر را انجام میدهند:
برنامه A:
A1: خواندن فایل
A2: درخواست به سرویس B
A3: پردازش اطلاعات
برنامه B:
B1: دریافت درخواست
B2: پردازش اطلاعات
B3: ارسال پاسخ
برنامه C:
C1: خواندن اطلاعات از روی دیسک
C2: پردازش اطلاعات
C3: ذخیره اطلاعات در دیتابیس
اگر این سه برنامه را بخواهیم بر روی یک CPU به صورت sequential اجرا کنیم, به صورت زیر بر روی اجرا خواهند شد:
Sequential running apps
حالا اگه این برنامه ها را به صورت همروند (concurrent) اجرا کنیم CPU زمان کمتری بیکار خواهد بود و نتیجه به صورت زیر اجرا خواهند شد:
در زمان هایی که عملیات IO انجام میشود, برنامه را از روی CPU خارج میکنیم و برنامه ای دیگر را بر روی CPU اجرا میکنیم. به این عملیات context switching میگویند. در هنگام context switching اطلاعات مرتبط با اجرای برنامه بر روی Ram قرار میگیرد و برای ادامهی برنامه, دوباره این اطلاعات بر روی CPU قرار میگیرد.
در مثال بالا هنگامی که یک برنامه در حال اجرا بر روی CPU است, عملیات IO مرتبط با برنامهی دیگر توسط قسمتی از کامپیوتر (برای مثال دیسک یا کارت شبکه) انجام میشود و وقتی که عملیات IO انجام شد, دوباره برنامهی اولیه ادامه پیدا میکند.
در صورتی چندین CPU داشته باشیم میتوانیم برنامه ها را به صورت موازی یا همزمان (parallel) اجرا کنیم. برای مثال اگر سه CPU داشته باشیم میتوانیم هرکدام از برنامه ها را بر روی یک CPU اجرا کنیم.
Parallel running apps
مفهوم process و thread
وقتی یک برنامه را اجرا میکنیم، سیستمعامل برای آن یک process میسازد. بنابراین یک process یک برنامه در حال اجرا همراه با تمام منابعی است که نیاز دارد.
یک process یا پروسه معمولا شامل بخش های زیر است:
Process
├── Code (program instructions): کد برنامه
├── Heap (dynamic memory): حافظهای که runtime میگیرد
├── Stack: حافظه تابعها
├── Open files
├── OS resources: فایل و سوکتها
└── Threads: واحدهای اجرای داخل پروسه
هر پروسه حافظه جداگانه درون رم دارد. بنابراین پروسه های مختلف دارای حافظه های جداگانه ای هستند.
Thread کوچکترین واحد اجرایی داخل یک process است که سیستم عامل میتواند آن را اجرا یا متوقف کند. اگر هر پروسه (process) را مانند یک کارخانه در نظر بگیریم, thread ها مانند کارگران آن کارخانه هستند. بدین معنی که process منابع را نگهداری میکند و thread کد را اجرا میکند.
هر thread یا نخ معمولا دارای بخشهای زیر است:
Thread
├── Program Counter
├── Registers
└── Stack
حافظهی یک پروسه (process) با thread ها به اشتراک گذاشته میشود. بنابراین تمامی thread های مرتبط با یک پروسه دارای حافظهی مشترک هستند. بنابراین موارد زیر در بین thread ها به اشتراک گذاشته میشود:
متغیرهای سراسری (global)
مموری هیپ (heap)
فایلها و سوکتها
اما هر thread دارای stack جداگانه ای است بنابراین هر thread دارای متغیرهای لوکال خودش است که توسط باقی thread ها قابل دسترسی نیست. حافظه stack معمولا در هنگام فراخوانی توابع مورد استفاده قرار میگیرد.
سیستم عامل thread ها را بر روی CPU زمانبندی و اجرا میکند و CPU با استفاده از Program Counter میفهمد دستور بعدی کدام است. بنابراین Thread یک context اجرایی است که مشخص میکند CPU از کدام دستور برنامه و با چه شرایطی (state) ادامه دهد.
به صورت مفصل تر context switching بدین معنی است سیستم عامل در هنگامی که درون یک thread وقفه ایجاد میشود آن را درون رم قرار میدهد و بعدتر هنگامی که آمادهی ادامه پیدا کردن باشد, دوباره برای ادامه پیدا کردن بر روی CPU لود میکند.
ساختن یک thread عملی پر هزینه است زیرا یکسری کارها توسط سیستم عامل باید انجام شود و همچنین stack که برای یک thread در نظر گرفته میشود بزرگ است برای مثال سایز stack برای یک thread در سیستم عامل لینوکس به صورت پیش فرض 8MB است! بنابراین اگه شما 16GB حافظه رم داشته میتوانید حداکثر 200 تا thread ایجاد کنید.
گوروتین (Goroutine)
تابعهایی که با دستور `go` فراخوانی میشوند، گوروتین (goroutine) نام دارند. گولنگ این گوروتینها را مدیریت میکند و آنها را بین threadها که روی هستههای CPU اجرا میشوند، توزیع میکند. در مقایسه با threadها، گوروتینها بسیار سبکتر (lightweight) هستند؛ به همین دلیل میتوان صدها یا حتی هزاران گوروتین ایجاد کرد.
گوروتین ها به صورت ذاتی به صورت همروند (concurrent) اجرا میشوند و در صورتی که چندین CPU وجود داشته باشد میتوانند به صورت موازی (parallel) نیز اجرا شوند.
برای ایجاد یک goroutine کافیست از دستور go برای فراخوانی یک تابع استفاده کنیم:
package main
import (
"fmt"
"time"
)
func main() {
go say("Go is awesome!")
go say("I love Go")
say("Hello stranger!")
// wait for 500 miliseconds
time.Sleep(500 * time.Millisecond)
}
func say(phrase string) {
fmt.Println(phrase)
}
در مثال بالا دو تابع را درون گوروتین های مجزا فراخوانی کردیم. بنابراین سه فراخوانی اول که با استفاده از دستور go انجام شدن به صورت مجزا اجرا میشوند و در نهایت فراخوانی say("Hello stranger!") انجام میشود.
چرا از دستور time.Sleep استفاده کردیم؟ بیاید مثال بالا رو یکبار بدون time.Sleep داشته باشیم.
package main
import "fmt"
func main() {
go say("Go is awesome!")
go say("I love Go")
say("Hello stranger!")
}
func say(phrase string) {
fmt.Println(phrase)
}
گوروتین ها به صورت مجزا اجرا میشوند, بنابراین بعد از اجرا شدن دستور say("Hello Stranger!") , کار تابع main تمام میشود و برنامه قبل از اینکه دو گوروتین دیگر فرصت اجرا شدن داشته باشند به کار خود خاتمه میدهد!
تابع main هم در واقع یک گوروتین است، اما بهصورت ضمنی (implicit) وقتی برنامه شروع میشود اجرا میشود. بنابراین در این حالت سه گوروتین داریم:
main
go say("I love Go")
say("Hello strager!")
تنها نکته این است که وقتی اجرای main تمام شود، کل برنامه متوقف میشود و در نتیجه همهی گوروتینهای دیگر نیز پایان پیدا میکنند. بنابراین میتوانیم از دستور time.Sleep استفاده کنیم تا گولنگ فرصت پیدا کند گوروتینها را بر روی Threadها زمانبندی کند تا توسط CPU اجرا شوند.
استفاده از sync.WaitGroup
استفاده از time.Sleep برای صبر کردن تا پایان گوروتینها ایدهی خوبی نیست، چون نمیتوانیم دقیقاً پیشبینی کنیم اجرای گوروتینها چقدر طول میکشد. یک راه بهتر استفاده از sync.WaitGroup است. در sync.WaitGroup یک شمارنده وجود دارد که با فراخوانی متد Add افزایش و با فراخوانی متد Done کاهش پیدا میکند. تابع Wait تا زمانی که مقدار شمارنده برابر با صفر نباشد صبر میکند.
کد مثال قبل را میتوانیم به صورت زیر بازنویسی کنیم:
package main
import (
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
wg.Add(1)
go func(wg *sync.WaitGroup) {
defer wg.Done()
say("Go is awesome!")
}(&wg)
wg.Add(1)
go func(wg *sync.WaitGroup) {
defer wg.Done()
say("I love Go")
}(&wg)
say("Hello stranger!")
wg.Wait()
}
func say(phrase string) {
fmt.Println(phrase)
}
در مثال بالا به ازای هر گوروتین شمارندهی درون متغیر wg را با فراخوانی تابع Add(1) یکی افزایش میدهیم و سپس وقتی که کار هر گوروتین تمام شد مقدار شمارنده را با فراخوانی Done یکی کاهش میدهیم. در نهایت نیز با فراخوانی Wait باعث میشویم تا برنامه منتظر اتمام کار تمامی گوروتین ها بماند. در این حالت اجرای برنامه تا زمانی که کار تمامی گوروتین ها تمام شود ادامه پیدا میکند.
در گولنگ ورژن 1.25 تابع WaitGroup.Go معرفی شد. این تابع کدی مشابه کد زیر دارد:
کد مثال قبل را میتوان با استفاده از waitGroup.Go به صورت زیر ساده سازی کرد:
package main
import (
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
wg.Go(func() {
say("Go is awesome!")
})
wg.Go(func() {
say("I love Go")
})
say("Hello stranger!")
wg.Wait()
}
func say(phrase string) {
fmt.Println(phrase)
}
چنلها (channels)
در گولنگ, گوروتینها میتوانند از طریق channelها مقادیر را به یکدیگر منتقل کنند. یک channel شبیه یک لوله است که یک goroutine میتواند چیزی را داخلش بیندازد و گوروتین دیگری آن را در طرف دیگر دریافت کند.
گوروتین های در حال ارتباط با استفاده از channel
هر چنل دارای یک دیتا تایپ است که مشخص میکند چه نوع داده ای را میتواند انتقال دهد. برای تعریف یک چنل می توانیم از chan به همراه دیتا تایپی که باید انتقال پیدا کند استفاده کنیم:
و اگر خلاف جهت چنل باشند یک پیام از چنل دریافت میشود:
msg := <- messages
وقتی یک گوروتین میخواهد مقداری را به یک channel ارسال کند یا مقداری از آن دریافت کند، اجرای آن در همان نقطه متوقف یا بلاک میشود تا عملیات تکمیل شود. در هنگام ارسال (messages <- “ping”)، گوروتین فرستنده منتظر میماند تا یک گوروتین دیگر آن مقدار را از چنل دریافت کند. در هنگام دریافت (<-messages)، گوروتین گیرنده منتظر میماند تا یک مقدار برای خواندن در چنل موجود شود.
به عبارت دیگر، چنل نقطهای است که گوروتینها در آن به هم وصل میشوند و اجرای آنها تا زمان تکمیل عملیات متوقف میشود. این همان مفهوم بلوک شدن (blocking) در زمان ارسال و دریافت است.
ارسال و دریافت در یک چنل در واقع یک جفت عملیات همزمان هستند. ارسال بدون دریافت کامل نمیشود (و همچنین بلعکس) و فقط زمانی که عمل دریافت انجام شد، هر دو گوروتین میتوانند به اجرای خود ادامه دهند.
برای درک عمیق تر گوروتین ها پیشنهاد میکنم منابع زیر رو نیز بررسی کنید: