Parallelität mit Goroutinen

In diesem Artikel werde ich über eine kleine Go-Bibliothek sprechen, die wir geschrieben haben, um die Verwendung von Goroutine zu vereinfachen und klaren Code zu schreiben. Mal sehen, wie dieses Paket herauskam.

Als Trendyol Indexing Team hatten wir ein in Scala geschriebenes Projekt, das Daten von vielen Microservices sammelte. Um die Arbeitsbelastung des aktuellen Teams und die Komplexität des Projekts zu reduzieren, haben wir das Indexing Offer-Team gebildet, dem auch ich angehöre.

Mein Team wird für das preis- und bestandsbezogene Geschäft zuständig sein. Das bedeutet, dass das neue Team vorerst mit 8 Microservices und 4 verschiedenen Teams arbeiten wird. Natürlich haben wir mit dieser Aufteilung nicht nur eine API, sondern auch die Konsumenten des jeweiligen Unternehmens in das neue Team verschoben.

Wir haben mehrere APIs in Go geschrieben und haben festgestellt, dass Go hinsichtlich Bootzeit, Speichernutzung und CPU-Auslastung eine gute Leistung erbringt. Mit diesen APIs erzielen wir mit wenigen Ressourcen einen hohen Durchsatz. Aus diesem Grund haben wir uns entschieden, unsere neue API mit Go zu schreiben.

Kommen wir zurück zu unserem Hauptthema. Beim Senden von Anfragen an Dienste treten die folgenden Situationen auf.

• Wir müssen die Anfrage in kleinere Teile aufteilen und diese senden.
• Einige Dienste sind unabhängig voneinander, sodass wir parallele Anfragen senden können.

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

Kleinere Stücke und Versand

Dieses Beispiel zeigt, wie eine Anfrage aufgeteilt und parallel gesendet wird. Die Ergebnisse aller parallelen Operationen werden in einer Liste gesammelt und die Funktion gibt einen Fehler zurück, wenn eine Operation fehlschlägt.

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
}

Der folgende Codeblock zeigt, wie eine parallele Anfrage an zwei verschiedene Dienste gesendet wird

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

Vielen Dank an Emre Odabas für seine Unterstützung und Ermutigung beim Schreiben dieses Artikels.