Konkurensi

Dukungan Go untuk konkurensi adalah alami, ini adalah inti dari bahasa ini. Kesulitan penggunaannya relatif kecil, developer tidak perlu terlalu memperhatikan implementasi底层 untuk membuat aplikasi konkurensi yang cukup baik, meningkatkan batas bawah developer.

Goroutine

Goroutine adalah thread ringan, atau disebut thread user-state, tidak dijadwalkan langsung oleh sistem operasi, tetapi dijadwalkan oleh scheduler bahasa Go sendiri saat runtime. Oleh karena itu overhead pergantian context sangat kecil, ini juga merupakan salah satu alasan mengapa performa konkurensi Go cukup baik. Konsep goroutine bukan pertama kali diperkenalkan oleh Go, Go juga bukan bahasa pertama yang mendukung goroutine, tetapi Go adalah bahasa pertama yang dapat membuat goroutine dan dukungan konkurensi sangat sederhana dan elegan.

Di Go, membuat goroutine sangat mudah, hanya dengan satu keyword go, dapat dengan cepat memulai goroutine. Keyword go harus diikuti oleh pemanggilan fungsi. Contoh sebagai berikut

TIP

Fungsi built-in yang memiliki nilai return tidak diizinkan mengikuti keyword go, seperti contoh error berikut

go make([]int,10) //  go discards result of make([]int, 10) (value of type []int)

func main() {
  go fmt.Println("hello world!")
  go hello()
  go func() {
    fmt.Println("hello world!")
  }()
}

func hello() {
  fmt.Println("hello world!")
}

Ketiga cara memulai goroutine di atas semuanya bisa, tetapi sebenarnya setelah contoh ini dieksekusi, dalam sebagian besar kasus tidak ada output yang dihasilkan. Goroutine dieksekusi secara konkuren, sistem membutuhkan waktu untuk membuat goroutine. Sebelum itu, goroutine utama sudah selesai berjalan. Begitu thread utama keluar, goroutine anak lainnya juga keluar. Selain itu, urutan eksekusi goroutine juga tidak pasti, tidak dapat diprediksi. Contoh seperti contoh berikut

func main() {
  fmt.Println("start")
  for i := 0; i < 10; i++ {
    go fmt.Println(i)
  }
  fmt.Println("end")
}

Ini adalah contoh memulai goroutine di dalam loop, tidak akan pernah dapat memprediksi dengan tepat apa yang akan dioutputkan. Mungkin goroutine anak belum mulai berjalan, goroutine utama sudah selesai, situasinya sebagai berikut

start
end

Atau hanya sebagian goroutine anak yang berhasil berjalan sebelum goroutine utama keluar, situasinya sebagai berikut

start
0
1
5
3
4
6
7
end

Cara paling sederhana adalah membuat goroutine utama menunggu sebentar, perlu menggunakan fungsi Sleep di bawah paket time, dapat membuat goroutine saat ini berhenti sementara, contoh sebagai berikut

func main() {
  fmt.Println("start")
  for i := 0; i < 10; i++ {
    go fmt.Println(i)
  }
    // Pause 1ms
  time.Sleep(time.Millisecond)
  fmt.Println("end")
}

Eksekusi lagi output sebagai berikut, dapat dilihat semua angka dioutputkan dengan lengkap, tidak ada yang terlewat

start
0
1
5
2
3
4
6
8
9
7
end

Tetapi urutannya masih acak, oleh karena itu buat setiap loop menunggu sebentar. Contoh sebagai berikut

func main() {
   fmt.Println("start")
   for i := 0; i < 10; i++ {
      go fmt.Println(i)
      time.Sleep(time.Millisecond)
   }
   time.Sleep(time.Millisecond)
   fmt.Println("end")
}

Output sekarang sudah dalam urutan normal

start
0
1
2
3
4
5
6
7
8
9
end

Hasil output contoh di atas sangat sempurna, apakah masalah konkurensi sudah teratasi? Tidak, sama sekali tidak. Untuk program konkuren, faktor yang tidak dapat dikontrol sangat banyak, timing eksekusi, urutan, waktu eksekusi proses, dll. Jika tugas goroutine anak di dalam loop bukan hanya output angka sederhana, tetapi tugas yang sangat besar dan kompleks, waktu yang tidak pasti, maka masih akan mengulang masalah sebelumnya. Contoh seperti kode berikut

func main() {
   fmt.Println("start")
   for i := 0; i < 10; i++ {
      go hello(i)
      time.Sleep(time.Millisecond)
   }
   time.Sleep(time.Millisecond)
   fmt.Println("end")
}

func hello(i int) {
   // Simulasi waktu konsumsi acak
   time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
   fmt.Println(i)
}

Output kode ini masih tidak pasti, berikut adalah salah satu kemungkinan

start
0
3
4
end

Oleh karena itu time.Sleep bukan solusi yang baik. Untungnya Go menyediakan banyak metode kontrol konkurensi, ada tiga metode kontrol konkurensi yang umum digunakan:

• channel: Pipeline
• WaitGroup: Semaphore
• Context: Context

Tiga metode memiliki situasi aplikasi yang berbeda, WaitGroup dapat mengontrol secara dinamis sejumlah goroutine yang ditentukan, Context lebih cocok untuk situasi di mana level nested goroutine anak lebih dalam, pipeline lebih cocok untuk komunikasi antar goroutine. Untuk kontrol lock tradisional, Go juga menyediakan dukungan:

• Mutex: Mutual exclusion lock
• RWMutex: Read-write mutual exclusion lock

Channel

channel, diterjemahkan sebagai channel. Go menjelaskan fungsi channel sebagai berikut:

Do not communicate by sharing memory; instead, share memory by communicating.

Yaitu berbagi memori melalui pesan, channel lahir untuk ini, ini adalah solusi untuk komunikasi antar goroutine, sekaligus dapat digunakan untuk kontrol konkurensi. Pertama mari kenali sintaks dasar channel. Di Go menggunakan keyword chan untuk merepresentasikan tipe channel, sekaligus harus mendeklarasikan tipe penyimpanan channel, untuk menentukan data apa yang disimpannya. Contoh berikut adalah tampilan channel biasa.

var ch chan int

Ini adalah statement deklarasi channel, saat ini channel belum diinisialisasi, nilainya nil, tidak dapat langsung digunakan.

Membuat

Saat membuat channel, hanya ada satu metode, yaitu menggunakan fungsi built-in make. Untuk channel, fungsi make menerima dua parameter, pertama adalah tipe channel, kedua adalah parameter opsional untuk ukuran buffer channel. Contoh sebagai berikut

intCh := make(chan int)
// Channel dengan ukuran buffer 1
strCh := make(chan string, 1)

Setelah menggunakan channel, harus ingat untuk menutup channel tersebut, menggunakan fungsi built-in close untuk menutup channel. Signature fungsi ini sebagai berikut

func close(c chan<- Type)

Contoh menutup channel sebagai berikut

func main() {
  intCh := make(chan int)
  // do something
  close(intCh)
}

Kadang-kadang menggunakan defer untuk menutup channel mungkin lebih baik.

Baca Tulis

Untuk channel, Go menggunakan dua operator yang sangat visual untuk merepresentasikan operasi baca tulis:

ch <-: Menunjukkan menulis data ke channel

<- ch: Menunjukkan membaca data dari channel

<- sangat visual menunjukkan arah aliran data, lihat contoh operasi baca tulis channel tipe int

func main() {
    // Jika tidak ada buffer akan menyebabkan deadlock
  intCh := make(chan int, 1)
  defer close(intCh)
    // Menulis data
  intCh <- 114514
    // Membaca data
  fmt.Println(<-intCh)
}

Contoh di atas membuat channel tipe int dengan ukuran buffer 1, menulis data 114514 ke dalamnya, lalu membaca data dan mengoutputkannya, terakhir menutup channel. Untuk operasi baca, ada nilai return kedua, nilai tipe boolean, digunakan untuk menunjukkan apakah data berhasil dibaca

ints, ok := <-intCh

Cara aliran data di channel sama seperti queue, yaitu First In First Out (FIFO). Operasi goroutine terhadap channel adalah sinkron, pada某一 saat, hanya satu goroutine yang dapat menulis data ke channel, dan hanya satu goroutine yang dapat membaca data dari channel.

Tanpa Buffer

Untuk channel tanpa buffer, karena kapasitas buffer adalah 0, jadi tidak dapat menyimpan data sementara. Karena channel tanpa buffer tidak dapat menyimpan data, saat menulis data ke channel harus segera ada goroutine lain untuk membaca data, jika tidak akan阻塞 menunggu. Sama juga saat membaca data, ini juga menjelaskan mengapa kode yang terlihat sangat normal di bawah ini akan terjadi deadlock.

func main() {
  // Membuat channel tanpa buffer
  ch := make(chan int)
  defer close(ch)
  // Menulis data
  ch <- 123
  // Membaca data
  n := <-ch
  fmt.Println(n)
}

Channel tanpa buffer tidak boleh digunakan secara sinkron, seharusnya memulai goroutine baru untuk mengirim data, seperti contoh berikut

func main() {
  // Membuat channel tanpa buffer
  ch := make(chan int)
  defer close(ch)
  go func() {
    // Menulis data
    ch <- 123
  }()
  // Membaca data
  n := <-ch
  fmt.Println(n)
}

Dengan Buffer

