コミュニティ 学ぶ ツールライブラリ レジャー
検索
日本語
ホームページ > バックエンド開発 > Golang > Go 言語の同時実行性と WorkerPool - パート 2

Go 言語の同時実行性と WorkerPool - パート 2

Go语言进阶学习
リリース： 2023-07-21 10:47:45
転載
1153 人が閲覧しました

コード構造

一般的な WorkerPool パッケージを作成しました。Workers を使用します。ビジネスで必要な同時実行性に基づいてタスクを処理します。ディレクトリ構造を見てみましょう: 

workerpool
├── pool.go
├── task.go
└── worker.go
ログイン後にコピー

ワーカープール ディレクトリはプロジェクトのルート ディレクトリにあります。タスクは処理する必要がある単一の作業単位であり、ワーカーはタスクを実行するために使用される単純なワーカー関数であり、プールはワーカーの作成と管理に使用されます。

実装

最初にタスク コードを確認します。 

// workerpool/task.go

package workerpool

import (
 "fmt"
)

type Task struct {
 Err  error
 Data interface{}
 f    func(interface{}) error
}

func NewTask(f func(interface{}) error, data interface{}) *Task {
 return &Task{f: f, Data: data}
}

func process(workerID int, task *Task) {
 fmt.Printf("Worker %d processes task %v\n", workerID, task.Data)
 task.Err = task.f(task.Data)
}
ログイン後にコピー

タスクは、処理タスクを保存する単純な構造です。データが必要です。タスクを作成する際にはDataと実行する関数fを渡し、process()関数でタスクを処理します。タスクを処理する際には関数 f にパラメータとして Data を渡し、実行結果を Task.Err に保存します。

Worker がタスクを処理する方法を見てみましょう: 

// workerpool/worker.go

package workerpool

import (
 "fmt"
 "sync"
)

// Worker handles all the work
type Worker struct {
 ID       int
 taskChan chan *Task
}

// NewWorker returns new instance of worker
func NewWorker(channel chan *Task, ID int) *Worker {
 return &Worker{
  ID:       ID,
  taskChan: channel,
 }
}

// Start starts the worker
func (wr *Worker) Start(wg *sync.WaitGroup) {
 fmt.Printf("Starting worker %d\n", wr.ID)

 wg.Add(1)
 go func() {
  defer wg.Done()
  for task := range wr.taskChan {
   process(wr.ID, task)
  }
 }()
}
ログイン後にコピー

Worker ID と保留中のタスクを保存するチャネルを含む小さな Worker 構造を作成しました。 Start() メソッドで、for range を使用して taskChan からタスクを読み取り、処理します。ご想像のとおり、複数のワーカーがタスクを同時に実行できます。

workerPool

Task と Worker を実装してタスクを処理しますが、何かが足りないようです。これらのワーカーとワーカーの生成の責任者は誰ですか?タスクを割り当てますか?送信しますか?答えは「ワーカープール」です。 

// workerpoo/pool.go

package workerpool

import (
 "fmt"
 "sync"
 "time"
)

// Pool is the worker pool
type Pool struct {
 Tasks   []*Task

 concurrency   int
 collector     chan *Task
 wg            sync.WaitGroup
}

// NewPool initializes a new pool with the given tasks and
// at the given concurrency.
func NewPool(tasks []*Task, concurrency int) *Pool {
 return &Pool{
  Tasks:       tasks,
  concurrency: concurrency,
  collector:   make(chan *Task, 1000),
 }
}

// Run runs all work within the pool and blocks until it's
// finished.
func (p *Pool) Run() {
 for i := 1; i <= p.concurrency; i++ {
  worker := NewWorker(p.collector, i)
  worker.Start(&p.wg)
 }

 for i := range p.Tasks {
  p.collector <- p.Tasks[i]
 }
 close(p.collector)

 p.wg.Wait()
}
ログイン後にコピー

上記のコードでは、プールはすべての保留中のタスクを保存し、タスクの同時処理のために同時実行数と同じ数のゴルーチンを生成します。共有キャッシュ チャネル -- ワーカー間のコレクター。

