Подробное применение пакета синхронизации языка Go

В параллельном программировании основная функция примитивов синхронизации, то есть того, что мы обычно называем блокировками, состоит в том, чтобы гарантировать, что несколько потоков илиgoroutineНет никакой путаницы при доступе к одному и тому же участку памяти.GoлингвистическийsyncПакет предоставляет общие примитивы синхронизации для параллельного программирования.Статья перепечатана в предыдущем выпуске "Примитивы синхронизации для параллельного программирования в Golangтакже подробно вMutex,RWMutex,WaitGroup,Onceа такжеCondПринцип реализации этих примитивов синхронизации. В сегодняшней статье давайте вернемся к прикладному уровню и сосредоточимся наsyncТакже будут представлены сценарии применения этих примитивов синхронизации в пакете.syncв упаковкеPoolа такжеMapсценарии применения и методы использования. Без дальнейших церемоний, давайте начнем.

sync.Mutex

sync.Mutexможет бытьsyncНаиболее широко используемый примитив в пакете. Он разрешает взаимоисключающий доступ (не одновременно) к общим ресурсам:

mutex := &sync.Mutex{}

mutex.Lock()
// Update共享变量 (比如切片，结构体指针等)
mutex.Unlock()

Следует отметить, что после первого использованияsync.Mutexсделать копию. (syncВсе примитивы пакета одинаковы). Если в структуре есть поле примитива синхронизации, оно должно передаваться по указателю.

sync.RWMutex

sync.RWMutexэто мьютекс для чтения и записи, который обеспечивает то, что мы только что видели вышеsync.MutexизLockа такжеUnLockметод (поскольку обе структуры реализуютsync.Lockerинтерфейс). Однако он также позволяет использоватьRLockа такжеRUnlockметод для параллельного чтения:

mutex := &sync.RWMutex{}

mutex.Lock()
// Update 共享变量
mutex.Unlock()

mutex.RLock()
// Read 共享变量
mutex.RUnlock()

sync.RWMutexразрешить существование по крайней мере одной блокировки чтения или одной блокировки записи, в то время какsync.MutexРазрешить существование блокировки чтения или записи.

Сравните производительность этих методов путем сравнительного анализа:

BenchmarkMutexLock-4       83497579         17.7 ns/op
BenchmarkRWMutexLock-4     35286374         44.3 ns/op
BenchmarkRWMutexRLock-4    89403342         15.3 ns/op

Видно заблокировано/разблокированоsync.RWMutexБлокировки чтения быстрее, чем блокировка/разблокировкаsync.Mutexбыстрее, с другой стороны, вsync.RWMutexзвонитьLock()/ Unlock()самая медленная операция.

Поэтому его следует использовать только в сценариях с частым чтением и нечастой записью.sync.RWMutex.

sync.WaitGroup

sync.WaitGroupЭто также часто используемый примитив синхронизации.goroutineжду группуgoroutineИсполнение завершено.

sync.WaitGroupИмеет внутренний счетчик. когда счетчик равен0время, тогдаWait()Метод возвращается немедленно. В противном случае он заблокирует выполнениеWait()методgoroutineпока счетчик не сравняется0время.

Чтобы увеличить счетчик, мы должны использоватьAdd(int)метод. Чтобы уменьшить его, мы можем использоватьDone()(уменьшить счетчик1), вы также можете передавать отрицательные числа вAddметод уменьшает счетчик на указанный размер,Done()Нижний слой метода сквозной.Add(-1)осуществленный.

В следующем примере мы начнем с восьмиgoroutine, и дождитесь их завершения:

wg := &sync.WaitGroup{}

for i := 0; i < 8; i++ {
  wg.Add(1)
  go func() {
    // Do something
    wg.Done()
  }()
}

wg.Wait()
// 继续往下执行...