Ketika channel memiliki buffer, seperti blocking queue, membaca channel kosong dan menulis channel yang sudah penuh akan menyebabkan阻塞. Channel tanpa buffer saat mengirim data, harus segera ada yang menerima, jika tidak akan terus阻塞. Untuk channel dengan buffer tidak perlu seperti ini, saat menulis ke channel dengan buffer, data akan dimasukkan ke buffer terlebih dahulu, hanya ketika kapasitas buffer penuh akan阻塞 menunggu goroutine untuk membaca data dari channel. Sama juga, saat membaca channel dengan buffer, akan membaca dari buffer terlebih dahulu, sampai buffer tidak ada data, baru akan阻塞 menunggu goroutine untuk menulis data ke channel. Oleh karena itu, contoh yang dapat menyebabkan deadlock di channel tanpa buffer di sini dapat berjalan dengan lancar.

func main() {
   // Membuat channel dengan buffer
   ch := make(chan int, 1)
   defer close(ch)
   // Menulis data
   ch <- 123
   // Membaca data
   n := <-ch
   fmt.Println(n)
}

Meskipun dapat berjalan dengan lancar, tetapi cara baca tulis sinkron ini sangat berbahaya. Begitu buffer channel kosong atau penuh, akan terus阻塞下去, karena tidak ada goroutine lain untuk menulis atau membaca data dari channel. Lihat contoh berikut

func main() {
  // Membuat channel dengan buffer
  ch := make(chan int, 5)
  // Membuat dua channel tanpa buffer
  chW := make(chan struct{})
  chR := make(chan struct{})
  defer func() {
    close(ch)
    close(chW)
    close(chR)
  }()
  // Bertanggung jawab menulis
  go func() {
    for i := 0; i < 10; i++ {
      ch <- i
      fmt.Println("menulis", i)
    }
    chW <- struct{}{}
  }()
  // Bertanggung jawab membaca
  go func() {
    for i := 0; i < 10; i++ {
            // Setiap kali membaca data membutuhkan 1 milidetik
      time.Sleep(time.Millisecond)
      fmt.Println("membaca", <-ch)
    }
    chR <- struct{}{}
  }()
  fmt.Println("menulis selesai", <-chW)
  fmt.Println("membaca selesai", <-chR)
}

Di sini total dibuat 3 channel, satu channel dengan buffer untuk komunikasi antar goroutine, dua channel tanpa buffer untuk mensinkronkan urutan eksekusi goroutine induk-anak. Goroutine yang bertanggung jawab membaca akan menunggu 1 milidetik sebelum setiap pembacaan, goroutine yang bertanggung jawab menulis paling banyak hanya dapat menulis 5 data sekaligus, karena buffer channel maksimal hanya 5. Sebelum ada goroutine untuk membaca, hanya dapat阻塞 menunggu. Oleh karena itu output contoh ini sebagai berikut

menulis 0
menulis 1
menulis 2
menulis 3
menulis 4 // Sekali tulis 5, buffer penuh, tunggu goroutine lain untuk membaca
membaca 0
menulis 5 // Baca satu, tulis satu
membaca 1
menulis 6
membaca 2
menulis 7
membaca 3
menulis 8
menulis 9
membaca 4
menulis selesai {} // Semua data terkirim, goroutine tulis selesai
membaca 5
membaca 6
membaca 7
membaca 8
membaca 9
membaca selesai {} // Semua data terbaca, goroutine baca selesai

Dapat dilihat goroutine tulis刚开始 langsung mengirim 5 data, setelah buffer penuh mulai阻塞 menunggu goroutine baca untuk membaca. Kemudian setiap goroutine baca membaca satu data setiap 1 milidetik, buffer ada ruang kosong, goroutine tulis menulis satu data, sampai semua data terkirim, goroutine tulis selesai eksekusi. Kemudian ketika goroutine baca membaca semua data di buffer, goroutine baca juga selesai eksekusi, terakhir goroutine utama keluar.

TIP

Melalui fungsi built-in len dapat mengakses jumlah data di buffer channel, melalui cap dapat mengakses ukuran buffer channel.

func main() {
   ch := make(chan int, 5)
   ch <- 1
   ch <- 2
   ch <- 3
   fmt.Println(len(ch), cap(ch))
}

Output

3 5

Memanfaatkan kondisi阻塞 channel, dapat dengan mudah menulis contoh goroutine utama menunggu goroutine anak selesai eksekusi

func main() {
   // Membuat channel tanpa buffer
   ch := make(chan struct{})
   defer close(ch)
   go func() {
      fmt.Println(2)
      // Menulis
      ch <- struct{}{}
   }()
   //阻塞 menunggu baca
   <-ch
   fmt.Println(1)
}

Output

2
1

Melalui channel dengan buffer juga dapat mengimplementasikan mutual exclusion lock sederhana, lihat contoh berikut

var count = 0

// Channel dengan ukuran buffer 1
var lock = make(chan struct{}, 1)

func Add() {
    // Lock
  lock <- struct{}{}
  fmt.Println("count saat ini", count, "eksekusi tambah")
  count += 1
    // Unlock
  <-lock
}

func Sub() {
    // Lock
  lock <- struct{}{}
  fmt.Println("count saat ini", count, "eksekusi kurang")
  count -= 1
    // Unlock
  <-lock
}

Karena ukuran buffer channel adalah 1, paling banyak hanya satu data yang tersimpan di buffer. Fungsi Add dan Sub akan mencoba mengirim data ke channel sebelum setiap operasi. Karena ukuran buffer adalah 1, jika goroutine lain sudah menulis data, buffer sudah penuh, goroutine saat ini harus阻塞 menunggu, sampai buffer kosong. Dengan demikian, pada某一 saat, paling banyak hanya satu goroutine yang dapat memodifikasi variabel count,这样就 mengimplementasikan mutual exclusion lock sederhana.

Poin Perhatian

Berikut adalah beberapa ringkasan, beberapa situasi berikut jika digunakan tidak tepat akan menyebabkan channel阻塞:

Baca tulis channel tanpa buffer

Saat melakukan operasi baca tulis sinkron langsung pada channel tanpa buffer akan menyebabkan goroutine saat ini阻塞

func main() {
   // Membuat channel tanpa buffer
   intCh := make(chan int)
   defer close(intCh)
   // Mengirim data
   intCh <- 1
   // Membaca data
   ints, ok := <-intCh
   fmt.Println(ints, ok)
}

Membaca channel dengan buffer kosong

Saat membaca channel dengan buffer kosong, akan menyebabkan goroutine saat ini阻塞

func main() {
   // Membuat channel dengan buffer
   intCh := make(chan int, 1)
   defer close(intCh)
   // Buffer kosong,阻塞 menunggu goroutine lain menulis data
   ints, ok := <-intCh
   fmt.Println(ints, ok)
}

Menulis channel dengan buffer penuh

Ketika buffer channel sudah penuh, menulis data akan menyebabkan goroutine saat ini阻塞

func main() {
  // Membuat channel dengan buffer
  intCh := make(chan int, 1)
  defer close(intCh)

  intCh <- 1
    // Penuh,阻塞 menunggu goroutine lain untuk membaca data
  intCh <- 1
}

Channel adalah nil

Ketika channel adalah nil, apapun operasi baca tulis akan menyebabkan goroutine saat ini阻塞

func main() {
  var intCh chan int
    // Menulis
  intCh <- 1
}

func main() {
  var intCh chan int
    // Membaca
  fmt.Println(<-intCh)
}

Kondisi阻塞 channel perlu dikuasai dan dikenali dengan baik, sebagian besar kasus masalah ini tersembunyi dengan sangat tersembunyi, tidak akan sejelas contoh.

Beberapa situasi berikut juga akan menyebabkan panic:

Menutup channel nil

Ketika channel adalah nil, menggunakan fungsi close untuk menutupnya akan menyebabkan panic

func main() {
  var intCh chan int
  close(intCh)
}

Menulis channel yang sudah ditutup

Menulis data ke channel yang sudah ditutup akan menyebabkan panic

func main() {
  intCh := make(chan int, 1)
  close(intCh)
  intCh <- 1
}

Menutup channel yang sudah ditutup

Dalam beberapa situasi, channel mungkin经过层层传递, caller mungkin juga tidak tahu siapa yang harus menutup channel, sehingga mungkin terjadi menutup channel yang sudah ditutup, akan terjadi panic.

func main() {
  ch := make(chan int, 1)
  defer close(ch)
  go write(ch)
  fmt.Println(<-ch)
}

func write(ch chan<- int) {
  // Hanya dapat mengirim data ke channel
  ch <- 1
  close(ch)
}

Channel Satu Arah

Channel dua arah adalah channel yang dapat ditulis dan dibaca, yaitu dapat dioperasikan di kedua sisi channel. Channel satu arah adalah channel yang hanya baca atau hanya tulis, yaitu hanya dapat dioperasikan di satu sisi channel. Membuat channel satu arah secara manual tidak terlalu berarti, karena tidak dapat membaca tulis channel maka kehilangan fungsi eksistensinya. Channel satu arah biasanya digunakan untuk membatasi perilaku channel, umumnya muncul di parameter fungsi dan nilai return, misalnya fungsi built-in close yang digunakan untuk menutup channel menggunakan channel satu arah di signature fungsi.

func close(c chan<- Type)

Atau fungsi After di bawah paket time yang umum digunakan

func After(d Duration) <-chan Time

Parameter fungsi close adalah channel hanya tulis, nilai return fungsi After adalah channel hanya baca, oleh karena itu sintaks channel satu arah sebagai berikut:

• Simbol panah <- di depan, adalah channel hanya baca, seperti <-chan int
• Simbol panah <- di belakang, adalah channel hanya tulis, seperti chan<- string

Saat mencoba menulis data ke channel hanya baca, tidak akan dapat dikompilasi

func main() {
  timeCh := time.After(time.Second)
  timeCh <- time.Now()
}

Error sebagai berikut, artinya sangat jelas

invalid operation: cannot send to receive-only channel timeCh (variable of type <-chan time.Time)

Sama juga untuk membaca data dari channel hanya tulis.