したがって、このワーク プールを実行すると、必要な数のワーカーを生成でき、コレクター チャネルがワーカー間で共有されます。次に、for range を使用してタスクを読み取り、読み取ったタスクをコレクターに書き込みます。コルーチン間の同期を実現するには、sync.WaitGroup を使用します。良い解決策が見つかったので、それをテストしてみましょう。 

// main.go

package main

import (
 "fmt"
 "time"

 "github.com/Joker666/goworkerpool/workerpool"
)

func main() {
 var allTask []*workerpool.Task
 for i := 1; i <= 100; i++ {
  task := workerpool.NewTask(func(data interface{}) error {
   taskID := data.(int)
   time.Sleep(100 * time.Millisecond)
   fmt.Printf("Task %d processed\n", taskID)
   return nil
  }, i)
  allTask = append(allTask, task)
 }

 pool := workerpool.NewPool(allTask, 5)
 pool.Run()
}
ログイン後にコピー

上記のコードは 100 個のタスクを作成し、5 つの同時プロセスを使用してこれらのタスクを処理します。

输出如下：

Worker 3 processes task 98
Task 92 processed
Worker 2 processes task 99
Task 98 processed
Worker 5 processes task 100
Task 99 processed
Task 100 processed
Took ===============> 2.0056295s
ログイン後にコピー

处理 100 个任务花费了 2s，如何我们将并发数提高到 10，我们会看到处理完所有任务只需要大约 1s。

我们通过实现 workerPool 构建了一个健壮的解决方案，具有并发性、错误处理、数据处理等功能。这是个通用的包，不耦合具体的实现。我们可以使用它来解决一些大问题。

进一步扩展：后台处理任务

实际上，我们还可以进一步扩展上面的解决方案，以便 worker 可以在后台等待我们投递新的任务并处理。为此，代码需要做一些修改，Task 结构体保持不变，但是需要小改下 Worker，看下面代码：

// workerpool/worker.go

// Worker handles all the work
type Worker struct {
 ID       int
 taskChan chan *Task
 quit     chan bool
}

// NewWorker returns new instance of worker
func NewWorker(channel chan *Task, ID int) *Worker {
 return &Worker{
  ID:       ID,
  taskChan: channel,
  quit:     make(chan bool),
 }
}

....

// StartBackground starts the worker in background waiting
func (wr *Worker) StartBackground() {
 fmt.Printf("Starting worker %d\n", wr.ID)

 for {
  select {
  case task := <-wr.taskChan:
   process(wr.ID, task)
  case <-wr.quit:
   return
  }
 }
}

// Stop quits the worker
func (wr *Worker) Stop() {
 fmt.Printf("Closing worker %d\n", wr.ID)
 go func() {
  wr.quit <- true
 }()
}
ログイン後にコピー

Worker 结构体新加 quit channel，并且新加了两个方法。StartBackgorund() 在 for 循环里使用 select-case 从 taskChan 队列读取任务并处理，如果从 quit 读取到结束信号就立即返回。Stop() 方法负责往 quit 写入结束信号。

添加完这两个新的方法之后，我们来修改下 Pool：

// workerpool/pool.go

type Pool struct {
 Tasks   []*Task
 Workers []*Worker

 concurrency   int
 collector     chan *Task
 runBackground chan bool
 wg            sync.WaitGroup
}

// AddTask adds a task to the pool
func (p *Pool) AddTask(task *Task) {
 p.collector <- task
}

// RunBackground runs the pool in background
func (p *Pool) RunBackground() {
 go func() {
  for {
   fmt.Print("⌛ Waiting for tasks to come in ...\n")
   time.Sleep(10 * time.Second)
  }
 }()

 for i := 1; i <= p.concurrency; i++ {
  worker := NewWorker(p.collector, i)
  p.Workers = append(p.Workers, worker)
  go worker.StartBackground()
 }

 for i := range p.Tasks {
  p.collector <- p.Tasks[i]
 }

 p.runBackground = make(chan bool)
 <-p.runBackground
}

// Stop stops background workers
func (p *Pool) Stop() {
 for i := range p.Workers {
  p.Workers[i].Stop()
 }
 p.runBackground <- true
}
ログイン後にコピー

