Shulou(Shulou.com)06/02 Report--

This article introduces the knowledge of "how to use sync/errGroup in Go concurrent programming". Many people will encounter this dilemma in the operation of actual cases, so let the editor lead you to learn how to deal with these situations. I hope you can read it carefully and be able to achieve something!

one。 Order

This article is a supplement to concurrent programming because there is a current project that probably requires the creation of the same number of consumers (goroutine) to consume data from different partitions based on the number of partition of kafka, but there is always some reason that consumer creation in one partition fails, but consumer creation in other partitions fails. The initial logic is to ignore the logic of partition failure and collect the successfully created partition consumers for getting messages for data processing. The code is not shown here.

The problem is clear: if only one consumer creation fails when initializing the partition consumer, the program should panic and exit even if the initialization fails.

But at the time of the design, the consumer was responsible for obtaining the data from a topic upstream of the kafka, and then after the data was processed, the processed data was sent to the downstream topic through the producer. Because at that time, the code coupling was so heavy that it could not be done through initialization, so the consumer could not be created only after the producer was started. This leads to a mix of creating consumers-> getting data-- > processing data. It was not until recently that I had time to solve this problem.

For example, three partitions create consumers of three partitions, and the consumers of each partition retrieve data from their own partitions. The three partitions initially use waitGroup to control the creation of three partitions, and the subsequent logic will be executed only after all three partitions have been created. But waitgroup is not a good solution: as long as one goroutine goes wrong and we no longer wait for another goroutine, the default creation of partition consumers fails, so we think of errGroup at this time.

II. ErrGroup

ErrGroup is the basic extension library for google open source. Download before using it.

Go get-u golang.org/x/sync2.1 function signature

Type Group struct {/ / contains filtered or unexported fields} func WithContext (ctx context.Context) (* Group, context.Context) func (g * Group) Go (f func () error) func (g * Group) Wait () error

The whole package has only one Group structure.

You can create a group with cancellation through WaitContext

Passing a func () error into the Go method will launch a goroutine to process

Wait is similar to the Wait method of WaitGroup. It exits after waiting for all goroutine to finish, and the error returned is an error err.

three。 Source code

3.1 Group

The cancel method cancel func () / / reuse WaitGroup wg sync.WaitGroup / / of type Group struct {/ / context is used to ensure that only one error errOnce sync.Once / / saves the first returned error err error} 3.2 WaitContext

Func WithContext (ctx context.Context) (* Group, context.Context) {ctx, cancel: = context.WithCancel (ctx) return & Group {cancel: cancel}, ctx}

WithContext is to use WithCancel to create a cancelable context to save the cancel assignment to Group, and then return the context back.

Note that there is a pit. In the later code, do not pass the ctx as the parent context to the downstream, because if the errgroup is cancelled, the context will be useless, which will lead to errors in downstream reuse.

3.3 Go

Func (g * Group) Go (f func () error) {g.wg.Add (1) go func () {defer g.wg.Done () / / determine whether cancel if err: = f () needs to be executed by executing the error value returned by the passed-in anonymous function. Err! = nil {/ / this is important to ensure that errors are executed only once g.errOnce.Do (func () {g.err = err if g.cancel! = nil {g.cancel ()})}} ()} ()}

The Go method is an encapsulation, which is equivalent to the enhancement of the go keyword. It starts a Ctrip, and then uses waitgroup to control whether it ends. If there is a non-nil error, it will save it, and if there is a cancel, it will call cancel to cancel it and make the ctx return.

3.4 Wait

Func (g * Group) Wait () error {g.wg.Wait () if g.cancel! = nil {g.cancel ()} return g.err}

The Wait method actually calls WaitGroup to wait. If there is a cancel, call it.

four。 Case

Based on errgroup to achieve a http server startup and shutdown, as well as linux signal signal registration and processing, to ensure that one exit, all logout exit.

Package main import ("context"fmt"log"net/http"os"os/signal"syscall"time"golang.org/x/sync/errgroup") func main () {g, ctx: = errgroup.WithContext (context.Background ()) mux: = http.NewServeMux () mux.HandleFunc ("/ ping", func (w http.ResponseWriter, r * http.Request) {_ _ = w.Write ([] byte ("pong")}) / / simulate a single service error to exit serverOut: = make (chan struct {}) mux.HandleFunc ("/ shutdown", func (w http.ResponseWriter, r * http.Request) {serverOut

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.