Channel dua arah dapat dikonversi menjadi channel satu arah, sebaliknya tidak dapat. Biasanya, jika meneruskan channel dua arah ke goroutine atau fungsi tertentu dan tidak ingin ia membaca/mengirim data, dapat menggunakan channel satu arah untuk membatasi perilaku pihak lain.

func main() {
   ch := make(chan int, 1)
   go write(ch)
   fmt.Println(<-ch)
}

func write(ch chan<- int) {
   // Hanya dapat mengirim data ke channel
   ch <- 1
}

Channel hanya baca juga sama prinsipnya

TIP

chan adalah tipe referensi, meskipun Go menggunakan pass-by-value untuk parameter fungsi, tetapi referensinya tetap sama, ini akan dijelaskan di prinsip channel selanjutnya.

for range

Melalui statement for range, dapat遍历 membaca data dari channel dengan buffer, seperti contoh

func main() {
  ch := make(chan int, 10)
  go func() {
    for i := 0; i < 10; i++ {
      ch <- i
    }
  }()
  for n := range ch {
    fmt.Println(n)
  }
}

Biasanya, for range saat遍历 struktur data iterable lainnya, akan memiliki dua nilai return, pertama adalah index, kedua adalah nilai elemen. Tetapi untuk channel, hanya ada satu nilai return, for range akan terus membaca elemen dari channel, ketika buffer channel kosong atau tanpa buffer, akan阻塞 menunggu, sampai ada goroutine lain menulis data ke channel baru akan terus membaca data. Oleh karena itu output sebagai berikut:

0
1
2
3
4
5
6
7
8
9
fatal error: all goroutines are asleep - deadlock!

Dapat dilihat kode di atas terjadi deadlock, karena goroutine anak sudah selesai eksekusi, sedangkan goroutine utama masih阻塞 menunggu goroutine lain untuk menulis data ke channel. Oleh karena itu channel harus ditutup setelah selesai menulis data. Ubah menjadi kode berikut

func main() {
   ch := make(chan int, 10)
   go func() {
      for i := 0; i < 10; i++ {
         ch <- i
      }
      // Menutup channel
      close(ch)
   }()
   for n := range ch {
      fmt.Println(n)
   }
}

Setelah menulis selesai menutup channel, kode di atas tidak akan terjadi deadlock lagi. Sebelumnya disebutkan membaca channel memiliki dua nilai return, saat for range遍历 channel, ketika tidak dapat berhasil membaca data, akan keluar dari loop. Nilai return kedua menunjukkan apakah dapat berhasil membaca data, bukan apakah channel sudah ditutup. Meskipun channel sudah ditutup, untuk channel dengan buffer, masih dapat membaca data, dan nilai return kedua masih true. Lihat contoh berikut

func main() {
  ch := make(chan int, 10)
  for i := 0; i < 5; i++ {
    ch <- i
  }
    // Menutup channel
  close(ch)
    // Membaca data lagi
  for i := 0; i < 6; i++ {
    n, ok := <-ch
    fmt.Println(n, ok)
  }
}

Hasil output

0 true
1 true
2 true
3 true
4 true
0 false

Karena channel sudah ditutup, meskipun buffer kosong, membaca data lagi tidak akan menyebabkan goroutine saat ini阻塞. Dapat dilihat pada iterasi keenam yang dibaca adalah nilai nol, dan ok adalah false.

TIP

Tentang timing menutup channel, sebaiknya menutup channel di pihak yang mengirim data ke channel, bukan di pihak penerima, karena sebagian besar kasus pihak penerima hanya tahu menerima data, tidak tahu kapan harus menutup channel.

WaitGroup

sync.WaitGroup adalah struct yang disediakan oleh paket sync, WaitGroup yaitu menunggu eksekusi, menggunakannya dapat dengan mudah mengimplementasikan efek menunggu sekelompok goroutine. Struct ini hanya mengekspos tiga metode.

Metode Add digunakan untuk menunjukkan jumlah goroutine yang akan ditunggu

func (wg *WaitGroup) Add(delta int)

Metode Done menunjukkan goroutine saat ini sudah selesai eksekusi

func (wg *WaitGroup) Done()

Metode Wait menunggu goroutine anak selesai, jika tidak akan阻塞

func (wg *WaitGroup) Wait()

WaitGroup sangat mudah digunakan,属于开箱即用. Implementasi internalnya adalah counter + semaphore, saat program dimulai memanggil Add untuk menginisialisasi counter, setiap kali goroutine selesai eksekusi memanggil Done, counter berkurang 1, sampai berkurang menjadi 0. Selama periode ini, goroutine utama memanggil Wait akan terus阻塞 sampai semua counter berkurang menjadi 0, lalu akan dibangunkan. Lihat contoh penggunaan sederhana

func main() {
  var wait sync.WaitGroup
  // Menentukan jumlah goroutine anak
  wait.Add(1)
  go func() {
    fmt.Println(1)
    // Selesai eksekusi
    wait.Done()
  }()
  // Menunggu goroutine anak
  wait.Wait()
  fmt.Println(2)
}

Kode ini selalu output 1 dulu lalu 2, goroutine utama akan menunggu goroutine anak selesai eksekusi lalu keluar.

1
2

Untuk contoh awal di pengenalan goroutine, dapat melakukan modifikasi sebagai berikut

func main() {
   var mainWait sync.WaitGroup
   var wait sync.WaitGroup
   // Counter 10
   mainWait.Add(10)
   fmt.Println("start")
   for i := 0; i < 10; i++ {
      // Di dalam loop counter 1
      wait.Add(1)
      go func() {
         fmt.Println(i)
         // Dua counter berkurang 1
         wait.Done()
         mainWait.Done()
      }()
      // Menunggu goroutine loop saat ini selesai eksekusi
      wait.Wait()
   }
   // Menunggu semua goroutine selesai eksekusi
   mainWait.Wait()
   fmt.Println("end")
}

Di sini menggunakan sync.WaitGroup menggantikan time.Sleep sebelumnya, urutan eksekusi konkuren goroutine lebih dapat dikontrol, tidak peduli berapa kali eksekusi, output selalu sebagai berikut

start
0
1
2
3
4
5
6
7
8
9
end

WaitGroup biasanya cocok untuk saat menyesuaikan jumlah goroutine secara dinamis, misalnya mengetahui jumlah goroutine sebelumnya, atau perlu menyesuaikan secara dinamis selama runtime. Nilai WaitGroup tidak boleh dicopy, nilai setelah dicopy juga tidak boleh terus digunakan, terutama saat meneruskannya sebagai parameter fungsi, harus meneruskan pointer bukan nilai. Jika menggunakan nilai copy, counter sama sekali tidak dapat bekerja pada WaitGroup yang sebenarnya, ini dapat menyebabkan goroutine utama terus阻塞 menunggu, program tidak akan dapat berjalan normal. Contoh seperti kode berikut

func main() {
  var mainWait sync.WaitGroup
  mainWait.Add(1)
  hello(mainWait)
  mainWait.Wait()
  fmt.Println("end")
}
func hello(wait sync.WaitGroup) {
  fmt.Println("hello")
  wait.Done()
}

Error提示 semua goroutine sudah keluar, tetapi goroutine utama masih menunggu,这样就 membentuk deadlock, karena panggilan Done di dalam fungsi hello pada parameter WaitGroup tidak akan bekerja pada mainWait yang asli. Oleh karena itu harus menggunakan pointer untuk meneruskan.

hello
fatal error: all goroutines are asleep - deadlock!

TIP

Ketika counter menjadi negatif, atau jumlah counter lebih besar dari jumlah goroutine anak, akan memicu panic.

Context

Context diterjemahkan sebagai context, adalah solusi kontrol konkurensi yang disediakan oleh Go. Dibandingkan channel dan WaitGroup, ia dapat mengontrol goroutine anak dan cucu dengan lebih baik serta goroutine dengan level yang lebih dalam. Context sendiri adalah interface, selama mengimplementasikan interface ini dapat disebut context, misalnya gin.Context di framework Web terkenal Gin. Paket standar context juga menyediakan beberapa implementasi, masing-masing adalah:

• emptyCtx
• cancelCtx
• timerCtx
• valueCtx

Context

Pertama lihat definisi interface Context, lalu pahami implementasi spesifiknya.

type Context interface {

   Deadline() (deadline time.Time, ok bool)

   Done() <-chan struct{}

   Err() error

   Value(key any) any
}

Deadline

Metode ini memiliki dua nilai return, deadline adalah waktu batas, yaitu waktu context harus dibatalkan. Nilai kedua adalah apakah deadline diset, jika tidak diset selalu false.

Deadline() (deadline time.Time, ok bool)

Done

Nilai returnnya adalah channel hanya baca tipe struct kosong, channel ini hanya berfungsi sebagai notifikasi, tidak mentransmisikan data apapun. Ketika pekerjaan yang dilakukan context harus dibatalkan, channel ini akan ditutup. Untuk beberapa context yang tidak mendukung pembatalan, mungkin mengembalikan nil.

Done() <-chan struct{}

Err

Metode ini mengembalikan error, digunakan untuk menunjukkan alasan penutupan context. Ketika channel Done belum ditutup, mengembalikan nil. Jika setelah ditutup, akan mengembalikan err untuk menjelaskan mengapa ditutup.

Err() error

Value

Metode ini mengembalikan nilai yang sesuai dengan key, jika key tidak ada, atau tidak mendukung metode ini, akan mengembalikan nil.

Value(key any) any

emptyCtx

Sesuai namanya, emptyCtx adalah context kosong, semua implementasi di paket context tidak diekspos ke luar, tetapi menyediakan fungsi yang sesuai untuk membuat context. emptyCtx dapat dibuat melalui context.Background dan context.TODO. Dua fungsi sebagai berikut

