آموزش استخر Goroutine مورچه‌ها

معرفی مورچگان Go

ants یک استخر goroutine با عملکرد بالا است که اجرای تعداد زیادی goroutine را پیاده‌سازی کرده و مدیریت می‌کند، اجازه می‌دهد محدودیت و استفاده مجدد منابع را داشته باشیم، به‌طوری که اجرای کارها هنگام توسعه برنامه‌های همزمان به صورت موثرتری صورت گیرد.

ویژگی‌ها

• اتوماتیک زمان‌بندی تعداد زیادی goroutine و استفاده مجدد از آن‌ها
• به‌طور منظم پاکسازی goroutine‌های منقضی برای صرفه‌جویی بیشتر از منابع
• ارائه تعداد زیادی رابط‌های مفید: ارسال کار، دریافت تعداد goroutine‌های در حال اجرا، تنظیم داینامیک اندازه استخر، آزادسازی استخر، و راه‌اندازی مجدد استخر
• کنترل مناسب بر روی panics برای جلوگیری از برنامه افت
• استفاده مجدد منابع به طور قابل توجهی مصرف حافظه را کاهش می‌دهد. در حالت وظایف همزمان دسته‌ای به مقیاس بزرگ، عملکرد بهتری نسبت به همزمانی goroutine بومی دارد
• مکانیزم غیر مسدودکننده

نحوه کار ants

نمودار جریان

تصاویر متحرک

نصب

استفاده از ورژن v1 ants:

go get -u github.com/panjf2000/ants

استفاده از ورژن v2 ants (فعال‌سازی GO111MODULE=on):

go get -u github.com/panjf2000/ants/v2

استفاده

هنگام نوشتن یک برنامه همروند Go که تعداد زیادی goroutine را راه‌اندازی می‌کند، به طور ضروری منابع سیستم (حافظه، CPU) را مصرف خواهد کرد. با استفاده از ants، شما می‌توانید یک استخر goroutine را نمونه‌سازی کنید تا goroutine‌ها را مجدداً استفاده کنید، منابع را صرفه‌جویی کنید و عملکرد را بهبود بخشید:

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"github.com/panjf2000/ants/v2"
)

var sum int32

func myFunc(i interface{}) {
	n := i.(int32)
	atomic.AddInt32(&sum, n)
	fmt.Printf("run with %d\n", n)
}

func demoFunc() {
	time.Sleep(10 * time.Millisecond)
	fmt.Println("Hello World!")
}

func main() {
	defer ants.Release()

	runTimes := 1000

	// استفاده از استخر معمولی.
	var wg sync.WaitGroup
	syncCalculateSum := func() {
		demoFunc()
		wg.Done()
	}
	for i := 0; i < runTimes; i++ {
		wg.Add(1)
		_ = ants.Submit(syncCalculateSum)
	}
	wg.Wait()
	fmt.Printf("goroutine‌های در حال اجرا: %d\n", ants.Running())
	fmt.Printf("انجام همه وظایف.\n")

	// استفاده از استخر با یک تابع،
	// تنظیم 10 برای ظرفیت استخر goroutine و 1 ثانیه برای مدت زمان منقضی.
	p, _ := ants.NewPoolWithFunc(10, func(i interface{}) {
		myFunc(i)
		wg.Done()
	})
	defer p.Release()
	// ارسال وظایف یکی‌یکی.
	for i := 0; i < runTimes; i++ {
		wg.Add(1)
		_ = p.Invoke(int32(i))
	}
	wg.Wait()
	fmt.Printf("goroutine‌های در حال اجرا: %d\n", p.Running())
	fmt.Printf("انجام همه وظایف، نتیجه %d است\n", sum)
}

پیکربندی استخر

// Option نماینده‌ی تابع اختیاری است.
type Option func(opts *Options)

// Options شامل تمام گزینه‌هایی است که هنگام نمونه‌برداری یک استخر ants اعمال خواهند شد.
type Options struct {
	// ExpiryDuration یک دوره برای گوروتین های گامطور برای پاکسازی کارگران منقضی شده است،
	// گامطور هر `ExpiryDuration` تمام کارگران را اسکن می‌کند و کارگرانی را که بیش از `ExpiryDuration` استفاده نشده اند پاک می‌کند.
	ExpiryDuration time.Duration

	// PreAlloc نشان می‌دهد که آیا باید از پیش اختصاص حافظه انجام دهد هنگام مقداردهی اولیه استخر یا نه.
	PreAlloc bool

	// حداکثر تعداد کارگوروتین مسدود شده بر روی pool.Submit.
	// 0 (مقدار پیش‌فرض) به این معنی است که هیچ محدودیتی وجود ندارد.
	MaxBlockingTasks int

	// هنگامی که Nonblocking درست باشد، Pool.Submit هرگز مسدود نخواهد شد.
	// ErrPoolOverload برگردانده می‌شود زمانی که امکان انجام Pool.Submit به  صورت یکجا نباشد.
	// هنگامی که Nonblocking درست باشد، MaxBlockingTasks ناکارامد است.
	Nonblocking bool

	// PanicHandler برای کنترل پریدگی ها از هر گوروتین کارگر استفاده می‌شود.
	// اگر nil باشد، پریدگی‌ها دوباره از گوروتین‌های کارگر پرتاب خواهد شد.
	PanicHandler func(interface{})

	// Logger لاگر شخصی سازی شده برای ثبت اطلاعات، اگر تنظیم نشود،
	// لاگر استاندارد پیش‌فرض از بسته log استفاده می‌شود.
	Logger Logger
}