каждый раз, когда вы создаетеgoroutine, мы будем использоватьwg.Add(1)увеличиватьwgвнутренний счетчик. мы также можемforвызывается перед цикломwg.Add(8).

При этом каждыйgoroutineКогда это сделано, оба используютwg.Done()уменьшатьwgвнутренний счетчик.

main goroutineбудет через восемьgoroutineоба исполняютwg.Done()изменить счетчик на0Затем вы можете продолжить выполнение.

sync.Map

sync.Mapэто параллельная версияGoлингвистическийmapМы можем:

использоватьStore(interface {}，interface {})Добавьте элементы.
использоватьLoad(interface {}) interface {}Получить элементы.
использоватьDelete(interface {})Удалить элементы.
использоватьLoadOrStore(interface {}，interface {}) (interface {}，bool)Получить или добавить элементы, которых раньше не было. если бы ключ был раньшеmapсуществует, возвращенное логическое значениеtrue.
использоватьRangeИтерация по элементам.

m := &sync.Map{}

// 添加元素
m.Store(1, "one")
m.Store(2, "two")

// 获取元素1
value, contains := m.Load(1)
if contains {
  fmt.Printf("%s\n", value.(string))
}

// 返回已存value，否则把指定的键值存储到map中
value, loaded := m.LoadOrStore(3, "three")
if !loaded {
  fmt.Printf("%s\n", value.(string))
}

m.Delete(3)

// 迭代所有元素
m.Range(func(key, value interface{}) bool {
  fmt.Printf("%d: %s\n", key.(int), value.(string))
  return true
})

Вышеупомянутая программа выведет:

one
three
1: one
2: two

Как вы видете,RangeМетод получает типfunc(key，value interface {})boolпараметры функции. если функция возвращаетfalse, итерация останавливается. Забавный факт, что даже если мы вернемся через постоянное времяfalse, временная сложность в худшем случае по-прежнемуO(n).

когда мы должны использоватьsync.Mapвместо обычногоmapиспользовать наsync.Mutex?

когда мыmapКогда есть частые чтения и нечастые записи.
когда несколькоgoroutineПри чтении, записи и перезаписи непересекающихся ключей. Что именно это означает? Например, если у нас есть реализация шардинга с набором из 4goroutine, каждыйgoroutineОтвечает за 25% ключей (каждый ответственный ключ не конфликтует). при этих обстоятельствах,sync.Mapявляется предпочтительным.

sync.Pool

sync.Poolэто параллельный пул, отвечающий за безопасное хранение набора объектов. Он имеет два метода экспорта:

Get() interface{}Используется для выборки элементов из параллельного пула.
Put(interface{})Добавляет объект в параллельный пул.

pool := &sync.Pool{}

pool.Put(NewConnection(1))
pool.Put(NewConnection(2))
pool.Put(NewConnection(3))

connection := pool.Get().(*Connection)
fmt.Printf("%d\n", connection.id)
connection = pool.Get().(*Connection)
fmt.Printf("%d\n", connection.id)
connection = pool.Get().(*Connection)
fmt.Printf("%d\n", connection.id)

выход:

1
3
2

должен быть в курсеGet()Метод будет случайным образом извлекать объекты из параллельного пула, и нет гарантии, что объекты, хранящиеся в параллельном пуле, будут получены в фиксированном порядке.

Может такжеsync.PoolУкажите метод создателя:

pool := &sync.Pool{
  New: func() interface{} {
    return NewConnection()
  },
}

connection := pool.Get().(*Connection)

Таким образом, каждый вызовGet(), будет возвращенpool.NewОбъект (в данном случае указатель), созданный функцией, указанной в .

Итак, когда использовать sync.Pool? Есть два варианта использования:

Во-первых, когда нам нужно повторно использовать общие и долгоживущие объекты (например, соединения с базой данных). Второй используется для оптимизации распределения памяти.