var (
  background = new(emptyCtx)
  todo       = new(emptyCtx)
)

func Background() Context {
  return background
}

func TODO() Context {
  return todo
}

Dapat dilihat hanya mengembalikan pointer emptyCtx. Tipe底层 emptyCtx sebenarnya adalah int, alasan tidak menggunakan struct kosong adalah karena instance emptyCtx harus memiliki alamat memori yang berbeda, ia tidak dapat dibatalkan, tidak ada deadline, juga tidak dapat mengambil nilai, metode yang diimplementasikan semuanya mengembalikan nilai nol.

type emptyCtx int

func (*emptyCtx) Deadline() (deadline time.Time, ok bool) {
   return
}

func (*emptyCtx) Done() <-chan struct{} {
   return nil
}

func (*emptyCtx) Err() error {
   return nil
}

func (*emptyCtx) Value(key any) any {
   return nil
}

emptyCtx biasanya digunakan sebagai context level teratas, diteruskan sebagai parent context saat membuat tiga context lainnya. Hubungan implementasi di paket context seperti gambar berikut

valueCtx

Implementasi valueCtx relatif sederhana, di dalamnya hanya berisi sepasang key-value, dan field tipe Context yang embedded.

type valueCtx struct {
   Context
   key, val any
}

Ia sendiri hanya mengimplementasikan metode Value, logikanya juga sangat sederhana, jika tidak ditemukan di context saat ini maka cari di parent context.

func (c *valueCtx) Value(key any) any {
   if c.key == key {
      return c.val
   }
   return value(c.Context, key)
}

Berikut lihat contoh penggunaan sederhana valueCtx

var waitGroup sync.WaitGroup

func main() {
  waitGroup.Add(1)
    // Meneruskan context
  go Do(context.WithValue(context.Background(), 1, 2))
  waitGroup.Wait()
}

func Do(ctx context.Context) {
    // Membuat timer baru
  ticker := time.NewTimer(time.Second)
  defer waitGroup.Done()
  for {
    select {
    case <-ctx.Done(): // Tidak akan pernah dieksekusi
    case <-ticker.C:
      fmt.Println("timeout")
      return
    default:
      fmt.Println(ctx.Value(1))
    }
    time.Sleep(time.Millisecond * 100)
  }
}

valueCtx banyak digunakan untuk mentransmisikan data di multi-level goroutine, tidak dapat dibatalkan, oleh karena itu ctx.Done selalu mengembalikan nil, select akan mengabaikan channel nil. Output terakhir sebagai berikut

2
2
2
2
2
2
2
2
2
2
timeout

cancelCtx

cancelCtx dan timerCtx keduanya mengimplementasikan interface canceler, tipe interface sebagai berikut

type canceler interface {
    // removeFromParent menunjukkan apakah menghapus diri dari parent context
    // err menunjukkan alasan pembatalan
  cancel(removeFromParent bool, err error)
    // Done mengembalikan channel, digunakan untuk memberi tahu alasan pembatalan
  Done() <-chan struct{}
}

Metode cancel tidak diekspos ke luar, saat membuat context dibungkus melalui closure sebagai nilai return untuk dipanggil pihak luar, seperti yang ditunjukkan di source code context.WithCancel

func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
   if parent == nil {
      panic("cannot create context from nil parent")
   }
   c := newCancelCtx(parent)
   // Mencoba menambahkan diri ke children parent
   propagateCancel(parent, &c)
   // Mengembalikan context dan fungsi
   return &c, func() { c.cancel(true, Canceled) }
}

cancelCtx diterjemahkan sebagai context yang dapat dibatalkan, saat membuat, jika parent mengimplementasikan canceler, akan menambahkan diri ke children parent, jika tidak akan terus mencari ke atas. Jika semua parent tidak mengimplementasikan canceler, akan memulai goroutine untuk menunggu parent dibatalkan, lalu ketika parent berakhir akan membatalkan context saat ini. Ketika memanggil cancelFunc, channel Done akan ditutup, setiap anak context ini juga akan ikut dibatalkan, terakhir akan menghapus diri dari parent. Berikut adalah contoh sederhana:

var waitGroup sync.WaitGroup

func main() {
  bkg := context.Background()
    // Mengembalikan cancelCtx dan fungsi cancel
  cancelCtx, cancel := context.WithCancel(bkg)
  waitGroup.Add(1)
  go func(ctx context.Context) {
    defer waitGroup.Done()
    for {
      select {
      case <-ctx.Done():
        fmt.Println(ctx.Err())
        return
      default:
        fmt.Println("menunggu pembatalan...")
      }
      time.Sleep(time.Millisecond * 200)
    }

  }(cancelCtx)
  time.Sleep(time.Second)
  cancel()
  waitGroup.Wait()
}

Output sebagai berikut

menunggu pembatalan...
menunggu pembatalan...
menunggu pembatalan...
menunggu pembatalan...
menunggu pembatalan...
context canceled

Lagi contoh dengan nested level yang lebih dalam

var waitGroup sync.WaitGroup

func main() {
   waitGroup.Add(3)
   ctx, cancelFunc := context.WithCancel(context.Background())
   go HttpHandler(ctx)
   time.Sleep(time.Second)
   cancelFunc()
   waitGroup.Wait()
}

func HttpHandler(ctx context.Context) {
   cancelCtxAuth, cancelAuth := context.WithCancel(ctx)
   cancelCtxMail, cancelMail := context.WithCancel(ctx)

   defer cancelAuth()
   defer cancelMail()
   defer waitGroup.Done()

   go AuthService(cancelCtxAuth)
   go MailService(cancelCtxMail)

   for {
      select {
      case <-ctx.Done():
         fmt.Println(ctx.Err())
         return
      default:
         fmt.Println("sedang memproses request http...")
      }
      time.Sleep(time.Millisecond * 200)
   }

}

func AuthService(ctx context.Context) {
   defer waitGroup.Done()
   for {
      select {
      case <-ctx.Done():
         fmt.Println("auth parent dibatalkan", ctx.Err())
         return
      default:
         fmt.Println("auth...")
      }
      time.Sleep(time.Millisecond * 200)
   }
}

func MailService(ctx context.Context) {
   defer waitGroup.Done()
   for {
      select {
      case <-ctx.Done():
         fmt.Println("mail parent dibatalkan", ctx.Err())
         return
      default:
         fmt.Println("mail...")
      }
      time.Sleep(time.Millisecond * 200)
   }
}

Contoh ini membuat 3 cancelCtx, meskipun parent cancelCtx saat dibatalkan akan membatalkan child context-nya, tetapi untuk jaga-jaga, jika membuat cancelCtx, setelah proses yang sesuai selesai harus memanggil fungsi cancel. Output sebagai berikut

sedang memproses request http...
auth...
mail...
mail...
auth...
sedang memproses request http...
auth...
mail...
sedang memproses request http...
sedang memproses request http...
auth...
mail...
auth...
sedang memproses request http...
mail...
context canceled
auth parent dibatalkan context canceled
mail parent dibatalkan context canceled

timerCtx

timerCtx di atas dasar cancelCtx menambahkan mekanisme timeout, paket context menyediakan dua fungsi untuk membuat, masing-masing adalah WithDeadline dan WithTimeout, keduanya fungsinya mirip, yang pertama menentukan waktu timeout spesifik, misalnya menentukan waktu spesifik 2023/3/20 16:32:00, yang kedua menentukan interval waktu timeout, misalnya 5 menit kemudian. Signature dua fungsi sebagai berikut

func WithDeadline(parent Context, d time.Time) (Context, CancelFunc)

func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc)

timerCtx akan otomatis membatalkan context saat ini setelah waktu kadaluarsa, proses pembatalan selain harus menutup timer tambahan, pada dasarnya sama dengan cancelCtx. Berikut adalah contoh penggunaan sederhana timerCtx

var wait sync.WaitGroup

func main() {
  deadline, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second))
  defer cancel()
  wait.Add(1)
  go func(ctx context.Context) {
    defer wait.Done()
    for {
      select {
      case <-ctx.Done():
        fmt.Println("context dibatalkan", ctx.Err())
        return
      default:
        fmt.Println("menunggu pembatalan...")
      }
      time.Sleep(time.Millisecond * 200)
    }
  }(deadline)
  wait.Wait()
}

Meskipun context akan otomatis dibatalkan setelah kadaluarsa, tetapi untuk jaga-jaga, setelah proses terkait selesai, sebaiknya membatalkan context secara manual. Output sebagai berikut

menunggu pembatalan...
menunggu pembatalan...
menunggu pembatalan...
menunggu pembatalan...
menunggu pembatalan...
context dibatalkan context deadline exceeded

WithTimeout sebenarnya sangat mirip dengan WithDeadline, implementasinya hanya sedikit enkapsulasi dan memanggil WithDeadline, penggunaan sama seperti contoh WithDeadline di atas, sebagai berikut

func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) {
   return WithDeadline(parent, time.Now().Add(timeout))
}

TIP

Sama seperti alokasi memori tanpa recovery akan menyebabkan memory leak, context juga merupakan resource, jika dibuat tetapi tidak pernah dibatalkan, juga akan menyebabkan context leak, oleh karena itu sebaiknya menghindari situasi ini.

Select

select di sistem Linux, adalah solusi IO multiplexing, sama juga, di Go select adalah struktur kontrol multiplexing channel. Apa itu multiplexing, sederhana用一句话概括: pada某一 saat, secara bersamaan memonitor apakah beberapa elemen tersedia, elemen yang dipantau dapat berupa request jaringan, file IO, dll. Elemen yang dipantau select di Go adalah channel, dan hanya channel. Sintaks select mirip dengan statement switch, berikut lihat seperti apa statement select