// WithOptions پذیرفتن پیکربندی کل گزینه‌ها را انجام می‌دهد.
func WithOptions(options Options) Option {
	return func(opts *Options) {
		*opts = options
	}
}

// WithExpiryDuration زمان فاصله زمانی تمیز کردن گوروتین ها را تعیین می‌کند.
func WithExpiryDuration(expiryDuration time.Duration) Option {
	return func(opts *Options) {
		opts.ExpiryDuration = expiryDuration
	}
}

// WithPreAlloc نشان می‌دهد که آیا باید حافظه‌ای برای کارگران اختصاص دهد یا خیر.
func WithPreAlloc(preAlloc bool) Option {
	return func(opts *Options) {
		opts.PreAlloc = preAlloc
	}
}

// WithMaxBlockingTasks تعداد حداکثر گوروتین های مسدود شده را زمانی که به ظرفیت استخر می‌رسد تعیین می‌کند.
func WithMaxBlockingTasks(maxBlockingTasks int) Option {
	return func(opts *Options) {
		opts.MaxBlockingTasks = maxBlockingTasks
	}
}

// WithNonblocking نشان می‌دهد که هنگامی که کارگران در دسترس نباشند، استخر nil برمیگرداند.
func WithNonblocking(nonblocking bool) Option {
	return func(opts *Options) {
		opts.Nonblocking = nonblocking
	}
}

// WithPanicHandler تنظیم کننده پریدگی را مشخص می‌کند.
func WithPanicHandler(panicHandler func(interface{})) Option {
	return func(opts *Options) {
		opts.PanicHandler = panicHandler
	}
}

// WithLogger یک لاگر شخصی سازی شده را تنظیم می‌کند.
func WithLogger(logger Logger) Option {
	return func(opts *Options) {
		opts.Logger = logger
	}
}

با استفاده از انواع توابع اختیاری هنگام فراخوانی NewPool/NewPoolWithFunc، مقادیر هر مورد پیکربندی در ants.Options می‌تواند تنظیم شود و بعد استفاده شود برای سفارشی کردن استخر گوروتین.

استخر سفارشی

ants پشتیبانی از نسخه‌برداری استخر کاربر با ظرفیت خاص است؛ با فراخوانی متد NewPool، استخر جدیدی با ظرفیت مشخص می‌تواند به صورت زیر نسخه برداری شود:

p, _ := ants.NewPool(10000)

ارسال وظیفه

وظایف با فراخوانی متد ants.Submit(func()) ارسال می‌شوند:

ants.Submit(func(){})

تنظیم پویای ظرفیت استخر گوروتین

برای تنظیم پویای ظرفیت استخر گوروتین، می‌توانید از متد Tune(int) استفاده کنید:

pool.Tune(1000) // ظرفیت آن را به 1000 تنظیم کنید
pool.Tune(100000) // ظرفیت آن را به 100000 تنظیم کنید

این متد مطمئن است وظیفه‌های چند نخی را تنظیم کند.

اختصاص دادن حافظه صف گوروتین

ants به شما امکان می‌دهد که حافظه را برای کل استخر پیش‌تخصیص دهید که می‌تواند عملکرد استخر گوروتین را در برخی از سناریوهای خاص بهبود بخشد. به عنوان مثال، در یک سناریویی که نیاز به یک استخر با ظرفیت بسیار بزرگ و هر وظیفه در داخل گوروتین زمان‌بر است، پیش‌تخصیص حافظه برای صف گوروتین کاهش اختصاص حافظه غیرضروری خواهد داد.

// زمانی که این تابع را فراخوانی می‌کنید، ants تمام ظرفیت استخر را از پیش وارد می‌کند
p, _ := ants.NewPool(100000, ants.WithPreAlloc(true))

رها کردن استخر

pool.Release()

راه‌اندازی مجدد استخر

// با فراخوانی متد Reboot()، می‌توانید یک استخری را که قبلاً تخریب شده است، دوباره فعال کرده و آن را دوباره به کار انداخت.
pool.Reboot()

درباره ترتیب اجرای وظایف

ants ترتیب اجرای وظایف را تضمین نمی‌کند و اجرای وظایف به طور لزومی با ترتیب ارسال مطابقت ندارد. این به این دلیل است که ants تمام وظایف ارسال شده را به صورت همزمان پردازش می‌کند و وظایف به کارگرانی اختصاص داده خواهند شد که به صورت همزمان در حال اجرا هستند، که منجر به اجرای وظایف به صورت همزمان و غیرقطعی می‌شود.