Go学习笔记-Sync Package

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170

package main

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

// 1. 使用Mutex保护共享资源
type Counter struct {
    mu    sync.Mutex
    value int
}

func (c *Counter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.value++
}

func (c *Counter) Decrement() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.value--
}

func (c *Counter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.value
}

// 2. 使用RWMutex优化读多写少场景
type ReadWriteCounter struct {
    mu    sync.RWMutex
    value int
}

func (c *ReadWriteCounter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.value++
}

func (c *ReadWriteCounter) Value() int {
    c.mu.RLock()
    defer c.mu.RUnlock()
    return c.value
}

// 3. 缓存示例
type Cache struct {
    mu   sync.RWMutex
    data map[string]string
}

func NewCache() *Cache {
    return &Cache{
        data: make(map[string]string),
    }
}

func (c *Cache) Get(key string) (string, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    value, exists := c.data[key]
    return value, exists
}

func (c *Cache) Set(key, value string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.data[key] = value
}

func (c *Cache) Delete(key string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    delete(c.data, key)
}

func (c *Cache) Size() int {
    c.mu.RLock()
    defer c.mu.RUnlock()
    return len(c.data)
}

func main() {
    // 1. Mutex基本使用
    fmt.Println("=== Mutex基本使用 ===")
    
    counter := &Counter{}
    var wg sync.WaitGroup
    
    // 启动多个goroutine并发增加计数器
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for j := 0; j < 100; j++ {
                counter.Increment()
            }
            fmt.Printf("Goroutine %d 完成\n", id)
        }(i)
    }
    
    wg.Wait()
    fmt.Printf("最终计数: %d (期望: 1000)\n", counter.Value())
    
    // 2. RWMutex性能对比
    fmt.Println("\n=== RWMutex性能对比 ===")
    
    rwCounter := &ReadWriteCounter{}
    
    // 写操作
    start := time.Now()
    for i := 0; i < 1000; i++ {
        rwCounter.Increment()
    }
    writeTime := time.Since(start)
    
    // 读操作
    start = time.Now()
    for i := 0; i < 10000; i++ {
        rwCounter.Value()
    }
    readTime := time.Since(start)
    
    fmt.Printf("写操作耗时: %v\n", writeTime)
    fmt.Printf("读操作耗时: %v\n", readTime)
    fmt.Printf("最终值: %d\n", rwCounter.Value())
    
    // 3. 缓存使用示例
    fmt.Println("\n=== 缓存使用示例 ===")
    
    cache := NewCache()
    
    // 并发写入
    var cacheWg sync.WaitGroup
    for i := 0; i < 5; i++ {
        cacheWg.Add(1)
        go func(id int) {
            defer cacheWg.Done()
            for j := 0; j < 10; j++ {
                key := fmt.Sprintf("key-%d-%d", id, j)
                value := fmt.Sprintf("value-%d-%d", id, j)
                cache.Set(key, value)
            }
        }(i)
    }
    
    cacheWg.Wait()
    fmt.Printf("缓存大小: %d\n", cache.Size())
    
    // 并发读取
    for i := 0; i < 3; i++ {
        cacheWg.Add(1)
        go func(id int) {
            defer cacheWg.Done()
            key := fmt.Sprintf("key-%d-5", id)
            if value, exists := cache.Get(key); exists {
                fmt.Printf("读取到 %s: %s\n", key, value)
            } else {
                fmt.Printf("未找到 %s\n", key)
            }
        }(i)
    }
    
    cacheWg.Wait()
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283

package main

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

// 1. WaitGroup基本使用
func basicWaitGroup() {
    fmt.Println("=== WaitGroup基本使用 ===")
    
    var wg sync.WaitGroup
    
    // 启动多个goroutine
    for i := 1; i <= 5; i++ {
        wg.Add(1) // 增加等待计数
        go func(id int) {
            defer wg.Done() // 完成时减少计数
            
            fmt.Printf("Worker %d 开始工作\n", id)
            time.Sleep(time.Duration(id) * 200 * time.Millisecond)
            fmt.Printf("Worker %d 完成工作\n", id)
        }(i)
    }
    
    fmt.Println("等待所有worker完成...")
    wg.Wait() // 等待计数归零
    fmt.Println("所有worker已完成")
}