func main() {
  // Membuat tiga channel
  chA := make(chan int)
  chB := make(chan int)
  chC := make(chan int)
  defer func() {
    close(chA)
    close(chB)
    close(chC)
  }()
  select {
  case n, ok := <-chA:
    fmt.Println(n, ok)
  case n, ok := <-chB:
    fmt.Println(n, ok)
  case n, ok := <-chC:
    fmt.Println(n, ok)
  default:
    fmt.Println("semua channel tidak tersedia")
  }
}

Penggunaan

Mirip dengan switch, select terdiri dari beberapa case dan satu default, branch default dapat dihilangkan. Setiap case hanya dapat mengoperasikan satu channel, dan hanya dapat melakukan satu operasi, baik baca atau tulis. Ketika ada beberapa case tersedia, select akan secara pseudo-acak memilih satu case untuk dieksekusi. Jika semua case tidak tersedia, akan mengeksekusi branch default, jika tidak ada branch default, akan阻塞 menunggu, sampai setidaknya ada satu case tersedia. Karena contoh di atas tidak menulis data ke channel, tentu semua case tidak tersedia, oleh karena itu output akhir adalah hasil eksekusi branch default. Sedikit modifikasi sebagai berikut:

func main() {
   chA := make(chan int)
   chB := make(chan int)
   chC := make(chan int)
   defer func() {
      close(chA)
      close(chB)
      close(chC)
   }()
   // Memulai goroutine baru
   go func() {
      // Menulis data ke channel A
      chA <- 1
   }()
   select {
   case n, ok := <-chA:
      fmt.Println(n, ok)
   case n, ok := <-chB:
      fmt.Println(n, ok)
   case n, ok := <-chC:
      fmt.Println(n, ok)
   }
}

Contoh di atas memulai goroutine baru untuk menulis data ke channel A, select karena tidak ada branch default, jadi akan terus阻塞 menunggu sampai ada case tersedia. Ketika channel A tersedia, setelah mengeksekusi branch yang sesuai goroutine utama langsung keluar. Jika ingin terus memonitor channel, dapat digunakan bersama loop for, sebagai berikut.

func main() {
  chA := make(chan int)
  chB := make(chan int)
  chC := make(chan int)
  defer func() {
    close(chA)
    close(chB)
    close(chC)
  }()
  go Send(chA)
  go Send(chB)
  go Send(chC)
  // loop for
  for {
    select {
    case n, ok := <-chA:
      fmt.Println("A", n, ok)
    case n, ok := <-chB:
      fmt.Println("B", n, ok)
    case n, ok := <-chC:
      fmt.Println("C", n, ok)
    }
  }
}

func Send(ch chan<- int) {
  for i := 0; i < 3; i++ {
    time.Sleep(time.Millisecond)
    ch <- i
  }
}

Begini memang ketiga channel dapat digunakan, tetapi infinite loop + select akan menyebabkan goroutine utama阻塞 permanen, oleh karena itu dapat ditaruh di goroutine baru, dan menambahkan beberapa logika lainnya.

func main() {
  chA := make(chan int)
  chB := make(chan int)
  chC := make(chan int)
  defer func() {
    close(chA)
    close(chB)
    close(chC)
  }()

  l := make(chan struct{})

  go Send(chA)
  go Send(chB)
  go Send(chC)

  go func() {
  Loop:
    for {
      select {
      case n, ok := <-chA:
        fmt.Println("A", n, ok)
      case n, ok := <-chB:
        fmt.Println("B", n, ok)
      case n, ok := <-chC:
        fmt.Println("C", n, ok)
      case <-time.After(time.Second): // Menetap timeout 1 detik
        break Loop // Keluar dari loop
      }
    }
    l <- struct{}{} // Memberitahu goroutine utama dapat keluar
  }()

  <-l
}

func Send(ch chan<- int) {
  for i := 0; i < 3; i++ {
    time.Sleep(time.Millisecond)
    ch <- i
  }
}

Contoh di atas melalui loop for配合 select untuk terus memonitor apakah ketiga channel tersedia, dan case keempat adalah channel timeout, setelah timeout akan keluar dari loop, mengakhiri goroutine anak. Output akhir sebagai berikut

C 0 true
A 0 true
B 0 true
A 1 true
B 1 true
C 1 true
B 2 true
C 2 true
A 2 true

Timeout

Contoh sebelumnya menggunakan fungsi time.After, nilai returnnya adalah channel hanya baca, fungsi ini配合 select dapat dengan sangat sederhana mengimplementasikan mekanisme timeout, contoh sebagai berikut

func main() {
  chA := make(chan int)
  defer close(chA)
  go func() {
    time.Sleep(time.Second * 2)
    chA <- 1
  }()
  select {
  case n := <-chA:
    fmt.Println(n)
  case <-time.After(time.Second):
    fmt.Println("timeout")
  }
}

阻塞 Permanen

Ketika statement select tidak ada apapun, akan阻塞 permanen, misalnya

func main() {
  fmt.Println("start")
  select {}
  fmt.Println("end")
}

end tidak akan pernah dioutputkan, goroutine utama akan terus阻塞, situasi ini biasanya memiliki tujuan khusus.

TIP

Jika mengoperasikan channel dengan nilai nil di case select, tidak akan menyebabkan阻塞, case tersebut akan diabaikan, tidak akan pernah dieksekusi. Contoh kode berikut tidak peduli berapa kali eksekusi hanya akan output timeout.

func main() {
   var nilCh chan int
   select {
   case <-nilCh:
      fmt.Println("read")
   case nilCh <- 1:
      fmt.Println("write")
   case <-time.After(time.Second):
      fmt.Println("timeout")
   }
}

Non-blocking

Melalui branch default select配合 channel, kita dapat mengimplementasikan operasi kirim terima non-blocking, seperti ditunjukkan berikut

func TrySend(ch chan int, ele int) bool  {
	select {
	case ch <- ele:
		return true
	default:
		return false
	}
}

func TryRecv(ch chan int) (int, bool)  {
	select {
	case ele, ok := <-ch:
		return ele, ok
	default:
		return 0, false
	}
}

Sama juga, dapat mengimplementasikan判断 non-blocking apakah context sudah berakhir

func IsDone(ctx context.Context) bool {
	select {
	case <-ctx.Done():
		return true
	default:
		return false
	}
}

Lock

Pertama lihat contoh berikut

var wait sync.WaitGroup
var count = 0

func main() {
   wait.Add(10)
   for i := 0; i < 10; i++ {
      go func(data *int) {
         // Simulasi waktu akses
         time.Sleep(time.Millisecond * time.Duration(rand.Intn(5000)))
         // Mengakses data
         temp := *data
         // Simulasi waktu komputasi
         time.Sleep(time.Millisecond * time.Duration(rand.Intn(5000)))
         ans := 1
         // Memodifikasi data
         *data = temp + ans
         fmt.Println(*data)
         wait.Done()
      }(&count)
   }
   wait.Wait()
   fmt.Println("hasil akhir", count)
}

Untuk contoh di atas, memulai sepuluh goroutine untuk melakukan operasi +1 pada count, dan menggunakan time.Sleep untuk mensimulasikan waktu konsumsi yang berbeda. Menurut intuisi, 10 goroutine mengeksekusi 10 operasi +1, hasil akhir pasti 10, hasil yang benar memang 10, tetapi faktanya tidak demikian, hasil eksekusi contoh di atas sebagai berikut:

1
2
3
3
2
2
3
3
3
4
hasil akhir 4

Dapat dilihat hasil akhir adalah 4, dan ini hanya salah satu dari banyak kemungkinan hasil. Karena waktu akses dan komputasi setiap goroutine berbeda, goroutine A mengakses data menghabiskan 500 milidetik, saat ini nilai count yang diakses adalah 1, kemudian menghabiskan 400 milidetik untuk komputasi, tetapi dalam 400 milidetik ini, goroutine B sudah selesai akses dan komputasi dan berhasil memperbarui nilai count. Goroutine A setelah selesai komputasi, nilai yang diakses awal goroutine A sudah kadaluarsa, tetapi goroutine A tidak tahu hal ini, masih menambahkan satu pada nilai yang diakses awal, dan memberikan nilai ke count. Dengan demikian, hasil eksekusi goroutine B tertimpa. Beberapa goroutine membaca dan mengakses data bersama seperti ini, terutama akan terjadi masalah seperti ini, oleh karena itu perlu menggunakan lock.

Mutex dan RWMutex di bawah paket sync di Go menyediakan dua implementasi mutual exclusion lock dan read-write lock, dan menyediakan API yang sangat sederhana dan mudah digunakan, lock hanya perlu Lock(), unlock juga hanya perlu Unlock(). Perlu diperhatikan, lock yang disediakan Go adalah non-recursive lock, yaitu不可重入 lock, oleh karena itu lock ulang atau unlock ulang akan menyebabkan fatal. Signifikansi lock adalah melindungi invariant, lock adalah berharap data tidak akan dimodifikasi oleh goroutine lain, sebagai berikut

func DoSomething() {
  Lock()
    // Dalam proses ini, data tidak akan dimodifikasi oleh goroutine lain
  Unlock()
}

Jika recursive lock, mungkin akan terjadi situasi berikut

func DoSomething() {
  Lock()
    DoOther()
  Unlock()
}

func DoOther() {
  Lock()
  // do other
  Unlock()
}

Fungsi DoSomething jelas tidak tahu apa yang mungkin dilakukan fungsi DoOther pada data, sehingga memodifikasi data, misalnya membuka beberapa goroutine anak merusak invariant. Ini tidak可行 di Go, setelah lock harus menjamin invarian invariant, saat ini lock ulang unlock ulang akan menyebabkan deadlock. Oleh karena itu saat menulis kode harus menghindari situasi di atas,必要时 saat lock segera menggunakan statement defer untuk unlock.

