Concomitanza con Goroutines

In questo articolo parlerò di una piccola libreria go che abbiamo scritto per rendere più facile l'uso di goroutine e per scrivere codice chiaro. Vediamo come è uscito questo pacchetto.

In qualità di Trendyol Indexing Team, avevamo un progetto scritto in Scala che raccoglieva dati da molti microservizi. Per ridurre il carico di lavoro dell'attuale team e la complessità del progetto, abbiamo formato il team Indexing Offer, di cui faccio parte anch'io.

Il mio team sarà responsabile degli affari relativi ai prezzi e alle azioni. Ciò significa che per il momento il nuovo team lavorerà con 8 microservizi e 4 team diversi. Naturalmente, con questa divisione, abbiamo spostato nel nuovo team non solo un'API ma anche i consumatori dell'attività in questione.

Abbiamo diverse API scritte in Go e abbiamo osservato che Go funziona bene in termini di tempo di avvio, utilizzo della memoria e utilizzo della CPU. Otteniamo un throughput elevato da queste API con poche risorse. Di conseguenza, abbiamo deciso di scrivere la nostra nuova API utilizzando Go.

Torniamo al nostro argomento principale. Abbiamo le seguenti situazioni durante l'invio di richieste ai servizi.

• Dobbiamo dividere la richiesta in blocchi più piccoli e inviarli.
• Alcuni servizi sono indipendenti l'uno dall'altro, quindi possiamo inviare richieste parallele.

Esempio in Scala

// 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)
}

Pezzi più piccoli e inviarli

Questo esempio mostra come suddividere una richiesta e inviarla in parallelo. I risultati di tutte le operazioni parallele vengono raccolti in un elenco e la funzione restituisce un errore se un'operazione non riesce.

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
}

Il blocco di codice seguente mostra come inviare una richiesta parallela a due diversi servizi

/***--- 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
}

Grazie a Emre Odabas per il suo supporto e incoraggiamento nello scrivere questo articolo.