よっしー
こんにちは。よっしーです(^^)
本日は、Go言語を効果的に使うためのガイドラインについて解説しています。
スポンサーリンク
背景
Go言語を学び始めて、より良いコードを書きたいと思い、Go言語の公式ドキュメント「Effective Go」を知りました。これは、いわば「Goらしいコードの書き方指南書」になります。単に動くコードではなく、効率的で保守性の高いコードを書くためのベストプラクティスが詰まっているので、これを読んだ時の内容を備忘として残しました。
チャンネル
マップと同様に、チャンネルはmakeで割り当てられ、結果の値は基になるデータ構造への参照として機能します。オプションの整数パラメータが提供された場合、それはチャンネルのバッファサイズを設定します。デフォルトは0で、バッファなしまたは同期チャンネルになります。
ci := make(chan int) // 整数のバッファなしチャンネル
cj := make(chan int, 0) // 整数のバッファなしチャンネル
cs := make(chan *os.File, 100) // Fileへのポインタのバッファ付きチャンネル
バッファなしチャンネルは通信—値の交換—と同期—2つの計算（ゴルーチン）が既知の状態にあることを保証する—を組み合わせます。
チャンネルを使用した多くの素晴らしいイディオムがあります。始めるための1つを紹介します。前のセクションでは、バックグラウンドでソートを起動しました。チャンネルにより、起動するゴルーチンがソートの完了を待つことができます。
c := make(chan int) // チャンネルを割り当て。
// ゴルーチンでソートを開始；完了時にチャンネルでシグナル。
go func() {
list.Sort()
c <- 1 // シグナルを送信；値は重要ではない。
}()
doSomethingForAWhile()
<-c // ソートの完了を待機；送信された値は破棄。
レシーバーは受信するデータがあるまで常にブロックします。チャンネルがバッファなしの場合、送信者はレシーバーが値を受信するまでブロックします。チャンネルにバッファがある場合、送信者は値がバッファにコピーされるまでのみブロックします。バッファが満杯の場合、これは何らかのレシーバーが値を取得するまで待つことを意味します。
バッファ付きチャンネルは、例えばスループットを制限するために、セマフォのように使用できます。この例では、入ってくるリクエストがhandleに渡され、これがチャンネルに値を送信し、リクエストを処理し、次の消費者のために「セマフォ」を準備するためにチャンネルから値を受信します。チャンネルバッファの容量は、processへの同時呼び出し数を制限します。
var sem = make(chan int, MaxOutstanding)
func handle(r *Request) {
sem <- 1 // アクティブキューが空になるまで待機。
process(r) // 長時間かかる可能性がある。
<-sem // 完了；次のリクエストの実行を有効化。
}
func Serve(queue chan *Request) {
for {
req := <-queue
go handle(req) // handleの完了を待たない。
}
}
MaxOutstandingのハンドラーがprocessを実行していると、それ以上のハンドラーは満杯のチャンネルバッファに送信しようとしてブロックされ、既存のハンドラーの1つが完了してバッファから受信するまで待ちます。
しかし、この設計には問題があります：Serveは入ってくるリクエストごとに新しいゴルーチンを作成しますが、任意の時点で実行できるのはMaxOutstandingのみです。その結果、リクエストが速すぎると、プログラムは無制限のリソースを消費する可能性があります。ゴルーチンの作成をゲートするようにServeを変更することで、この欠陥に対処できます：
func Serve(queue chan *Request) {
for req := range queue {
sem <- 1
go func() {
process(req)
<-sem
}()
}
}
（Go 1.22より前のバージョンでは、このコードにはバグがあることに注意してください：ループ変数は全てのゴルーチンで共有されます。詳細についてはGo wikiを参照してください。）
リソースをうまく管理する別のアプローチは、固定数のhandleゴルーチンを開始し、すべてがリクエストチャンネルから読み取るようにすることです。ゴルーチンの数はprocessへの同時呼び出し数を制限します。このServe関数は、終了を告げられるチャンネルも受け入れます。ゴルーチンを起動した後、そのチャンネルからの受信をブロックします。
func handle(queue chan *Request) {
for r := range queue {
process(r)
}
}
func Serve(clientRequests chan *Request, quit chan bool) {
// ハンドラーを開始
for i := 0; i < MaxOutstanding; i++ {
go handle(clientRequests)
}
<-quit // 終了を告げられるまで待機。
}
チャンネルとは？
チャンネルは、ゴルーチン間でデータを安全に受け渡しするための仕組みです。「パイプ」のようなもので、一方から値を送信し、他方で受信できます。
チャンネルの基本的な作成と使用
package main
import (
"fmt"
"time"
)
func basicChannelExample() {
fmt.Println("=== 基本的なチャンネルの例 ===")
// バッファなしチャンネルの作成
ch := make(chan string)
// 送信者ゴルーチン
go func() {
time.Sleep(1 * time.Second)
ch <- "Hello, Channel!" // チャンネルに送信
fmt.Println("メッセージを送信しました")
}()
fmt.Println("メッセージを待機中...")
message := <-ch // チャンネルから受信
fmt.Println("受信したメッセージ:", message)
}
func main() {
basicChannelExample()
}
バッファなし vs バッファ付きチャンネル
package main
import (
"fmt"
"time"
)
func unbufferedChannelExample() {
fmt.Println("=== バッファなしチャンネル ===")
ch := make(chan int) // バッファなし
go func() {
fmt.Println("送信開始")
ch <- 42 // 受信者が現れるまでブロック
fmt.Println("送信完了")
}()
time.Sleep(2 * time.Second) // 意図的に待機
fmt.Println("受信開始")
value := <-ch
fmt.Printf("受信した値: %d\n", value)
}
func bufferedChannelExample() {
fmt.Println("\n=== バッファ付きチャンネル ===")
ch := make(chan int, 3) // バッファサイズ3
go func() {
for i := 1; i <= 5; i++ {
fmt.Printf("送信: %d\n", i)
ch <- i
if i <= 3 {
fmt.Printf(" -> バッファに格納（ブロックなし）\n")
} else {
fmt.Printf(" -> バッファ満杯のため待機\n")
}
}
close(ch)
}()
time.Sleep(2 * time.Second)
fmt.Println("受信開始:")
for value := range ch {
fmt.Printf("受信: %d\n", value)
time.Sleep(500 * time.Millisecond)
}
}
func main() {
unbufferedChannelExample()
bufferedChannelExample()
}
ソート完了待ちの例
package main
import (
"fmt"
"math/rand"
"sort"
"time"
)
func sortingExample() {
fmt.Println("=== ソート完了待ちの例 ===")
// ランダムなデータを生成
data := make([]int, 10)
for i := range data {
data[i] = rand.Intn(100)
}
fmt.Printf("ソート前: %v\n", data)
c := make(chan int) // 完了通知用チャンネル
// バックグラウンドでソートを実行
go func() {
fmt.Println("ソート開始...")
time.Sleep(2 * time.Second) // ソート処理をシミュレート
sort.Ints(data)
fmt.Println("ソート完了")
c <- 1 // 完了シグナル（値は重要でない）
}()
// その他の処理を実行
fmt.Println("他の処理を実行中...")
for i := 0; i < 3; i++ {
fmt.Printf("処理 %d...\n", i+1)
time.Sleep(500 * time.Millisecond)
}
// ソート完了を待機
fmt.Println("ソート完了を待機中...")
<-c // ブロックして待機
fmt.Printf("ソート後: %v\n", data)
}
func main() {
sortingExample()
}
セマフォパターン：リソース制限
package main
import (
"fmt"
"math/rand"
"sync"
"time"
)
const MaxOutstanding = 3
type Request struct {
ID int
Data string
}
// セマフォとして使用するチャンネル
var sem = make(chan int, MaxOutstanding)
func process(r *Request) {
// 重い処理をシミュレート
processingTime := time.Duration(rand.Intn(2000)+1000) * time.Millisecond
fmt.Printf("リクエスト %d 処理開始 (%v)\n", r.ID, processingTime)
time.Sleep(processingTime)
fmt.Printf("リクエスト %d 処理完了\n", r.ID)
}
func handle(r *Request) {
sem <- 1 // セマフォを取得（空きスロットを待機）
process(r) // 実際の処理
<-sem // セマフォを解放（次のリクエストを許可）
}
func serveV1(queue chan *Request) {
fmt.Println("サーバーV1開始（問題あり版）")
for {
req := <-queue
go handle(req) // 各リクエストに対して新しいゴルーチン
}
}
func semaphoreExample() {
fmt.Println("=== セマフォパターンの例 ===")
queue := make(chan *Request, 10)
// リクエストを生成
go func() {
for i := 1; i <= 8; i++ {
request := &Request{
ID: i,
Data: fmt.Sprintf("データ-%d", i),
}
queue <- request
fmt.Printf("リクエスト %d をキューに追加\n", i)
time.Sleep(200 * time.Millisecond)
}
}()
// サーバーを開始
go serveV1(queue)
// 実行時間を制限
time.Sleep(8 * time.Second)
fmt.Println("セマフォ例終了\n")
}
func main() {
semaphoreExample()
}
改良版：ゴルーチン制限付きサーバー
package main
import (
"fmt"
"math/rand"
"time"
)
const MaxOutstanding = 3
type Request struct {
ID int
Data string
}
func process(r *Request) {
processingTime := time.Duration(rand.Intn(1000)+500) * time.Millisecond
fmt.Printf(" -> リクエスト %d 処理中 (%v)\n", r.ID, processingTime)
time.Sleep(processingTime)
fmt.Printf(" <- リクエスト %d 完了\n", r.ID)
}
// 改良版1：ゴルーチン作成を制限
func serveV2(queue chan *Request) {
fmt.Println("サーバーV2開始（改良版1）")
sem := make(chan int, MaxOutstanding)
for req := range queue {
sem <- 1 // 空きスロットを待機
go func(r *Request) {
process(r)
<-sem // スロットを解放
}(req)
}
}
// 改良版2：固定数のワーカー
func handle(queue chan *Request, workerID int) {
fmt.Printf("ワーカー %d 開始\n", workerID)
for r := range queue {
fmt.Printf("ワーカー %d がリクエスト %d を処理\n", workerID, r.ID)
process(r)
}
fmt.Printf("ワーカー %d 終了\n", workerID)
}
func serveV3(clientRequests chan *Request, quit chan bool) {
fmt.Println("サーバーV3開始（改良版2 - 固定ワーカー）")
// 固定数のワーカーを開始
for i := 0; i < MaxOutstanding; i++ {
go handle(clientRequests, i+1)
}
<-quit // 終了シグナルを待機
fmt.Println("サーバーV3終了")
}
func improvedServerExample() {
fmt.Println("=== 改良版サーバーの例 ===")
// V2のテスト
fmt.Println("\n--- V2テスト ---")
queue2 := make(chan *Request, 5)
go func() {
for i := 1; i <= 6; i++ {
queue2 <- &Request{ID: i, Data: fmt.Sprintf("データV2-%d", i)}
time.Sleep(200 * time.Millisecond)
}
close(queue2)
}()
serveV2(queue2)
time.Sleep(3 * time.Second)
// V3のテスト
fmt.Println("\n--- V3テスト ---")
queue3 := make(chan *Request, 5)
quit := make(chan bool)
go func() {
for i := 1; i <= 8; i++ {
queue3 <- &Request{ID: i, Data: fmt.Sprintf("データV3-%d", i)}
time.Sleep(300 * time.Millisecond)
}
close(queue3)
// 少し待ってから終了シグナル
time.Sleep(2 * time.Second)
quit <- true
}()
serveV3(queue3, quit)
}
func main() {
improvedServerExample()
}
実用的な例：データパイプライン
package main
import (
"fmt"
"math/rand"
"sync"
"time"
)
// データパイプラインの例
func dataGenerator(output chan<- int, count int) {
defer close(output)
for i := 1; i <= count; i++ {
data := rand.Intn(100)
output <- data
fmt.Printf("生成: %d\n", data)
time.Sleep(100 * time.Millisecond)
}
fmt.Println("データ生成完了")
}
func dataProcessor(input <-chan int, output chan<- int, processorID int) {
defer close(output)
for data := range input {
// 処理時間をシミュレート
time.Sleep(200 * time.Millisecond)
processed := data * 2
output <- processed
fmt.Printf("プロセッサ %d: %d -> %d\n", processorID, data, processed)
}
fmt.Printf("プロセッサ %d 完了\n", processorID)
}
func dataCollector(input <-chan int, results *[]int, wg *sync.WaitGroup) {
defer wg.Done()
for processed := range input {
*results = append(*results, processed)
fmt.Printf("収集: %d\n", processed)
}
fmt.Println("データ収集完了")
}
func pipelineExample() {
fmt.Println("=== データパイプラインの例 ===")
// パイプライン用のチャンネル
rawData := make(chan int, 5)
processedData := make(chan int, 5)
var results []int
var wg sync.WaitGroup
// パイプラインの各段階を開始
go dataGenerator(rawData, 8)
go dataProcessor(rawData, processedData, 1)
wg.Add(1)
go dataCollector(processedData, &results, &wg)
// 完了を待機
wg.Wait()
fmt.Printf("最終結果: %v\n", results)
}
func main() {
pipelineExample()
}
複数チャンネルの操作：select文
package main
import (
"fmt"
"time"
)
func selectExample() {
fmt.Println("=== select文の例 ===")
ch1 := make(chan string)
ch2 := make(chan string)
timeout := make(chan bool)
// 複数のゴルーチンからデータを送信
go func() {
time.Sleep(1 * time.Second)
ch1 <- "チャンネル1からのメッセージ"
}()
go func() {
time.Sleep(2 * time.Second)
ch2 <- "チャンネル2からのメッセージ"
}()
go func() {
time.Sleep(3 * time.Second)
timeout <- true
}()
// selectで複数チャンネルを監視
for i := 0; i < 3; i++ {
select {
case msg1 := <-ch1:
fmt.Println("受信:", msg1)
case msg2 := <-ch2:
fmt.Println("受信:", msg2)
case <-timeout:
fmt.Println("タイムアウトしました")
return
case <-time.After(4 * time.Second):
fmt.Println("全体タイムアウト")
return
}
}
}
func main() {
selectExample()
}
重要なポイント
1. チャンネルの種類
- バッファなし: 同期的、送信者と受信者が同時に準備完了
- バッファ付き: 非同期的、バッファが満杯になるまで送信可能
2. 基本操作
ch := make(chan int) // バッファなしチャンネル
ch := make(chan int, 10) // バッファ付きチャンネル
ch <- value // 送信
value := <-ch // 受信
close(ch) // チャンネルを閉じる
3. ブロッキングの挙動
- 受信者: データがあるまでブロック
- 送信者（バッファなし）: 受信者が受信するまでブロック
- 送信者（バッファ付き）: バッファが満杯になるまでブロックしない
4. 実用的なパターン
- 完了通知: ゴルーチンの完了を待機
- セマフォ: リソースへの同時アクセス数を制限
- ワーカープール: 固定数のワーカーでタスク処理
- パイプライン: データの段階的処理
5. 設計上の利点
- 型安全: チャンネルは特定の型のデータのみ送信可能
- データ競合の回避: チャンネル経由でのみデータを共有
- 明確な同期: 通信が同期の役割も果たす
6. 注意点
- デッドロック: 送信者と受信者の組み合わせに注意
- チャンネルの閉じ忘れ: rangeループで使用する場合は必ずclose
- ゴルーチンリーク: 無限に待機するゴルーチンに注意
チャンネルは、Goの並行性プログラミングの核心であり、安全で効率的なゴルーチン間通信を可能にします。適切に使用することで、複雑な並行処理を簡潔に表現できます。
おわりに
本日は、Go言語を効果的に使うためのガイドラインについて解説しました。
よっしー
何か質問や相談があれば、コメントをお願いします。また、エンジニア案件の相談にも随時対応していますので、お気軽にお問い合わせください。
それでは、また明日お会いしましょう(^^)
コメント