Mutual Exclusion Lock

sync.Mutex adalah implementasi mutual exclusion lock yang disediakan Go, ia mengimplementasikan interface sync.Locker

type Locker interface {
   // Lock
   Lock()
   // Unlock
   Unlock()
}

Menggunakan mutual exclusion lock dapat dengan sangat sempurna menyelesaikan masalah di atas, contoh sebagai berikut

var wait sync.WaitGroup
var count = 0

var lock sync.Mutex

func main() {
  wait.Add(10)
  for i := 0; i < 10; i++ {
    go func(data *int) {
      // Lock
      lock.Lock()
      // Simulasi waktu akses
      time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
      // Mengakses data
      temp := *data
      // Simulasi waktu komputasi
      time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
      ans := 1
      // Memodifikasi data
      *data = temp + ans
      // Unlock
      lock.Unlock()
      fmt.Println(*data)
      wait.Done()
    }(&count)
  }
  wait.Wait()
  fmt.Println("hasil akhir", count)
}

Setiap goroutine sebelum mengakses data, pertama lock, setelah update selesai unlock, goroutine lain ingin mengakses harus pertama获得 lock, jika tidak akan阻塞 menunggu. Dengan demikian, tidak ada masalah di atas, oleh karena itu output sebagai berikut

1
2
3
4
5
6
7
8
9
10
hasil akhir 10

Read-Write Lock

Mutual exclusion lock cocok untuk situasi di mana frekuensi operasi baca dan tulis差不多, untuk beberapa data yang lebih banyak baca daripada tulis, jika menggunakan mutual exclusion lock, akan menyebabkan banyak kompetisi lock goroutine yang tidak perlu, ini akan mengonsumsi banyak resource sistem, saat ini perlu menggunakan read-write lock, yaitu read-write mutual exclusion lock. Untuk satu goroutine:

• Jika获得 read lock, goroutine lain melakukan operasi tulis akan阻塞, goroutine lain melakukan operasi baca tidak akan阻塞
• Jika获得 write lock, goroutine lain melakukan operasi tulis akan阻塞, goroutine lain melakukan operasi baca akan阻塞

Implementasi read-write mutual exclusion lock di Go adalah sync.RWMutex, ia juga mengimplementasikan interface Locker, tetapi ia menyediakan lebih banyak metode yang tersedia, sebagai berikut:

// Lock baca
func (rw *RWMutex) RLock()

// Mencoba lock baca
func (rw *RWMutex) TryRLock() bool

// Unlock baca
func (rw *RWMutex) RUnlock()

// Lock tulis
func (rw *RWMutex) Lock()

// Mencoba lock tulis
func (rw *RWMutex) TryLock() bool

// Unlock tulis
func (rw *RWMutex) Unlock()

Dua operasi mencoba lock TryRlock dan TryLock adalah non-blocking, berhasil lock akan mengembalikan true, saat tidak dapat获得 lock tidak akan阻塞 tetapi mengembalikan false. Implementasi internal read-write mutual exclusion lock tetap mutual exclusion lock, bukan berarti membagi read lock dan write lock就有 dua lock, dari awal sampai akhir hanya ada satu lock. Berikut lihat contoh penggunaan read-write mutual exclusion lock

var wait sync.WaitGroup
var count = 0

var rw sync.RWMutex

func main() {
  wait.Add(12)
  // Lebih banyak baca daripada tulis
  go func() {
    for i := 0; i < 3; i++ {
      go Write(&count)
    }
    wait.Done()
  }()
  go func() {
    for i := 0; i < 7; i++ {
      go Read(&count)
    }
    wait.Done()
  }()
  // Menunggu goroutine anak selesai
  wait.Wait()
  fmt.Println("hasil akhir", count)
}

func Read(i *int) {
  time.Sleep(time.Millisecond * time.Duration(rand.Intn(500)))
  rw.RLock()
  fmt.Println("dapat read lock")
  time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
  fmt.Println("melepaskan read lock", *i)
  rw.RUnlock()
  wait.Done()
}

func Write(i *int) {
  time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
  rw.Lock()
  fmt.Println("dapat write lock")
  temp := *i
  time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
  *i = temp + 1
  fmt.Println("melepaskan write lock", *i)
  rw.Unlock()
  wait.Done()
}

Contoh ini memulai 3 goroutine tulis, 7 goroutine baca, saat membaca data akan pertama获得 read lock, goroutine baca dapat normal获得 read lock, tetapi akan阻塞 goroutine tulis, saat获得 write lock, akan sekaligus阻塞 goroutine baca dan goroutine tulis, sampai melepaskan write lock. Dengan demikian mengimplementasikan mutual exclusion antara goroutine baca dan goroutine tulis, menjamin kebenaran data. Output contoh sebagai berikut:

dapat read lock
dapat read lock
dapat read lock
dapat read lock
melepaskan read lock 0
melepaskan read lock 0
melepaskan read lock 0
melepaskan read lock 0
dapat write lock
melepaskan write lock 1
dapat read lock
dapat read lock
dapat read lock
melepaskan read lock 1
melepaskan read lock 1
melepaskan read lock 1
dapat write lock
melepaskan write lock 2
dapat write lock
melepaskan write lock 3
hasil akhir 3

TIP

Untuk lock, tidak boleh meneruskan dan menyimpan sebagai nilai, harus menggunakan pointer.

Condition Variable

Condition variable, muncul dan digunakan bersama mutual exclusion lock, oleh karena itu beberapa orang mungkin salah menyebutnya condition lock, tetapi ia bukan lock, ia adalah mekanisme komunikasi. sync.Cond di Go menyediakan implementasi untuk ini, dan signature fungsi untuk membuat condition variable sebagai berikut:

func NewCond(l Locker) *Cond

Dapat dilihat prasyarat membuat condition variable adalah perlu membuat lock, sync.Cond menyediakan metode berikut untuk digunakan

//阻塞 menunggu condition berlaku, sampai dibangunkan
func (c *Cond) Wait()

// Membangunkan satu goroutine yang阻塞 karena condition
func (c *Cond) Signal()

// Membangunkan semua goroutine yang阻塞 karena condition
func (c *Cond) Broadcast()

Condition variable sangat mudah digunakan, sedikit modifikasi contoh read-write mutual exclusion lock di atas

var wait sync.WaitGroup
var count = 0

var rw sync.RWMutex

// Condition variable
var cond = sync.NewCond(rw.RLocker())

func main() {
  wait.Add(12)
  // Lebih banyak baca daripada tulis
  go func() {
    for i := 0; i < 3; i++ {
      go Write(&count)
    }
    wait.Done()
  }()
  go func() {
    for i := 0; i < 7; i++ {
      go Read(&count)
    }
    wait.Done()
  }()
  // Menunggu goroutine anak selesai
  wait.Wait()
  fmt.Println("hasil akhir", count)
}

func Read(i *int) {
  time.Sleep(time.Millisecond * time.Duration(rand.Intn(500)))
  rw.RLock()
  fmt.Println("dapat read lock")
  // Jika condition tidak terpenuhi akan terus阻塞
  for *i < 3 {
    cond.Wait()
  }
  time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
  fmt.Println("melepaskan read lock", *i)
  rw.RUnlock()
  wait.Done()
}

func Write(i *int) {
  time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
  rw.Lock()
  fmt.Println("dapat write lock")
  temp := *i
  time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
  *i = temp + 1
  fmt.Println("melepaskan write lock", *i)
  rw.Unlock()
  // Membangunkan semua goroutine yang阻塞 karena condition variable
  cond.Broadcast()
  wait.Done()
}

Saat membuat condition variable, karena di sini condition variable bekerja pada goroutine baca, oleh karena itu read lock diteruskan sebagai mutual exclusion lock, jika langsung meneruskan read-write mutual exclusion lock akan menyebabkan masalah unlock ulang goroutine tulis. Di sini diteruskan adalah sync.rlocker, diperoleh melalui metode RWMutex.RLocker.

func (rw *RWMutex) RLocker() Locker {
   return (*rlocker)(rw)
}

type rlocker RWMutex

func (r *rlocker) Lock()   { (*RWMutex)(r).RLock() }
func (r *rlocker) Unlock() { (*RWMutex)(r).RUnlock() }

Dapat dilihat rlocker hanya membungkus operasi read lock read-write mutual exclusion lock, sebenarnya referensi yang sama, tetap lock yang sama. Saat goroutine baca membaca data, jika kurang dari 3 akan terus阻塞 menunggu, sampai data lebih besar dari 3, sedangkan goroutine tulis setelah memperbarui data akan mencoba membangunkan semua goroutine yang阻塞 karena condition variable, oleh karena itu output akhir sebagai berikut

dapat read lock
dapat read lock
dapat read lock
dapat read lock
dapat write lock
melepaskan write lock 1
dapat read lock
dapat write lock
melepaskan write lock 2
dapat read lock
dapat read lock
dapat write lock
melepaskan write lock 3 // Goroutine tulis ketiga selesai eksekusi
melepaskan read lock 3
melepaskan read lock 3
melepaskan read lock 3
melepaskan read lock 3
melepaskan read lock 3
melepaskan read lock 3
melepaskan read lock 3
hasil akhir 3

Dari hasil dapat dilihat, ketika goroutine tulis ketiga selesai memperbarui data, tujuh goroutine baca yang阻塞 karena condition variable semuanya pulih berjalan.

TIP

Untuk condition variable, harus menggunakan for bukan if, harus menggunakan loop untuk判断 apakah condition terpenuhi, karena goroutine saat dibangunkan tidak dapat menjamin condition saat ini sudah terpenuhi.

for !condition {
  cond.Wait()
}

sync

