【Go】原子并发操作

一、基本概念

支持的数据类型

主要函数

使用场景

二、基础代码实例

开协程给原子变量做加法

统计多个变量

原子标志判断

三、并发日志记录器

四、并发计数器与性能监控

五、优雅的停止并发任务

worker函数

Main函数

应用价值

Go语言中，原子并发操作是非常常用的，确保协程环境中对资源的共访是安全的。Go的sync/atomic包提供了一系列底层的原子性操作函数，允许你在基本数据类型上执行无锁的线程安全操作。使用原子操作可以避免在并发访问时使用互斥锁（mutexes），从而在某些情况下提高性能。

一、基本概念

原子操作可以保证在多线程环境中，单个操作是不可中断的，即在完成之前不会被线程切换影响。这是通过硬件级别的支持实现的，确保了操作的原子性。

支持的数据类型

Go的sync/atomic包支持几种基本数据类型的原子操作：

int32, int64
uint32, uint64
uintptr
Pointer（对于任意类型的原子指针操作）

主要函数

AddInt32, AddInt64, AddUint32, AddUint64: 原子加法操作。
LoadInt32, LoadInt64, LoadUint32, LoadUint64: 原子加载操作，用于读取一个值。
StoreInt32, StoreInt64, StoreUint32, StoreUint64: 原子存储操作，用于写入一个值。
SwapInt32, SwapInt64, SwapUint32, SwapUint64: 原子交换操作，写入新值并返回旧值。
CompareAndSwapInt32, CompareAndSwapInt64, CompareAndSwapUint32, CompareAndSwapUint64: 比较并交换操作，如果当前值等于旧值，则写入新值。

使用场景

原子操作通常用在性能敏感且要求高并发的场景，例如：

实时计数器
状态标志
无锁数据结构的实现

原子操作提供了一种比锁更轻量级的并发控制方法，尤其适用于操作简单且频繁的场景。不过，原子操作的使用需要更谨慎，以避免复杂逻辑中可能的逻辑错误。在设计并发控制策略时，适当的选择使用锁还是原子操作，可以帮助你更好地平衡性能和开发效率。

二、基础代码实例

开协程给原子变量做加法

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
)

func main() {

	// We'll use an atomic integer type to represent our
	// (always-positive) counter.
	var ops atomic.Uint64

	// A WaitGroup will help us wait for all goroutines
	// to finish their work.
	var wg sync.WaitGroup

	// We'll start 50 goroutines that each increment the
	// counter exactly 1000 times.
	for i := 0; i < 500; i++ {
		wg.Add(1)

		go func() {
			for c := 0; c < 10000; c++ {

				// To atomically increment the counter we use `Add`.
				ops.Add(1)
			}

			wg.Done()
		}()
	}

	// Wait until all the goroutines are done.
	wg.Wait()

	// Here no goroutines are writing to 'ops', but using
	// `Load` it's safe to atomically read a value even while
	// other goroutines are (atomically) updating it.
	fmt.Println("ops:", ops.Load())
}

统计多个变量

我们可以使用多个原子变量来跟踪不同类型的操作。

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
)

func main() {
	var adds atomic.Uint64
	var subs atomic.Uint64
	var wg sync.WaitGroup

	for i := 0; i < 50; i++ {
		wg.Add(1)
		go func() {
			for c := 0; c < 1000; c++ {
				adds.Add(1)
			}
			wg.Done()
		}()

		wg.Add(1)
		go func() {
			for c := 0; c < 1000; c++ {
				subs.Add(1)
			}
			wg.Done()
		}()
	}

	wg.Wait()
	fmt.Println("Adds:", adds.Load(), "Subs:", subs.Load())
}

原子标志判断

使用原子变量作为一个简单的标志来控制是否所有协程都应该停止工作。

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

func main() {
	var ops atomic.Uint64
	var stopFlag atomic.Bool
	var wg sync.WaitGroup

	for i := 0; i < 50; i++ {
		wg.Add(1)
		go func() {
			for !stopFlag.Load() {
				ops.Add(1)
				time.Sleep(10 * time.Millisecond) // 减缓增加速度
			}
			wg.Done()
		}()
	}

	time.Sleep(500 * time.Millisecond) // 运行一段时间后
	stopFlag.Store(true)               // 设置停止标志
	wg.Wait()
	fmt.Println("ops:", ops.Load())
}