Pool 结构体添加了两个成员：Workers 和 runBackground，Workers 保存所有的 worker，runBackground 用于维持 pool 存活状态。

添加了三个新的方法，AddTask() 方法用于往 collector 添加任务；RunBackground() 方法衍生出一个无限运行的 goroutine，以便 pool 维持存活状态，因为 runBackground 信道是空，读取空的 channel 会阻塞，所以 pool 能维持运行状态。接着，在协程里面启动 worker；Stop() 方法用于停止 worker，并且给 runBackground 发送停止信号以便结束 RunBackground() 方法。

我们来看下具体是如何工作的。

如果是在现实的业务场景中，pool 将会与 HTTP 服务器一块运行并消耗任务。我们通过 for 无限循环模拟这种这种场景，如果满足某一条件，pool 将会停止。

// main.go

...

pool := workerpool.NewPool(allTask, 5)
go func() {
 for {
  taskID := rand.Intn(100) + 20

  if taskID%7 == 0 {
   pool.Stop()
  }

  time.Sleep(time.Duration(rand.Intn(5)) * time.Second)
  task := workerpool.NewTask(func(data interface{}) error {
   taskID := data.(int)
   time.Sleep(100 * time.Millisecond)
   fmt.Printf("Task %d processed\n", taskID)
   return nil
  }, taskID)
  pool.AddTask(task)
 }
}()
pool.RunBackground()
ログイン後にコピー

当执行上面的代码时，我们就会看到有随机的 task 被投递到后台运行的 workers，其中某一个 worker 会读取到任务并完成处理。当满足某一条件时，程序便会停止退出。

以上がGo 言語の同時実行性と WorkerPool - パート 2の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

関連ラベル：
言語を移動 同時
ソース：Go语言进阶学习
前の記事：Go 言語のオブジェクト指向設計原則とベスト プラクティス 次の記事：Go と Goroutines を使用した効率的な同時機械学習トレーニング システムの実装
このウェブサイトの声明
この記事の内容はネチズンが自主的に寄稿したものであり、著作権は原著者に帰属します。このサイトは、それに相当する法的責任を負いません。盗作または侵害の疑いのあるコンテンツを見つけた場合は、admin@php.cn までご連絡ください。
著者別の最新記事
最新の問題
MySQL: 「<ここの変数またはサブクエリ> として...からテキストを選択します」 次のデータを含む次のテーブルがあります: id text language 1 german text german 2 english text english 私が望むのは、次...
から 2024-04-06 19:25:54
0
2
529
透過的なブラウザを起動して、Web アプリケーションと PWA のエクスペリエンスを向上させます [静的 HTML、CSS、JavaScript をサポート] [ユーザーが Web アプリケーション/PWA でデスクトップの壁紙を表示できるようにします] 私は現在、ソケットを使用した Web アプリケーション (チャット) を開発しているジュニアのフロントエンド開発者兼デザイナーです。私と友人が使用している言語は、Scala、HTM...
から 2024-04-05 13:50:12
0
1
1367
正規表現を使用した HTML での単語カウント これもこれと同じ問題です。しかし、私はJavaScriptを使用していないので、「innerText」は私にとって解決策ではなく、正規表現が一連の処理を行わずに/(<.*?&...
から 2024-04-01 22:44:06
0
1
416
ユーザーがあるページで画像をアップロードまたは変更し、別のページに表示できるようにするにはどうすればよいですか? ユーザーがあるページに写真をアップロードして、別のページに表示できるようにするにはどうすればよいですか?ユーザーは画像を変更でき、他のページでも同様に変更されます。また、php/j...
から 2024-04-01 18:33:31
0
1
332
Python: Pymysql が応答なしでサイレントにハングしますか? Python を使用した最初のアプリケーション。アプリケーションが pymysql 呼び出しでハングします。このライブラリは安定していますか?ここで MySQL データベースに接続...
から 2024-04-01 10:55:37
0
1
321
関連トピック
詳細>
人気のおすすめ
人気のチュートリアル
詳細>
関連するチュートリアル
人気のおすすめ
最新のコース
最新のダウンロード
詳細>
ウェブエフェクト
公式サイト
サイト素材
フロントエンドテンプレート