Sebagian besar alat konkurensi di Go disediakan oleh paket standar sync, di atas sudah介绍过了 sync.WaitGroup, sync.Locker dll. Selain itu, paket sync juga menyediakan beberapa alat lain yang dapat digunakan.

Once

Saat menggunakan beberapa struktur data, jika struktur data ini terlalu besar, dapat mempertimbangkan menggunakan lazy loading, yaitu baru menginisialisasi struktur data ini saat benar-benar akan menggunakannya. Seperti contoh berikut

type MySlice []int

func (m *MySlice) Get(i int) (int, bool) {
   if *m == nil {
      return 0, false
   } else {
      return (*m)[i], true
   }
}

func (m *MySlice) Add(i int) {
   // Saat benar-benar menggunakan slice, baru mempertimbangkan untuk menginisialisasi
   if *m == nil {
      *m = make([]int, 0, 10)
   }
   *m = append(*m, i)
}

Masalahnya adalah, jika hanya satu goroutine yang menggunakan pasti tidak ada masalah, tetapi jika beberapa goroutine mengakses mungkin akan terjadi masalah. Misalnya goroutine A dan B secara bersamaan memanggil metode Add, A eksekusi sedikit lebih cepat, sudah selesai inisialisasi, dan berhasil menambahkan data, kemudian goroutine B menginisialisasi lagi, dengan demikian langsung menimpa data yang ditambahkan goroutine A, ini adalah masalahnya.

Ini adalah masalah yang ingin diselesaikan sync.Once, sesuai namanya, Once diterjemahkan sebagai sekali, sync.Once menjamin operasi yang ditentukan hanya akan dieksekusi sekali dalam kondisi konkuren. Penggunaannya sangat sederhana, hanya mengekspos satu metode Do, signature sebagai berikut:

func (o *Once) Do(f func())

Saat menggunakan, hanya perlu meneruskan operasi inisialisasi ke metode Do, sebagai berikut

var wait sync.WaitGroup

func main() {
  var slice MySlice
  wait.Add(4)
  for i := 0; i < 4; i++ {
    go func() {
      slice.Add(1)
      wait.Done()
    }()
  }
  wait.Wait()
  fmt.Println(slice.Len())
}

type MySlice struct {
  s []int
  o sync.Once
}

func (m *MySlice) Get(i int) (int, bool) {
  if m.s == nil {
    return 0, false
  } else {
    return m.s[i], true
  }
}

func (m *MySlice) Add(i int) {
  // Saat benar-benar menggunakan slice, baru mempertimbangkan untuk menginisialisasi
  m.o.Do(func() {
    fmt.Println("inisialisasi")
    if m.s == nil {
      m.s = make([]int, 0, 10)
    }
  })
  m.s = append(m.s, i)
}

func (m *MySlice) Len() int {
  return len(m.s)
}

Output sebagai berikut

inisialisasi
4

Dari hasil output dapat dilihat, semua data berhasil ditambahkan ke slice, operasi inisialisasi hanya dieksekusi sekali. Sebenarnya implementasi sync.Once sangat sederhana, menghapus komentar logika kode sebenarnya hanya 16 baris, prinsipnya adalah lock + operasi atom. Source code sebagai berikut:

type Once struct {
    // Digunakan untuk判断 apakah operasi sudah dieksekusi
  done uint32
  m    Mutex
}

func (o *Once) Do(f func()) {
  // Atomic load data
  if atomic.LoadUint32(&o.done) == 0 {
    o.doSlow(f)
  }
}

func (o *Once) doSlow(f func()) {
    // Lock
  o.m.Lock()
    // Unlock
  defer o.m.Unlock()
    //判断 apakah sudah dieksekusi
  if o.done == 0 {
        // Setelah selesai eksekusi修改 done
    defer atomic.StoreUint32(&o.done, 1)
    f()
  }
}

Pool

Tujuan desain sync.Pool adalah untuk menyimpan object sementara untuk reuse berikutnya, adalah object pool concurrent-safe sementara, menyimpan object yang暂时 tidak digunakan ke dalam pool, tidak perlu membuat object tambahan saat penggunaan berikutnya dapat langsung reuse, mengurangi frekuensi alokasi dan release memori, poin paling penting adalah mengurangi tekanan GC. sync.Pool total hanya memiliki dua metode, sebagai berikut:

// Meminta satu object
func (p *Pool) Get() any

// Memasukkan satu object
func (p *Pool) Put(x any)

Dan sync.Pool memiliki field New yang diekspos ke luar, digunakan untuk menginisialisasi object saat pool tidak dapat meminta object

New func() any

Berikut演示 dengan contoh

var wait sync.WaitGroup

// Object pool sementara
var pool sync.Pool

// Digunakan untuk menghitung total berapa object yang dibuat selama proses
var numOfObject atomic.Int64

// BigMemData Anggap ini adalah struct yang占用 memori besar
type BigMemData struct {
   M string
}

func main() {
   pool.New = func() any {
      numOfObject.Add(1)
      return BigMemData{"memori besar"}
   }
   wait.Add(1000)
   // Di sini memulai 1000 goroutine
   for i := 0; i < 1000; i++ {
      go func() {
         // Meminta object
         val := pool.Get()
         // Menggunakan object
         _ = val.(BigMemData)
         // Melepaskan object setelah selesai digunakan
         pool.Put(val)
         wait.Done()
      }()
   }
   wait.Wait()
   fmt.Println(numOfObject.Load())
}

Contoh ini memulai 1000 goroutine terus-menerus meminta dan melepaskan object di pool, jika tidak menggunakan object pool, maka 1000 goroutine perlu masing-masing instantiate object, dan 1000 object setelah instantiate ini perlu GC melepaskan memori setelah selesai digunakan, jika ada puluhan ribu goroutine atau biaya membuat object ini sangat tinggi, dalam situasi ini akan占用 memori besar dan带来 tekanan sangat besar pada GC, menggunakan object pool dapat reuse object mengurangi frekuensi instantiate, misalnya output contoh di atas mungkin sebagai berikut:

5

Meskipun memulai 1000 goroutine, seluruh proses hanya membuat 5 object, jika tidak menggunakan object pool 1000 goroutine akan membuat 1000 object, peningkatan yang dibawa oleh optimasi ini jelas, terutama saat volume konkurensi sangat besar dan biaya instantiate object sangat tinggi lebih dapat mencerminkan keunggulan.

Saat menggunakan sync.Pool perlu memperhatikan beberapa poin:

• Object sementara: sync.Pool hanya cocok untuk menyimpan object sementara, object di pool mungkin dihapus oleh GC tanpa pemberitahuan apapun, oleh karena itu tidak disarankan menyimpan koneksi jaringan, koneksi database这类 ke dalam sync.Pool.
• Tidak dapat diprediksi: sync.Pool saat meminta object, tidak dapat memprediksi object ini adalah baru dibuat atau reuse, juga tidak dapat mengetahui ada berapa object di pool
• Concurrent-safe:官方 menjamin sync.Pool pasti concurrent-safe, tetapi tidak menjamin fungsi New yang digunakan untuk membuat object pasti concurrent-safe, fungsi New diteruskan oleh pengguna, oleh karena itu concurrent-safety fungsi New harus dijaga oleh pengguna sendiri, ini juga alasan mengapa contoh di atas menggunakan atomic value untuk menghitung object.

TIP

Poin terakhir yang perlu diperhatikan, setelah selesai menggunakan object, harus melepaskan kembali ke pool, jika digunakan tidak dilepaskan maka penggunaan object pool tidak ada artinya.

Paket standar fmt memiliki contoh penggunaan object pool, di fungsi fmt.Fprintf

func Fprintf(w io.Writer, format string, a ...any) (n int, err error) {
   // Meminta buffer print
   p := newPrinter()
   p.doPrintf(format, a)
   n, err = w.Write(p.buf)
   // Melepaskan setelah selesai digunakan
   p.free()
   return
}

Di antaranya implementasi fungsi newPointer dan metode free sebagai berikut

func newPrinter() *pp {
   // Satu object yang diminta dari object pool
   p := ppFree.Get().(*pp)
   p.panicking = false
   p.erroring = false
   p.wrapErrs = false
   p.fmt.init(&p.buf)
   return p
}

func (p *pp) free() {
    // Agar ukuran buffer di object pool kurang lebih sama untuk kontrol buffer yang lebih elastis
    // Buffer yang terlalu besar tidak perlu dikembalikan ke object pool
  if cap(p.buf) > 64<<10 {
    return
  }
  // Field direset setelah melepaskan object ke pool
  p.buf = p.buf[:0]
  p.arg = nil
  p.value = reflect.Value{}
  p.wrappedErr = nil
  ppFree.Put(p)
}

Map

sync.Map adalah implementasi Map concurrent-safe yang disediakan官方,开箱即用, penggunaannya sangat sederhana, berikut adalah metode yang diekspos struct ini:

// Membaca nilai berdasarkan key, nilai return akan mengembalikan nilai yang sesuai dan apakah nilai ini ada
func (m *Map) Load(key any) (value any, ok bool)

// Menyimpan pasangan key-value
func (m *Map) Store(key, value any)

// Menghapus pasangan key-value
func (m *Map) Delete(key any)

// Jika key sudah ada, mengembalikan nilai asli, jika tidak menyimpan nilai baru dan mengembalikan, saat berhasil membaca nilai, loaded adalah true, jika tidak adalah false
func (m *Map) LoadOrStore(key, value any) (actual any, loaded bool)

// Menghapus pasangan key-value, dan mengembalikan nilai aslinya, nilai loaded tergantung pada apakah key ada
func (m *Map) LoadAndDelete(key any) (value any, loaded bool)

//遍历 Map, saat f() mengembalikan false, akan berhenti遍历
func (m *Map) Range(f func(key, value any) bool)