三、并发日志记录器

在一个多协程环境中，我们可能需要记录日志，但又希望避免因为并发写入而导致的问题。我们可以使用sync.Mutex来确保日志写入的原子性。

package main

import (
	"fmt"
	"log"
	"os"
	"sync"
)

var (
	logger   *log.Logger
	logMutex sync.Mutex
)

func main() {
	file, err := os.OpenFile("log.log", os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0666)

	if err != nil {
		log.Fatalln("Failed to open log file")

	}
	defer file.Close()
	logger = log.New(file, "", log.Ldate|log.Ltime|log.Lshortfile)
	var wg sync.WaitGroup

	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func(id int) {
			logMutex.Lock()
			logger.Println("Goroutine %d is runing...", id)
			logMutex.Unlock()
			wg.Done()
		}(i)
	}
	wg.Wait()
	fmt.Println("All goroutines finished")
}

四、并发计数器与性能监控

在一个网络服务器或数据库中，我们可能需要监控并发请求的数量或特定资源的使用情况，使用原子操作可以无锁地实现这一点。

package main

import (
	"fmt"
	"net/http"
	"sync/atomic"
)

var requestCount atomic.Int64

func handler(w http.ResponseWriter, r *http.Request) {
	requestCount.Add(1)
	fmt.Fprintf(w, "Hello, visitor number %d!", requestCount.Load())
}

func main() {
	http.HandleFunc("/", handler)
	http.ListenAndServe(":8080", nil)
}

五、优雅的停止并发任务

在处理诸如网络服务或后台任务处理器的程序时，我们可能需要在收到停止信号后优雅地中断并发任务。

package main

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

func worker(ctx context.Context, id int, wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		select {
		case <-ctx.Done():
			fmt.Printf("Worker %d stopping\n", id)
			return
		default:
			fmt.Printf("Worker %d is working\n", id)
			time.Sleep(time.Second)
		}
	}
}

func main() {
	var wg sync.WaitGroup
	ctx, cancel := context.WithCancel(context.Background())

	for i := 1; i <= 5; i++ {
		wg.Add(1)
		go worker(ctx, i, &wg)
	}

	time.Sleep(5 * time.Second)
	cancel() // 发送取消信号
	wg.Wait()
	fmt.Println("All workers stopped.")
}

worker函数

func worker(ctx context.Context, id int, wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		select {
		case <-ctx.Done():
			fmt.Printf("Worker %d stopping\n", id)
			return
		default:
			fmt.Printf("Worker %d is working\n", id)
			time.Sleep(time.Second)
		}
	}
}

worker是一个协程函数，接受一个context.Context对象、一个整数id作为工人标识，和一个sync.WaitGroup来同步协程。
defer wg.Done(): 确保在函数返回时调用wg.Done()，表明该协程的工作已完成，这对于等待组来维护协程计数非常重要。
select语句用于处理多个通道操作。在这里，它监听ctx.Done()通道，这是context提供的方式，用于接收上下文取消事件。
- 当ctx.Done()通道接收到信号时（这发生在主协程调用cancel()函数时），输出停止信息，并通过return退出无限循环，结束协程执行。
- default分支在没有信号时执行，模拟工作负载并通过time.Sleep(time.Second)模拟一秒钟的工作时间。

Main函数

ctx, cancel := context.WithCancel(context.Background()): 创建一个可取消的上下文ctx，和一个cancel函数，用于发送取消信号。
循环启动5个工作协程，每个通过go worker(ctx, i, &wg)启动，并传递上下文、ID和等待组。
time.Sleep(5 * time.Second): 主协程等待5秒钟，给工作协程一定时间执行。
cancel(): 调用取消函数，这会向ctx.Done()发送信号，导致所有监听该通道的工作协程接收到取消事件并停止执行。
wg.Wait(): 阻塞直到所有工作协程调用wg.Done()，表明它们已经停止。
输出“All workers stopped.”表示所有工作协程已经优雅地停止。