Рассмотрим пример функции, которая записывает буфер и сохраняет результат в файл в файле. использоватьsync.Pool, мы можем повторно использовать пространство, выделенное для буфера, повторно используя один и тот же объект между различными вызовами функций. Первым шагом является получение ранее выделенного буфера (если это первый вызов, создайте его, но это абстрактно). Потом,deferОперация состоит в том, чтобы вернуть буфер обратноsync.Poolсередина.

func writeFile(pool *sync.Pool, filename string) error {
	buf := pool.Get().(*bytes.Buffer)

  defer pool.Put(buf)

	// Reset 缓存区，不然会连接上次调用时保存在缓存区里的字符串foo
	// 编程foofoo 以此类推
	buf.Reset()

	buf.WriteString("foo")

	return ioutil.WriteFile(filename, buf.Bytes(), 0644)
}

sync.Once

sync.Once— это простой, но мощный примитив, обеспечивающий выполнение функции только один раз. В приведенном ниже примере есть только одинgoroutineОтображается выходное сообщение:

once := &sync.Once{}
for i := 0; i < 4; i++ {
	i := i
	go func() {
		once.Do(func() {
			fmt.Printf("first %d\n", i)
		})
	}()
}

мы использовалиDo(func ())метод, чтобы указать раздел, который можно вызвать только один раз.

sync.Cond

sync.Condможет бытьsyncНаименее распространенный из примитивов синхронизации, предоставляемых пакетом, он используется для передачи сигналов (один-к-одному) или широковещательной рассылки (один-ко-многим) наgoroutine. Давайте рассмотрим сценарий, в котором мы должны спроситьgoroutineУказывает, что первый элемент общего фрагмента был обновлен. Создайтеsync.Condнужноsync.Lockerобъект (sync.Mutexилиsync.RWMutex):

cond := sync.NewCond(&sync.Mutex{})

Затем напишем функцию, отвечающую за отображение первого элемента слайса:

func printFirstElement(s []int, cond *sync.Cond) {
	cond.L.Lock()
	cond.Wait()
	fmt.Printf("%d\n", s[0])
	cond.L.Unlock()
}

мы можем использоватьcond.LДоступ к внутреннему мьютексу. Как только замок будет получен, мы позвонимcond.Wait(), что сделает текущийgoroutineБлокируется до получения сигнала.

давай вернемсяmain goroutine. Мы поделимся фрагментом, передав его и ранее созданныйsync.CondсоздаватьprintFirstElementбассейн. Затем мы звонимget()функция, сохраняющая результат вs[0]в и сигнал:

s := make([]int, 1)
for i := 0; i < runtime.NumCPU(); i++ {
	go printFirstElement(s, cond)
}

i := get()
cond.L.Lock()
s[0] = i
cond.Signal()
cond.L.Unlock()

Этот сигнал деактивируетgoroutineзаблокированное состояние, разблокированоgoroutineпокажетs[0]значение, хранящееся в .

Однако некоторые могут возразить, что наш код неисправен.GoОдин из основных принципов:

Не общайтесь через общую память, вместо этого делитесь памятью через общение.

Действительно, в этом примере лучше использоватьchannelпройтиget()Возвращаемое значение. Но мы также упомянулиsync.CondМожет также использоваться для широковещательных сигналов. Давайте изменим приведенный выше пример наSignal()позвони вместо этогоBroadcast().

i := get()
cond.L.Lock()
s[0] = i
cond.Broadcast()
cond.L.Unlock()

В этом случае все горутины будут запущены. Как мы все знаем,channelЭлементы в будут состоять только изgoroutineполучено. пройти черезchannelЕдинственный выход - отключить аналоговое вещаниеchannel.

Когда канал закрыт, после того, как данные, которые были отправлены в канал, были успешно получены, последующие операции приема больше не будут блокироваться, и они немедленно вернут нулевое значение.

Но таким образом можно транслировать только один раз. Поэтому, хотя существует много споров, несомненно,sync.Condинтересная особенность .