7 шаблонов конкурентности в Go, которые вам стоит знать

Конкурентность — одна из самых мощных возможностей Go, и её освоение критически важно для создания масштабируемых и эффективных приложений. В этой статье мы рассмотрим 7 шаблонов конкурентности в Go, которые помогут вам писать надёжный код.

1. Пул воркеров

Описание: Пул воркеров создаёт фиксированное количество горутин, которые обрабатывают задачи из общей очереди. Этот шаблон полезен для управления количеством одновременно выполняемых задач и оптимизации использования ресурсов.

package main  import ( "fmt" "sync" "time" )  func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) { defer wg.Done() for job := range jobs { fmt.Printf("Воркер %d начал задачу %d\n", id, job) time.Sleep(time.Second) fmt.Printf("Воркер %d завершил задачу %d\n", id, job) results <- job * 2 } }  func main() { const numJobs = 5 const numWorkers = 3 jobs := make(chan int, numJobs) results := make(chan int, numJobs) var wg sync.WaitGroup  for i := 1; i <= numWorkers; i++ { wg.Add(1) go worker(i, jobs, results, &wg) }  for j := 1; j <= numJobs; j++ { jobs <- j } close(jobs)  wg.Wait() close(results)  for result := range results { fmt.Println("Результат:", result) } } 

Реальный сценарий: Веб-сервер, обрабатывающий входящие HTTP-запросы, где каждый запрос обрабатывается воркером из пула.

2. Fan-Out / Fan-In

Описание: Fan-Out запускает несколько горутин для параллельной обработки данных, а Fan-In собирает результаты в единый канал. Этот шаблон полезен для параллельной обработки с последующей агрегацией.

package main  import ( "fmt" "sync" )  func producer(id int, ch chan<- int, wg *sync.WaitGroup) { defer wg.Done() for i := 0; i < 5; i++ { ch <- i fmt.Printf("Производитель %d создал %d\n", id, i) } } func consumer(id int, in <-chan int, out chan<- int, wg *sync.WaitGroup) { defer wg.Done() for v := range in { out <- v * 2 fmt.Printf("Потребитель %d обработал %d\n", id, v) } } func main() { numProducers := 2 numConsumers := 2 input := make(chan int, 10) output := make(chan int, 10) var wg sync.WaitGroup for i := 1; i <= numProducers; i++ { wg.Add(1) go producer(i, input, &wg) } wg.Wait() close(input) for i := 1; i <= numConsumers; i++ { wg.Add(1) go consumer(i, input, output, &wg) } wg.Wait() close(output) for result := range output { fmt.Println("Результат:", result) } } 

Реальный сценарий: Конвейер обработки данных, где разные этапы выполняются разными наборами воркеров.

3. Пайплайн

Описание: Пайплайн объединяет несколько этапов обработки, где каждый этап выполняет преобразование данных и передаёт их следующему этапу. Подходит для последовательной обработки данных.

package main  import "fmt"  func stage1(nums []int) <-chan int { out := make(chan int) go func() { for _, n := range nums { out <- n } close(out) }() return out }  func stage2(in <-chan int) <-chan int { out := make(chan int) go func() { for n := range in { out <- n * 2 } close(out) }() return out }  func stage3(in <-chan int) <-chan int { out := make(chan int) go func() { for n := range in { out <- n + 1 } close(out) }() return out }  func main() { nums := []int{1, 2, 3, 4, 5} c1 := stage1(nums) c2 := stage2(c1) c3 := stage3(c2) for result := range c3 { fmt.Println(result) } } 

Реальный сценарий: Система обработки изображений, где изображение проходит через этапы масштабирования, фильтрации и кодирования.

4. Публикация-Подписка

Описание: Шаблон «Публикация-Подписка» позволяет публиковать сообщения для нескольких подписчиков. Полезен в системах, где разные сервисы должны независимо реагировать на события.

