Параллелизм с горутинами

В этой статье я расскажу о небольшой библиотеке go, которую мы написали, чтобы упростить использование горутины и написание понятного кода. Посмотрим, как вышел этот пакет.

Как команда Trendyol Indexing, у нас был проект, написанный на Scala, который собирал данные из многих микросервисов. Чтобы уменьшить нагрузку на текущую команду и сложность проекта, мы сформировали команду Indexing Offer, частью которой я также являюсь.

Моя команда будет отвечать за ценовой и биржевой бизнес. Это означает, что новая команда пока будет работать с 8 микросервисами и 4 разными командами. Конечно, с этим подразделением мы перенесли в новую команду не только API, но и потребителей соответствующего бизнеса.

У нас есть несколько API, написанных на Go, и мы заметили, что Go хорошо работает с точки зрения времени загрузки, использования памяти и использования ЦП. Мы получаем высокую пропускную способность от этих API с небольшим количеством ресурсов. В результате мы решили написать наш новый API на Go.

Вернемся к нашей основной теме. У нас бывают следующие ситуации при отправке запросов в сервисы.

• Нам нужно разбить запрос на более мелкие куски и отправить их.
• Некоторые сервисы не зависят друг от друга, поэтому мы можем отправлять параллельные запросы.

Пример на Скале

// Future.sequence
def ChunkRequestAndGetAll(ids: Array[String]): Future[Seq[String]] = {
  Future.sequence(ids.grouped(chunkSize).map(chunkedIds => {
    Future {
      APICall(chunkedIds)
    }
  }).toSeq)
}

// for-yield
def ForYieldExample() = {
  val f1: Future[String] = Future {APICall("call")}
  val f2: Future[Int] = Future {AnotherAPICall()}

  val result = for {
    r1 <- f1
    r2 <- f2
  } yield (anyFunc(r1, r2))

  Await.result(result, 5.seconds)
}

Меньшие куски и их отправка

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

func ChunkRequestAndGetAll(ids []string) ([]string, error) {
 var chunkOfIds [][]string = chunk(ids, chunkSize)
 
 ch := make(chan result, len(chunkOfIds))
 for i := range chunkOfIds {
  chunkedIds := chunkOfIds[i]
  go func() {
   response, apiErr := APICall(chunkedIds)
   ch <- result{response, apiErr}
  }()
 }

 log.Println("Waiting for results...")
 results := make([]string, len(ids))
 var err error
 for i := 0; i < len(ids); i++ {
  rs := <-ch
  if rs.err != nil {
   err = rs.err
   break
  }
  results[i] = rs.response
 }

 return results, err
}

func ChunkRequestAndGetAll(ids []string) ([]string, error) {
 var chunkOfIds [][]string = chunk(ids, chunkSize)
 
futures := make([]*future.Future[string], len(chunkOfIds))
 for i := range chunkOfIds {
  chunkedIds := chunkOfIds[i]
  futures[i] = future.Run(func() (string, error) {
   return APICall(chunkedIds)
  })
 }

 log.Println("Waiting for result...")
 results, err := future.GetAll(futures)
 return results, err
}

В приведенном ниже блоке кода показано, как отправить параллельный запрос к двум разным службам.

/***--- With directly use channel and goroutine ---***/
func DirectlyUse() error {
 ch1 := make(chan result[string])
 go func() {
  response, err := APICall("call")
  ch1 <- result[string]{response, err}
 }()

 ch2 := make(chan result[int64])
 go func() {
  response, err := AnotherAPICall()
  ch2 <- result[int64]{response, err}
 }()

 res1 := <-ch1
 if res1.err == nil {
  return res1.err
 }
 res2 := <-ch2
 if res2.err != nil {
  return res2.err
 }
 //somethings
}

/***--- With future struct ---***/
func WithFutureStruct() error {
 fut1 := future.Run(func() (string, error) {
  return APICall("call")
 })

 fut2 := future.Run(func() (int64, error) {
  return AnotherAPICall()
 })

 res1, err1 := fut1.Get()
 res2, err2 := fut2.Get()

 if err1 == nil {
  return err1
 }

 if err2 != nil {
  return err2
 }
 //somethings
}

/***--- Alternative future struct ---***/
func WithFutureStruct() error {
 fut1 := future.Run(func() (any, error) {
  return APICall("call")
 })

 fut2 := future.Run(func() (any, error) {
  return AnotherAPICall()
 })

 err := future.WaitFor(fut1, fut2)
 if err != nil {
  return err
 }
 result1 := future.GetResult[string](fut1)
 result2 := future.GetResult[int64](fut2)
 //somethings
}

Спасибо Эмре Одабасу за его поддержку и помощь в написании этой статьи.