// 2. 分阶段任务处理
func phaseProcessing() {
    fmt.Println("\n=== 分阶段任务处理 ===")
    
    data := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
    
    // 第一阶段：数据预处理
    var phase1Wg sync.WaitGroup
    preprocessed := make([]int, len(data))
    
    for i, value := range data {
        phase1Wg.Add(1)
        go func(index, val int) {
            defer phase1Wg.Done()
            // 模拟预处理
            time.Sleep(50 * time.Millisecond)
            preprocessed[index] = val * 2
            fmt.Printf("预处理 %d -> %d\n", val, preprocessed[index])
        }(i, value)
    }
    
    phase1Wg.Wait()
    fmt.Println("第一阶段完成")
    
    // 第二阶段：数据处理
    var phase2Wg sync.WaitGroup
    results := make([]int, len(preprocessed))
    
    for i, value := range preprocessed {
        phase2Wg.Add(1)
        go func(index, val int) {
            defer phase2Wg.Done()
            // 模拟处理
            time.Sleep(30 * time.Millisecond)
            results[index] = val + 10
            fmt.Printf("处理 %d -> %d\n", val, results[index])
        }(i, value)
    }
    
    phase2Wg.Wait()
    fmt.Println("第二阶段完成")
    fmt.Printf("最终结果: %v\n", results)
}

// 3. sync.Once使用
var (
    instance *Singleton
    once     sync.Once
)

type Singleton struct {
    data string
}

func GetInstance() *Singleton {
    once.Do(func() {
        fmt.Println("创建单例实例")
        time.Sleep(100 * time.Millisecond) // 模拟初始化时间
        instance = &Singleton{
            data: "单例数据",
        }
    })
    return instance
}

func onceExample() {
    fmt.Println("\n=== sync.Once使用 ===")
    
    var wg sync.WaitGroup
    
    // 多个goroutine同时尝试获取单例
    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            
            fmt.Printf("Goroutine %d 尝试获取单例\n", id)
            singleton := GetInstance()
            fmt.Printf("Goroutine %d 获取到单例: %s\n", id, singleton.data)
        }(i)
    }
    
    wg.Wait()
}

// 4. 配置初始化示例
type Config struct {
    DatabaseURL string
    APIKey      string
    Debug       bool
}

var (
    config     *Config
    configOnce sync.Once
)

func loadConfig() *Config {
    configOnce.Do(func() {
        fmt.Println("加载配置...")
        time.Sleep(200 * time.Millisecond) // 模拟从文件或网络加载配置
        
        config = &Config{
            DatabaseURL: "postgres://localhost:5432/mydb",
            APIKey:      "secret-api-key",
            Debug:       true,
        }
        
        fmt.Println("配置加载完成")
    })
    return config
}

func configExample() {
    fmt.Println("\n=== 配置初始化示例 ===")
    
    var wg sync.WaitGroup
    
    // 多个服务同时需要配置
    services := []string{"UserService", "OrderService", "PaymentService"}
    
    for _, service := range services {
        wg.Add(1)
        go func(serviceName string) {
            defer wg.Done()
            
            fmt.Printf("%s 需要配置\n", serviceName)
            cfg := loadConfig()
            fmt.Printf("%s 获取到配置: Debug=%t\n", serviceName, cfg.Debug)
        }(service)
    }
    
    wg.Wait()
}

// 5. 错误处理与WaitGroup
func errorHandlingWithWaitGroup() {
    fmt.Println("\n=== 错误处理与WaitGroup ===")
    
    var wg sync.WaitGroup
    errors := make(chan error, 5)
    
    // 模拟可能出错的任务
    tasks := []int{1, 2, 3, 4, 5}
    
    for _, task := range tasks {
        wg.Add(1)
        go func(taskID int) {
            defer wg.Done()
            
            // 模拟任务执行
            time.Sleep(time.Duration(taskID*100) * time.Millisecond)
            
            // 模拟错误（任务3会失败）
            if taskID == 3 {
                errors <- fmt.Errorf("任务 %d 执行失败", taskID)
                return
            }
            
            fmt.Printf("任务 %d 执行成功\n", taskID)
        }(task)
    }
    
    // 等待所有任务完成
    go func() {
        wg.Wait()
        close(errors)
    }()
    
    // 收集错误
    var errorList []error
    for err := range errors {
        errorList = append(errorList, err)
    }
    
    if len(errorList) > 0 {
        fmt.Printf("发现 %d 个错误:\n", len(errorList))
        for _, err := range errorList {
            fmt.Printf("  - %v\n", err)
        }
    } else {
        fmt.Println("所有任务都成功完成")
    }
}

// 6. 工作池与WaitGroup
type WorkerPool struct {
    workerCount int
    jobs        chan func()
    wg          sync.WaitGroup
}

func NewWorkerPool(workerCount int) *WorkerPool {
    return &WorkerPool{
        workerCount: workerCount,
        jobs:        make(chan func(), 100),
    }
}