package main  import ( "fmt" "sync" "time" )  type PubSub struct { mu       sync.Mutex channels map[string][]chan string }  func NewPubSub() *PubSub { return &PubSub{ channels: make(map[string][]chan string), } }  func (ps *PubSub) Subscribe(topic string) <-chan string { ch := make(chan string) ps.mu.Lock() ps.channels[topic] = append(ps.channels[topic], ch) ps.mu.Unlock() return ch }  func (ps *PubSub) Publish(topic, msg string) { ps.mu.Lock() for _, ch := range ps.channels[topic] { ch <- msg } ps.mu.Unlock() }  func (ps *PubSub) Close(topic string) { ps.mu.Lock() for _, ch := range ps.channels[topic] { close(ch) } ps.mu.Unlock() }  func main() { ps := NewPubSub()  subscriber1 := ps.Subscribe("news") subscriber2 := ps.Subscribe("news")  var wg sync.WaitGroup wg.Add(2)  go func() { defer wg.Done() for msg := range subscriber1 { fmt.Println("Подписчик 1 получил:", msg) } }()  go func() { defer wg.Done() for msg := range subscriber2 { fmt.Println("Подписчик 2 получил:", msg) } }()  ps.Publish("news", "Срочные новости!") ps.Publish("news", "Ещё новости!")  time.Sleep(time.Second) ps.Close("news") wg.Wait() } 

Реальный сценарий: Система обмена сообщениями, где сервисы подписываются на определённые типы событий.

5. Select с таймаутом

Описание: Использование select с таймаутом позволяет избежать бесконечных блокировок. Полезно, когда нужно выполнить действие или прервать операцию, если она занимает слишком много времени.

package main  import ( "fmt" "time" )  func main() { c := make(chan string)  go func() { time.Sleep(2 * time.Second) c <- "результат" }()  select { case res := <-c: fmt.Println("Получено:", res) case <-time.After(1 * time.Second): fmt.Println("Таймаут") } } 

Реальный сценарий: Сетевой клиент, который пытается подключиться к серверу и останавливается, если сервер не отвечает вовремя.

6. Семафор

Описание: Семафор ограничивает количество горутин, которые могут одновременно обращаться к ресурсу. Полезен для управления конкурентностью и предотвращения перегрузки ресурсов.

package main  import ( "fmt" "sync" "time" )  func worker(id int, sem chan struct{}, wg *sync.WaitGroup) { defer wg.Done() sem <- struct{}{} // Захват семафора fmt.Printf("Воркер %d начал\n", id) time.Sleep(time.Second) fmt.Printf("Воркер %d завершил\n", id) <-sem // Освобождение семафора }  func main() { const numWorkers = 5 const maxConcurrent = 2 sem := make(chan struct{}, maxConcurrent) var wg sync.WaitGroup  for i := 1; i <= numWorkers; i++ { wg.Add(1) go worker(i, sem, &wg) }  wg.Wait() } 

Реальный сценарий: Пул подключений к базе данных, где одновременно допускается ограниченное количество подключений.

7. Ограничение частоты

Описание: Ограничение частоты управляет скоростью обработки событий с помощью тикера. Полезно, когда нужно контролировать частоту выполнения задач, например, запросов к API.

package main  import ( "fmt" "time" )  func main() { rate := time.Second ticker := time.NewTicker(rate) defer ticker.Stop()  requests := make(chan int, 5) for i := 1; i <= 5; i++ { requests <- i } close(requests)  for req := range requests { <-ticker.C // Ожидание следующего тика fmt.Println("Обработка запроса", req) } }

Реальный сценарий: Шлюз API, ограничивающий количество запросов, которые пользователь может сделать за определённый период.

Заключение

Шаблоны конкурентности в Go необходимы для создания эффективных и масштабируемых приложений. Освоение этих шаблонов позволит вам эффективно управлять конкурентностью, оптимизировать использование ресурсов и повысить производительность ваших приложений.