Երկու խորհուրդ՝ Go Concurrency-ին ավելի լավ կառավարելու համար

Ահա երկու սխալներ, որոնք ես արել եմ գորուտինների հետ, և ինչպես կարող եք խուսափել դրանցից:

Go-ի ծրագրավորողներից շատերը, հավանաբար, կարող են համաձայնել, որ Go-ն անհեշտ է դարձնում միաժամանակության հասնելը: Գորութինների և ալիքների միջոցով մենք հեշտությամբ կարող ենք միասին գործարկել անկախ գործառույթներ, պլանավորել ֆոնային առաջադրանքներ և շատ ավելին:

Երբ ես աշխատում եմ Go-ում լայնածավալ համակարգերի հետ, օրհնություն է ունենալ Go-ի նրբագեղ և հարուստ աջակցությունը միաժամանակ: Դա ինձ օգնեց արագացնել տվյալների մշակումը և առավելագույնի հասցնել ապարատային ռեսուրսների օգտագործումը:

Բայց, ինչպես մի անգամ ասել է Բեն Փարքերը.

🕸 ️ Մեծ ուժով մեծ պատասխանատվություն է գալիս…

Գորութինների մասին ավելի ճշմարիտ խոսքեր երբեք չեն ասվել: Որքան էլ դրանք հզոր լինեն, հեշտ է դրանք չարաշահել։ Ես ականատես եմ եղել և մի քանի սխալներ եմ թույլ տվել գորուտինների հետ կապված։ Նրանցից ոմանք առաջացրել են կենդանի, արտադրական խնդիրներ։

Այս հոդվածում ես ձեզ հետ կկիսվեմ երկու օգտակար խորհուրդներով, որոնք կարող են օգնել ձեզ ավելի խելացի լինել գորուտինների հետ: Այս խորհուրդները ծնվել են իրական պատահարներից, այնպես որ կապվեք և եկեք սկսենք: 🏃

Օգտագործեք համաժամանակյա սահմանաչափ

Միաժամանակության օգտագործման դեպքերից մեկն այն է, որ հետին պլանում թանկ առաջադրանքներ կատարելն է՝ առանց հիմնական տրամաբանական հոսքը արգելափակելու: Այսպիսով, մեր կոդը կարող է կարգավորել նման առաջադրանքներից շատերը (գրեթե) միասին՝ փոխարենը սպասելու իրենց հերթին:

Թանկարժեք առաջադրանքի օրինակներից մեկը կլինի տվյալների պահեստում տվյալների զանգվածային տեղադրումը կամ թարմացումը:

Դիտարկենք պարզ համակարգ. Ամեն անգամ, երբ կան տվյալների փոփոխություններ, սերվերը կստեղծի հաղորդագրություններ դեպի հերթ: Այնուհետև աշխատողը կսպառի այս հաղորդագրությունները, կխմբավորի դրանք խմբաքանակներով և կուղարկի պահեստ՝ թարմացումների համար:

Աշխատողի կոդը կարող է նման լինել ստորև: Դա շատ պարզեցված է, բայց դրա էությունը կա:

type Message struct {
  // Contain data updates
}

type Worker struct {
  msgChan          chan *Message // channel to receive messages from queue
  batch            []*Message    // a batch of messages
  maxBatchSize     uint32        // the maximum size of the batch
}

func newWorker() *Worker {
  return &Worker {
    msgChan:      make(chan *Message, 10000), // an internal queue size of 10000
    maxBatchSize: 5000,
  }
}

// Start the worker
func (w *Worker) start() {
  go w.consume()   // a goroutine to consume
  go w.process()   // a goroutine to process
  
  // Do something else...
}

// Runs in a loop to consume from queue
func (w *Worker) consume() {
  for {
    // blocks until there is a message
    msg := consumeFromQueue()
    w.msgChan <- msg 
  }
}

// Runs in a loop to process messages
func (w *Worker) process() {
  var batch []*Message
  
  for {
    // blocks until there is a message
    msg := <- w.msgChan
    
    batch = w.batch
    if batch == nil {
      batch = make([]*Message, 0, w.maxBatchSize)
    }
    batch = append(batch, msg)
    
    // only commit when batch is full
    if len(batch) < w.maxBatchSize {
      continue
    }
    
    // batch update
    go func(batch *[]Message) {
      updateDataByBatch(batch)
    }(batch)
    
    w.batch = nil // start new batch
  }
}