Berikut contoh sederhana untuk演示 penggunaan dasar sync.Map

func main() {
  var syncMap sync.Map
  // Menyimpan data
  syncMap.Store("a", 1)
  syncMap.Store("a", "a")
  // Membaca data
  fmt.Println(syncMap.Load("a"))
  // Membaca dan menghapus
  fmt.Println(syncMap.LoadAndDelete("a"))
  // Membaca atau menyimpan
  fmt.Println(syncMap.LoadOrStore("a", "hello world"))
  syncMap.Store("b", "goodbye world")
  //遍历 map
  syncMap.Range(func(key, value any) bool {
    fmt.Println(key, value)
    return true
  })
}

Output

a true
a true
hello world false
a hello world
b goodbye world

Selanjutnya lihat contoh penggunaan map secara konkuren:

func main() {
  myMap := make(map[int]int, 10)
  var wait sync.WaitGroup
  wait.Add(10)
  for i := 0; i < 10; i++ {
    go func(n int) {
      for i := 0; i < 100; i++ {
        myMap[n] = n
      }
      wait.Done()
    }(i)
  }
  wait.Wait()
}

Contoh di atas menggunakan map biasa, memulai 10 goroutine terus-menerus menyimpan data, jelas ini sangat mungkin memicu fatal, hasilnya kemungkinan besar sebagai berikut

fatal error: concurrent map writes

Menggunakan sync.Map dapat menghindari masalah ini

func main() {
  var syncMap sync.Map
  var wait sync.WaitGroup
  wait.Add(10)
  for i := 0; i < 10; i++ {
    go func(n int) {
      for i := 0; i < 100; i++ {
        syncMap.Store(n, n)
      }
      wait.Done()
    }(i)
  }
  wait.Wait()
  syncMap.Range(func(key, value any) bool {
    fmt.Println(key, value)
    return true
  })
}

Output sebagai berikut

8 8
3 3
1 1
9 9
6 6
5 5
7 7
0 0
2 2
4 4

Demi concurrent-safety pasti perlu mengorbankan tertentu, performa sync.Map sekitar 10-100 kali lebih rendah dari map.

Atomic

Dalam ilmu komputer, atom atau operasi primitif, biasanya digunakan untuk menyatakan beberapa operasi yang tidak dapat dibagi lebih halus, karena operasi ini tidak dapat dibagi menjadi langkah lebih kecil, sebelum selesai eksekusi, tidak akan diinterupsi oleh goroutine lain apapun, oleh karena itu hasil eksekusi baik berhasil atau gagal, tidak ada situasi ketiga, jika muncul situasi lain, maka itu bukan operasi atom. Contoh seperti kode berikut:

func main() {
  a := 0
  if a == 0 {
    a = 1
  }
  fmt.Println(a)
}

Kode di atas adalah branch判断 sederhana, meskipun kode sangat sedikit, tetapi juga bukan operasi atom, operasi atom sebenarnya didukung oleh level instruksi hardware.

Tipe

Untungnya sebagian besar kasus tidak perlu menulis assembly sendiri, paket standar sync/atomic di Go sudah menyediakan API terkait operasi atom, ia menyediakan beberapa tipe berikut untuk melakukan operasi atom.

atomic.Bool{}
atomic.Pointer[]{}
atomic.Int32{}
atomic.Int64{}
atomic.Uint32{}
atomic.Uint64{}
atomic.Uintptr{}
atomic.Value{}

Di antaranya tipe atom Pointer mendukung generics, tipe Value mendukung menyimpan tipe apapun, selain itu juga menyediakan banyak fungsi untuk memudahkan operasi. Karena granularitas operasi atom terlalu halus, dalam sebagian besar kasus, lebih cocok menangani data dasar这类.

TIP

Operasi atom di paket atmoic hanya memiliki signature fungsi, tidak ada implementasi spesifik, implementasi spesifik ditulis oleh assembly plan9.

Penggunaan

Setiap tipe atom akan menyediakan tiga metode berikut:

• Load(): Mendapatkan nilai secara atom
• Swap(newVal type) (old type): Menukar nilai secara atom, dan mengembalikan nilai lama
• Store(val type): Menyimpan nilai secara atom

Tipe berbeda mungkin memiliki metode tambahan lainnya, misalnya tipe integer akan menyediakan metode Add untuk mengimplementasikan operasi tambah kurang atom. Berikut contoh演示 dengan tipe int64:

func main() {
  var aint64 atomic.Uint64
  // Menyimpan nilai
  aint64.Store(64)
  // Menukar nilai
  aint64.Swap(128)
  // Menambah
  aint64.Add(112)
    // Memuat nilai
  fmt.Println(aint64.Load())
}

Atau dapat langsung menggunakan fungsi

func main() {
   var aint64 int64
   // Menyimpan nilai
   atomic.StoreInt64(&aint64, 64)
   // Menukar nilai
   atomic.SwapInt64(&aint64, 128)
   // Menambah
   atomic.AddInt64(&aint64, 112)
   // Memuat
   fmt.Println(atomic.LoadInt64(&aint64))
}

Penggunaan tipe lain juga sangat mirip, output akhir adalah:

240

CAS

Paket atomic juga menyediakan operasi CompareAndSwap, yaitu CAS. Ini adalah inti implementasi optimistic lock dan lock-free data structure. Optimistic lock sendiri bukan lock, adalah cara kontrol konkurensi tanpa lock dalam kondisi konkuren: thread/goroutine sebelum memodifikasi data, tidak akan pertama lock, tetapi pertama membaca data, melakukan komputasi, lalu saat submit修改 menggunakan CAS untuk判断 apakah ada thread lain memodifikasi data ini selama periode ini. Jika tidak (nilai masih sama dengan nilai yang dibaca sebelumnya), maka修改 berhasil; jika tidak, gagal dan retry. Oleh karena itu alasan disebut optimistic lock adalah karena ia selalu optimis mengasumsikan data bersama tidak akan dimodifikasi, hanya saat menemukan data tidak dimodifikasi baru akan mengeksekusi operasi yang sesuai, sedangkan mutual exclusion yang dikenal sebelumnya adalah pessimistic lock, mutual exclusion selalu pesimis mengasumsikan data bersama pasti akan dimodifikasi, oleh karena itu saat operasi akan lock, setelah selesai operasi akan unlock. Karena konkurensi yang diimplementasikan tanpa lock, keamanan dan efisiensinya relatif terhadap lock lebih tinggi, banyak struktur data concurrent-safe menggunakan CAS untuk implementasi, tetapi efisiensi sebenarnya harus dikombinasikan dengan situasi penggunaan spesifik. Lihat contoh berikut:

var lock sync.Mutex

var count int

func Add(num int) {
   lock.Lock()
   count += num
   lock.Unlock()
}

Ini adalah contoh menggunakan mutual exclusion lock, setiap kali menambah angka akan pertama lock, setelah selesai eksekusi akan unlock, proses akan menyebabkan goroutine lain阻塞, selanjutnya menggunakan CAS untuk改造:

var count int64

func Add(num int64) {
  for {
    expect := atomic.LoadInt64(&count)
    if atomic.CompareAndSwapInt64(&count, expect, expect+num) {
      break
    }
  }
}

Untuk CAS, ada tiga parameter, nilai memori, nilai harapan, nilai baru. Saat eksekusi, CAS akan membandingkan nilai harapan dengan nilai memori saat ini, jika nilai memori sama dengan nilai harapan, akan mengeksekusi operasi selanjutnya, jika tidak tidak melakukan apapun. Untuk operasi atom di bawah paket atomic Go, fungsi terkait CAS perlu meneruskan alamat, nilai harapan, nilai baru, dan akan mengembalikan nilai boolean apakah berhasil mengganti. Misalnya signature fungsi operasi CAS tipe int64 sebagai berikut:

func CompareAndSwapInt64(addr *int64, old, new int64) (swapped bool)

Di contoh CAS, pertama akan mendapatkan nilai harapan melalui LoadInt64, lalu menggunakan CompareAndSwapInt64 untuk membandingkan dan menukar, jika tidak berhasil akan terus loop, sampai berhasil. Operasi tanpa lock seperti ini tidak akan menyebabkan goroutine阻塞, tetapi loop terus-menerus bagi CPU tetap merupakan overhead yang tidak kecil, oleh karena itu dalam beberapa implementasi gagal mencapai jumlah tertentu mungkin akan放弃 operasi. Tetapi untuk operasi di atas, hanya penjumlahan angka sederhana, operasi yang terlibat tidak kompleks, oleh karena itu完全可以 mempertimbangkan implementasi tanpa lock.

TIP

Sebagian besar kasus, hanya membandingkan nilai tidak dapat mencapai concurrent-safety, misalnya masalah ABA yang disebabkan CAS, perlu menggunakan version tambahan untuk menyelesaikan masalah.

Value

Struct atomic.Value, dapat menyimpan nilai tipe apapun, struct sebagai berikut

type Value struct {
   // Tipe any
   v any
}

Meskipun dapat menyimpan tipe apapun, tetapi ia tidak dapat menyimpan nil, dan nilai yang disimpan sebelumnya dan sesudahnya harus konsisten, dua contoh berikut tidak dapat dikompilasi

func main() {
   var val atomic.Value
   val.Store(nil)
   fmt.Println(val.Load())
}
// panic: sync/atomic: store of nil value into Value

func main() {
   var val atomic.Value
   val.Store("hello world")
   val.Store(114514)
   fmt.Println(val.Load())
}
// panic: sync/atomic: store of inconsistently typed value into Value

Selain itu, penggunaannya tidak terlalu berbeda dengan tipe atom lainnya, dan perlu diperhatikan, semua tipe atom tidak boleh copy nilai, tetapi harus menggunakan pointer mereka.