Golangのsyncパッケージによる同期・排他制御

1 はじめに
2 sync.WaitGroup
3 sync.Mutex
4 sync.RWMutex
5 sync.Map
6 sync.Once
7 sync.Pool
8 さいごに
9 おすすめ書籍

はじめに

以前にgoroutineとchannelとContextという記事を書きましたが、今回は channel を使用しないで並行処理の同期や排他制御を簡単に行える、 sync パッケージについて学習したいと思います。

sync.WaitGroup

sync.WaitGroup は複数の Goroutine の完了を待つことが出来ます。

package main

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

func main() {
	var wg sync.WaitGroup

	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func(i int) {
			time.Sleep(2 * time.Second)
			fmt.Println("End:", i)
			wg.Done()
		}(i)
	}

	wg.Wait()
}

package main

import (

"fmt"

"sync"

"time"

)

func main() {

var wg sync.WaitGroup

for i := 0; i < 100; i++ {

wg.Add(1)

go func(i int) {

time.Sleep(2 * time.Second)

fmt.Println("End:", i)

wg.Done()

}(i)

}

wg.Wait()

}

sync.WaitGroup は複数の Goroutine の完了を待つことが出来ます。
sync.WaitGroup は基本的にはただのカウンタですが、カウンタがゼロになるまで処理を待つことができます。上記のコードだと Add(1) でカウンタをインクリメントし、 Done() でデクリメントしています。そして Wait() はカウンタがゼロになるまで待ちます。

sync.Mutex

channel 以外の方法で変数に対して異なる goroutine がアクセスしても、競合によるエラーが起きないようにするための機能です。

package main

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

func parallel(wg *sync.WaitGroup, m *sync.Mutex) {
    m.Lock()
    defer m.Unlock()

    fmt.Println("あ")
    time.Sleep(100 * time.Millisecond)
    fmt.Println("い")
    time.Sleep(100 * time.Millisecond)
    fmt.Println("う")
    time.Sleep(100 * time.Millisecond)
    fmt.Println("え")
    time.Sleep(100 * time.Millisecond)
    fmt.Println("お")
    wg.Done()
}

func main() {
    runtime.GOMAXPROCS(runtime.NumCPU())

    wg := new(sync.WaitGroup)
    m := new(sync.Mutex)
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go parallel(wg, m)
    }
    wg.Wait()
}

package main

import (

"fmt"

"runtime"

"sync"

"time"

)

func parallel(wg *sync.WaitGroup, m *sync.Mutex) {

m.Lock()

defer m.Unlock()

fmt.Println("あ")

time.Sleep(100 * time.Millisecond)

fmt.Println("い")

time.Sleep(100 * time.Millisecond)

fmt.Println("う")

time.Sleep(100 * time.Millisecond)

fmt.Println("え")

time.Sleep(100 * time.Millisecond)

fmt.Println("お")

wg.Done()

}

func main() {

runtime.GOMAXPROCS(runtime.NumCPU())

wg := new(sync.WaitGroup)

m := new(sync.Mutex)

for i := 0; i < 3; i++ {

wg.Add(1)

go parallel(wg, m)

}

wg.Wait()

}

sync.Mutex の Lock 関数と Unlock 関数で排他ロックの取得と解除ができます。排他ロックなので、排他ロックを得た goroutine が存在する場合は、ロックを得ようとする他の goroutine は処理を待ち、以下のような結果になります。

あ
い
う
え
お
あ
い
う
え
お
あ
い
う
え
お

あ

い

う

え

お

あ

い

う

え

お

あ

い

う

え

お

sync.RWMutex

sync.RWMutex の Rlock 関数と Runlock 関数で共有ロックの取得と解除ができます。どういうことなのかというと、 Rlock 関数でロックを行なった goroutine 同士であれば処理を進めることができるようになります。また、共有ロックを得た goroutine が存在する場合は、排他ロックを取得しようとする　goroutine は待ちます。

package main

import (
    "fmt"
    "math/rand"
    "strings"
    "sync"
    "time"
)