Ուշադրություն դարձրեք, թե ինչպես է տվյալների յուրաքանչյուր խմբաքանակ մշակվում իր գորուտինայում: Այս դիզայնով հաղորդագրությունների մեծ հոսքը չի արգելափակվի, և աշխատողը կարող է դրանք մշակել շատ արագ տեմպերով:

Կարո՞ղ եք տեսնել այս դիզայնի թերությունը:

Հաղորդագրությունների յուրաքանչյուր խմբաքանակ աշխատողի մեջ հիշողություն է զբաղեցնում: Հիշողությունը կթողարկվի միայն այն բանից հետո, երբ գորուտինը կկատարվի իր խմբաքանակով, այսինքն՝ updateDataByBatch-ը վերադառնա:

Եթե ​​ուշացումներ լինեն updateDataByBatch-ում հաղորդագրությունների մեծ հոսքի ժամանակ, գորուտինները կսկսեն կուտակվել: Քանի որ դրանցից յուրաքանչյուրը պարունակում է տվյալների մեկ փաթեթ, դա կարող է առաջացնել հիշողության սպառման ավելցուկ:

Ստորև բերված է հիշողության մոնիտորինգը, որը ես ստեղծել եմ իմ աշխատողների համար: Դուք կարող եք տեսնել հիշողության օգտագործման հետևողական աճերը բարձր ուշացման դեպքերի ժամանակ: Այն նույնիսկ մոտեցավ 100% օգտագործմանը՝ առաջացնելով հիշողությունից դուրս և սերվերի խափանումներ:

Սա դասական կատարողական սխալ է, և դրա լուծումը շատ պարզ է: Մենք պարզապես պետք է կիրառենք համաժամանակյա սահմանաչափ:

Միաժամանակյա սահմանաչափը սահմանափակում է գորուտինների քանակը, որոնք ծրագիրը կարող է ունենալ առաջադրանք կատարելիս: Վերևի օրինակում մենք պետք է սահմանափակենք գորուտինների քանակը, որոնք կարող են գոյություն ունենալ updateDataByBatch-ը զանգահարելիս:

Նման սահմանափակում կարող է սահմանվել՝ օգտագործելով դատարկ struct struct{} տիպի ալիքը:

const concurrencyLimit = 300

type Worker struct {
  // ...
  wgChan chan struct{} // a channel to limit concurrency
}

func newWorker() *Worker {
  return &Worker{
    // ...
    wgChan: make(chan struct{}, concurrencyLimit), // a max of 300 goroutines
  }
}

func (w *Worker) process() {
  var batch []*Message
  
  for {
    // ...
    
    // blocks unless there is space to start a new goroutine
    w.wgChan <- struct{}{}
    
    go func(batch *[]Message) {
      defer func() {
        <- w.wgChan // goroutine is done!
      }()
      
      updateDataByBatch(batch)
    }(batch)
    
    // ...
  }
}

Վերևի բարելավված տարբերակում wg նշանակում է Սպասեք խումբ: Նոր գորուտին սկսելու համար w.wgChan <- struct{}{} պետք է հաջողվի: Եթե ​​այն արգելափակում է, դա նշանակում է, որ գորուտինների թիվն արդեն սահմանին է:

Երբ գորուտինը կատարվում է, այն դատարկում է մեկ տարածություն wgChan ալիքում հետաձգված ֆունկցիայի միջոցով: Սա թույլ է տալիս մյուս արգելափակված գորուտիններին շարունակել իրենց առաջադրանքները:

Սահմանափակելով գորուտինների քանակը՝ մենք նաև սահմանափակում ենք սպասող տվյալների խմբաքանակների քանակը, որոնք կարող են գոյություն ունենալ աշխատողում: Այդ դեպքում հիշողության գերլարումը ավելի քիչ հավանական է:

Նրբագեղորեն դադարեցրեք Goroutine-ները

Բացի խմբաքանակով տվյալների թարմացումից, մեծ քանակությամբ տվյալների զտումը նույնպես թանկ խնդիր է:

Մասնավորապես, հաշվի առնելով եզակի ID-ների մի հատվածը, ես կցանկանայի ստուգել, ​​թե արդյոք դրանցից յուրաքանչյուրը առկա է տվյալների պահպանման մեջ: Եթե ​​դրանք կան, ես ուզում եմ առբերել դրանց համապատասխան արժեքները և տեղադրել դրանք նոր հատվածում՝ նույն ցուցանիշով, ինչ իրենց ID-ները:

Միամիտ մոտեցումը կլինի յուրաքանչյուր ID-ի միջով անցնելը և դրանք պահեստում փնտրելը: Բայց մենք Go-ի մշակողներ ենք, ուստի եկեք պտտենք մի քանի գորուտիններ և միաժամանակ որոնենք յուրաքանչյուր ID:

type Item struct {
  id    int
  value string
}

func getItems(itemIDs []int) ([]*Item, error) {
  var (
    concurencyLimit = 100 // set the limit!
    wgChan          = make(chan struct{}, concurencyLimit)
  )
  
  ctx := context.Background()
  ctx, cancel := context.WithTimeout(ctx, 5*time.Second) // set timeout
  defer cancel()
  
  items := make([]*Item, len(itemIDs))
  
  for i := range itemIDs {
    select {
      case <- ctx.Done():
        return nil, errors.New("timeout!") // return error if timeout
      case wgChan <- struct{}{}:
    }
    
    go func(i int, itemID int) {
      defer func() {
        <- wgChan // goroutine done!
      }()
      
      //get item by ID, returns nil if not found
      item := getItem(itemID)
      
      // place at the same index as itemID
      items[i] = item
    }(i, itemIDs[i])
  }
  
  // wait for all goroutines to finish
  for c := 0; c < concurrencyLimit; c++ {
    wgChan <- struct{}{}
  }
  
  return items, nil
}

Յուրաքանչյուր itemID-ի համար մենք կսկսենք նոր գորուտին` տվյալների պահպանման մեջ համապատասխան տարրը որոնելու համար: Նախքան ֆունկցիայի դուրս գալը, մենք պետք է համոզվենք, որ թռիչքի բոլոր գորուտինները դուրս են եկել:

Սա հայտնի է որպես նրբագեղ դադարեցում: Դրան կարելի է հասնել ստորև բերված օղակով:

Ժամանակի ցանկացած պահի մենք կարող ենք առավելագույնը concurrencyLimit գորուտիններ ունենալ: Երբ հանգույցն ավարտվում է առանց խցանման, դա երաշխավորում է, որ բոլոր գորուտինները դուրս են եկել: Այնուհետև գործառույթը կարող է ապահով կերպով դուրս գալ:

Ի՞նչ կլինի, եթե չլինի նրբագեղ դադարեցում: Բացի թերի արդյունքներ վերադարձնելուց, ֆունկցիան, ամենայն հավանականությամբ, խուճապի կմատնվի ինդեքսից դուրս գտնվող սխալի պատճառով:

Դա պայմանավորված է նրանով, որ թռիչքի ցանկացած գորուտին կփորձի ինդեքսավորել items հատվածը i դիրքում: Բայց քանի որ ֆունկցիան արդեն դուրս է եկել, items հատվածն այլևս գոյություն չունի:

Ահա թեստ ձեզ համար: Ես հեռացրել եմ որոշ կոդ՝ ֆունկցիան խելագարված դարձնելու համար: Դադարեցրեք այստեղ, եթե ցանկանում եք պարզել խնդիրը:

getItems ֆունկցիան նրբագեղորեն չի դադարեցվել, երբ համատեքստի ժամանակն ավարտվում է 5 վայրկյանից հետո: Դա խուճապի կմատնվի, եթե որևէ գորութին հետաձգի տվյալների բեռնումը պահեստից:

Սա ցույց է տալիս, թե որքան կարևոր է լիովին հասկանալ, թե ինչպես է յուրաքանչյուր գորուտին իրեն պահում ծրագրում՝ սկզբից մինչև վերջ:

Վերջնական մտքեր

Միաժամանակյա ծրագրավորումը դժվար է դարձնում այն, որ մեր ուղեղն ավելի լավ է հասկանում այն ​​բաները, որոնք հաջորդաբար հոսում են: Դժվար է կանխատեսել անսպասելին, երբ մինի ծրագրերը գործում են ծրագրում:

Խստորեն հետևելով իմ տված երկու խորհուրդներին՝ դուք կխուսափեիք նորեկի որոշ սխալներից, որոնք կարող եք թույլ տալ գորուտինների հետ: Միակ ճանապարհը, որով դուք կարող եք ավելի լավ դառնալ միաժամանակյա ծրագրավորման մեջ, ավելի բարդ կոդերի պրակտիկան և կարդալն է:

Ի՞նչ կասեք գորուտինների հետ ձեր փորձի մասին: Ունե՞ք որևէ խորհուրդ, որը կցանկանայիք կիսվել: Ինձ տեղյակ պահեք!