func (wp *WorkerPool) Start() {
    for i := 0; i < wp.workerCount; i++ {
        wp.wg.Add(1)
        go wp.worker(i)
    }
}

func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    
    for job := range wp.jobs {
        fmt.Printf("Worker %d 执行任务\n", id)
        job()
    }
}

func (wp *WorkerPool) Submit(job func()) {
    wp.jobs <- job
}

func (wp *WorkerPool) Stop() {
    close(wp.jobs)
    wp.wg.Wait()
}

func workerPoolExample() {
    fmt.Println("\n=== 工作池与WaitGroup ===")
    
    pool := NewWorkerPool(3)
    pool.Start()
    
    // 提交任务
    for i := 1; i <= 10; i++ {
        taskID := i
        pool.Submit(func() {
            time.Sleep(200 * time.Millisecond)
            fmt.Printf("任务 %d 完成\n", taskID)
        })
    }
    
    // 停止工作池
    pool.Stop()
    fmt.Println("工作池已停止")
}

func main() {
    basicWaitGroup()
    phaseProcessing()
    onceExample()
    configExample()
    errorHandlingWithWaitGroup()
    workerPoolExample()
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331

package main

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

// 1. sync.Cond使用示例
type Queue struct {
    mu    sync.Mutex
    cond  *sync.Cond
    items []string
}

func NewQueue() *Queue {
    q := &Queue{
        items: make([]string, 0),
    }
    q.cond = sync.NewCond(&q.mu)
    return q
}

func (q *Queue) Enqueue(item string) {
    q.mu.Lock()
    defer q.mu.Unlock()
    
    q.items = append(q.items, item)
    fmt.Printf("入队: %s (队列长度: %d)\n", item, len(q.items))
    
    // 通知等待的消费者
    q.cond.Signal()
}

func (q *Queue) Dequeue() string {
    q.mu.Lock()
    defer q.mu.Unlock()
    
    // 等待队列非空
    for len(q.items) == 0 {
        fmt.Println("队列为空，等待...")
        q.cond.Wait()
    }
    
    item := q.items[0]
    q.items = q.items[1:]
    fmt.Printf("出队: %s (队列长度: %d)\n", item, len(q.items))
    
    return item
}

func (q *Queue) Size() int {
    q.mu.Lock()
    defer q.mu.Unlock()
    return len(q.items)
}

// 2. 生产者-消费者模式
func producerConsumerExample() {
    fmt.Println("=== 生产者-消费者模式 ===")
    
    queue := NewQueue()
    var wg sync.WaitGroup
    
    // 启动消费者
    for i := 1; i <= 2; i++ {
        wg.Add(1)
        go func(consumerID int) {
            defer wg.Done()
            
            for j := 0; j < 5; j++ {
                item := queue.Dequeue()
                fmt.Printf("消费者 %d 处理: %s\n", consumerID, item)
                time.Sleep(200 * time.Millisecond)
            }
        }(i)
    }
    
    // 启动生产者
    wg.Add(1)
    go func() {
        defer wg.Done()
        
        for i := 1; i <= 10; i++ {
            item := fmt.Sprintf("item-%d", i)
            queue.Enqueue(item)
            time.Sleep(100 * time.Millisecond)
        }
    }()
    
    wg.Wait()
    fmt.Printf("最终队列大小: %d\n", queue.Size())
}

// 3. 原子操作示例
type AtomicCounter struct {
    value int64
}

func (c *AtomicCounter) Increment() {
    atomic.AddInt64(&c.value, 1)
}

func (c *AtomicCounter) Decrement() {
    atomic.AddInt64(&c.value, -1)
}

func (c *AtomicCounter) Value() int64 {
    return atomic.LoadInt64(&c.value)
}

func (c *AtomicCounter) CompareAndSwap(old, new int64) bool {
    return atomic.CompareAndSwapInt64(&c.value, old, new)
}

func atomicExample() {
    fmt.Println("\n=== 原子操作示例 ===")
    
    counter := &AtomicCounter{}
    var wg sync.WaitGroup
    
    // 并发增加
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            
            for j := 0; j < 1000; j++ {
                counter.Increment()
            }
            
            fmt.Printf("Goroutine %d 完成增加操作\n", id)
        }(i)
    }
    
    // 并发减少
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            
            for j := 0; j < 1000; j++ {
                counter.Decrement()
            }
            
            fmt.Printf("Goroutine %d 完成减少操作\n", id)
        }(i)
    }
    
    wg.Wait()
    fmt.Printf("最终计数: %d (期望: 5000)\n", counter.Value())
    
    // 测试CompareAndSwap
    fmt.Println("\n测试CompareAndSwap:")
    currentValue := counter.Value()
    newValue := currentValue + 100
    
    if counter.CompareAndSwap(currentValue, newValue) {
        fmt.Printf("CAS成功: %d -> %d\n", currentValue, newValue)
    } else {
        fmt.Printf("CAS失败，当前值: %d\n", counter.Value())
    }
}