func init() {
    rand.Seed(time.Now().Unix())
}

func sleep() {
    time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
}

func reader(c chan int, m *sync.RWMutex, wg *sync.WaitGroup) {
    sleep()
    m.RLock()
    c <- 1
    sleep()
    c <- -1
    m.RUnlock()
    wg.Done()
}

func writer(c chan int, m *sync.RWMutex, wg *sync.WaitGroup) {
    sleep()
    m.Lock()
    c <- 1
    sleep()
    c <- -1
    m.Unlock()
    wg.Done()
}

func main() {
    var m sync.RWMutex
    var rs, ws int
    rsCh := make(chan int)
    wsCh := make(chan int)
    go func() {
        for {
            select {
            case n := <-rsCh:
                rs += n
            case n := <-wsCh:
                ws += n
            }
            fmt.Printf("%s%s\n", strings.Repeat("R", rs),
                    strings.Repeat("W", ws))
        }
    }()
    wg := sync.WaitGroup{}
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go reader(rsCh, &m, &wg)
    }
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go writer(wsCh, &m, &wg)
    }
    wg.Wait()
}

package main

import (

"fmt"

"math/rand"

"strings"

"sync"

"time"

)

func init() {

rand.Seed(time.Now().Unix())

}

func sleep() {

time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)

}

func reader(c chan int, m *sync.RWMutex, wg *sync.WaitGroup) {

sleep()

m.RLock()

c <- 1

sleep()

c <- -1

m.RUnlock()

wg.Done()

}

func writer(c chan int, m *sync.RWMutex, wg *sync.WaitGroup) {

sleep()

m.Lock()

c <- 1

sleep()

c <- -1

m.Unlock()

wg.Done()

}

func main() {

var m sync.RWMutex

var rs, ws int

rsCh := make(chan int)

wsCh := make(chan int)

go func() {

for {

select {

case n := <-rsCh:

rs += n

case n := <-wsCh:

ws += n

}

fmt.Printf("%s%s\n", strings.Repeat("R", rs),

strings.Repeat("W", ws))

}

}()

wg := sync.WaitGroup{}

for i := 0; i < 3; i++ {

wg.Add(1)

go reader(rsCh, &m, &wg)

}

for i := 0; i < 3; i++ {

wg.Add(1)

go writer(wsCh, &m, &wg)

}

wg.Wait()

}

実行すると以下のようになります。

W

R
RR
RRR
RR
R

W

W

RRR

sync.Map

Go の map は goroutine で Read しているときに別の goroutine から Write してはいけないというルールがあります。これを行なってしまうと panic が起きてしまいます。例えば以下の処理を実行すると異常終了を起こします。

package main

import (
	"fmt"
	"math/rand"
)

func main() {
	m := map[int]struct{}{}
	go func() {
		for {
			m[rand.Intn(100000)] = struct{}{}
		}
	}()

	for {
		fmt.Println(m[rand.Intn(100000)])
	}
}

package main

import (

"fmt"

"math/rand"

)

