Concurrence avec Goroutines

Dans cet article, je parlerai d'une petite bibliothèque go que nous avons écrite pour faciliter l'utilisation de goroutine et pour écrire du code clair. Voyons comment ce paquet est sorti.

En tant qu'équipe d'indexation de Trendyol, nous avions un projet écrit en Scala qui collectait des données à partir de nombreux microservices. Afin de réduire la charge de travail de l'équipe actuelle et la complexité du projet, nous avons formé l'équipe Indexing Offer, dont je fais également partie.

Mon équipe sera responsable des affaires liées aux prix et aux stocks. Cela signifie que la nouvelle équipe travaillera avec 8 microservices et 4 équipes différentes pour le moment. Bien sûr, avec cette division, nous avons déplacé non seulement une API mais aussi les consommateurs de l'entreprise concernée vers la nouvelle équipe.

Nous avons plusieurs API écrites en Go, et nous avons observé que Go fonctionne bien en termes de temps de démarrage, d'utilisation de la mémoire et d'utilisation du processeur. Nous obtenons un débit élevé de ces API avec peu de ressources. En conséquence, nous avons décidé d'écrire notre nouvelle API en utilisant Go.

Revenons à notre sujet principal. Nous avons les situations suivantes lors de l'envoi de demandes aux services.

• Nous devons diviser la demande en plus petits morceaux et les envoyer.
• Certains services sont indépendants les uns des autres, nous pouvons donc envoyer des demandes parallèles.

Exemple en 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)
}

De plus petits morceaux et de les envoyer

Cet exemple montre comment fractionner une requête et l'envoyer en parallèle. Les résultats de toutes les opérations parallèles sont rassemblés dans une liste et la fonction renvoie une erreur si une opération échoue.

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
}

Le bloc de code ci-dessous montre comment envoyer une requête parallèle à deux services différents

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

Merci à Emre Odabas pour son soutien et ses encouragements dans la rédaction de cet article.