// 4. 原子值操作
type Config struct {
    MaxConnections int
    Timeout        time.Duration
    Debug          bool
}

type AtomicConfig struct {
    value atomic.Value
}

func NewAtomicConfig(config Config) *AtomicConfig {
    ac := &AtomicConfig{}
    ac.value.Store(config)
    return ac
}

func (ac *AtomicConfig) Load() Config {
    return ac.value.Load().(Config)
}

func (ac *AtomicConfig) Store(config Config) {
    ac.value.Store(config)
}

func atomicValueExample() {
    fmt.Println("\n=== 原子值操作示例 ===")

    config := NewAtomicConfig(Config{
        MaxConnections: 100,
        Timeout:        5 * time.Second,
        Debug:          false,
    })

    var wg sync.WaitGroup

    // 读取配置的goroutine
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()

            for j := 0; j < 3; j++ {
                cfg := config.Load()
                fmt.Printf("Reader %d: MaxConn=%d, Debug=%t\n",
                    id, cfg.MaxConnections, cfg.Debug)
                time.Sleep(100 * time.Millisecond)
            }
        }(i)
    }

    // 更新配置的goroutine
    wg.Add(1)
    go func() {
        defer wg.Done()

        time.Sleep(200 * time.Millisecond)
        newConfig := Config{
            MaxConnections: 200,
            Timeout:        10 * time.Second,
            Debug:          true,
        }
        config.Store(newConfig)
        fmt.Println("配置已更新")
    }()

    wg.Wait()
}

// 5. 性能计数器
type PerformanceCounter struct {
    requests    int64
    errors      int64
    totalTime   int64
    maxTime     int64
    minTime     int64
}

func (pc *PerformanceCounter) RecordRequest(duration time.Duration) {
    atomic.AddInt64(&pc.requests, 1)
    atomic.AddInt64(&pc.totalTime, int64(duration))
    
    // 更新最大时间
    for {
        current := atomic.LoadInt64(&pc.maxTime)
        if int64(duration) <= current {
            break
        }
        if atomic.CompareAndSwapInt64(&pc.maxTime, current, int64(duration)) {
            break
        }
    }
    
    // 更新最小时间
    for {
        current := atomic.LoadInt64(&pc.minTime)
        if current != 0 && int64(duration) >= current {
            break
        }
        if atomic.CompareAndSwapInt64(&pc.minTime, current, int64(duration)) {
            break
        }
    }
}

func (pc *PerformanceCounter) RecordError() {
    atomic.AddInt64(&pc.errors, 1)
}

func (pc *PerformanceCounter) Stats() (requests, errors int64, avgTime, maxTime, minTime time.Duration) {
    requests = atomic.LoadInt64(&pc.requests)
    errors = atomic.LoadInt64(&pc.errors)
    totalTime := atomic.LoadInt64(&pc.totalTime)
    maxTime = time.Duration(atomic.LoadInt64(&pc.maxTime))
    minTime = time.Duration(atomic.LoadInt64(&pc.minTime))
    
    if requests > 0 {
        avgTime = time.Duration(totalTime / requests)
    }
    
    return
}

func performanceCounterExample() {
    fmt.Println("\n=== 性能计数器示例 ===")
    
    counter := &PerformanceCounter{}
    var wg sync.WaitGroup
    
    // 模拟并发请求
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            
            // 模拟请求处理时间
            start := time.Now()
            time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
            duration := time.Since(start)
            
            // 模拟10%的错误率
            if rand.Float32() < 0.1 {
                counter.RecordError()
            } else {
                counter.RecordRequest(duration)
            }
        }()
    }
    
    wg.Wait()
    
    requests, errors, avgTime, maxTime, minTime := counter.Stats()
    fmt.Printf("请求总数: %d\n", requests)
    fmt.Printf("错误总数: %d\n", errors)
    fmt.Printf("平均响应时间: %v\n", avgTime)
    fmt.Printf("最大响应时间: %v\n", maxTime)
    fmt.Printf("最小响应时间: %v\n", minTime)
}

func main() {
    producerConsumerExample()
    atomicExample()
    atomicValueExample()
    performanceCounterExample()
}