func main() {

m := map[int]struct{}{}

go func() {

for {

m[rand.Intn(100000)] = struct{}{}

}

}()

for {

fmt.Println(m[rand.Intn(100000)])

}

こちらに対応するには、上記の sync.Mutex を使うことで対応可能ですが、 sync.Map を使用することで、ロックの制御を気にせずに map を使用することができます。

package main

import (
	"fmt"
	"math/rand"
	"sync"
)

func main() {
	var m sync.Map

	go func() {
		for {
			m.Store(rand.Intn(100000), struct{}{})
		}
	}()

	for {
		fmt.Println(m.Load(rand.Intn(100000)))
	}
}

package main

import (

"fmt"

"math/rand"

"sync"

)

func main() {

var m sync.Map

go func() {

for {

m.Store(rand.Intn(100000), struct{}{})

}

}()

for {

fmt.Println(m.Load(rand.Intn(100000)))

}

sync.Once

sync.Once は関数を一度だけ実行するようにできます。

package main

import (
    "fmt"
    "runtime"
    "sync"
)

var once = new(sync.Once)

func greeting(wg *sync.WaitGroup) {
    once.Do(func() {
        fmt.Println("一度だけ")
    })

    fmt.Println("連続")
    wg.Done()
}

func main() {
    runtime.GOMAXPROCS(runtime.NumCPU())

    defer fmt.Println("最後")

    wg := new(sync.WaitGroup)
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go greeting(wg)
    }
    wg.Wait()
}

package main

import (

"fmt"

"runtime"

"sync"

)

var once = new(sync.Once)

func greeting(wg *sync.WaitGroup) {

once.Do(func() {

fmt.Println("一度だけ")

})

fmt.Println("連続")

wg.Done()

}

func main() {

runtime.GOMAXPROCS(runtime.NumCPU())

defer fmt.Println("最後")

wg := new(sync.WaitGroup)

for i := 0; i < 5; i++ {

wg.Add(1)

go greeting(wg)

}

wg.Wait()

}

実行すると以下のようになります。

一度だけ
連続
連続
連続
連続
連続
最後

一度だけ

連続

最後

sync.Pool

sync.Pool はスレッドセーフなメモリプールです。使い方は、 sync.Pool 構造体を生成する時の New フィールドにオブジェクト生成の関数を指定します。必要時に New の関数が呼ばれます。あるスレッドで、 Get メソッドを呼び、もしプールに無かったら New 関数でオブジェクトを生成します。そして Get でオブジェクトを取得します。もしプールにオブジェクトが在れば、そのオブジェクトを返します。使い終わったら Put メソッドでプールに戻します。
これの繰り返しでオブジェクトをリサイクルすることが出来ます。また、スレッドセーフですので、スレッド間でメモリプールする事ができます。

package main

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

type Hoge struct {
	name  string
	value int
}

func main() {

	pool := sync.Pool{
		New: func() interface{} {
			return &Hoge{
				name:  "hoge",
				value: 0,
			}
		},
	}
	wg := &sync.WaitGroup{}

	poolFunc := func() {
		hoge := pool.Get().(*Hoge)
		if hoge.value == 0 {
			fmt.Println("New")
		} else {
			fmt.Println("Cache")
		}
		hoge.value += 1
		pool.Put(hoge)
		time.Sleep(1 * time.Microsecond)
		wg.Done()
	}

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go poolFunc()
	}

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go poolFunc()
	}

	wg.Wait()
	fmt.Println("End")
}

package main

import (

"fmt"

"sync"

"time"

)

type Hoge struct {

name string

value int

}

func main() {

pool := sync.Pool{

New: func() interface{} {

return &Hoge{

name: "hoge",

value: 0,

}

wg := &sync.WaitGroup{}

poolFunc := func() {

hoge := pool.Get().(*Hoge)

if hoge.value == 0 {

fmt.Println("New")

} else {

fmt.Println("Cache")

}

hoge.value += 1

pool.Put(hoge)

time.Sleep(1 * time.Microsecond)

wg.Done()

}

for i := 0; i < 10; i++ {

wg.Add(1)

go poolFunc()

}

for i := 0; i < 10; i++ {

wg.Add(1)

go poolFunc()

}

wg.Wait()

fmt.Println("End")

}

実行すると以下のようになります。

New
New
New
Cache
New
New
New
Cache
New
New
New
New
New
New
New
New
New
New
New
Cache
End

New

Cache

New

Cache

New

Cache

End

さいごに

sync パッケージの紹介でしたがいかがでしょうか。一部機能にしか触れられておらず、これからも勉強していこうと思います。

日	月	火	水	木	金	土
			1	2	3	4
5	6	7	8	9	10	11
12	13	14	15	16	17	18
19	20	21	22	23	24	25
26	27	28	29	30

Golangのsyncパッケージによる同期・排他制御

はじめに

sync.WaitGroup

sync.Mutex

sync.RWMutex

sync.Map

sync.Once

sync.Pool

さいごに

おすすめ書籍

フォロー

アプリ情報