githubGitHub
发布时间

go 锁事件示例

互斥锁 vs 读写锁性能对比

package main

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

type DataStore struct {
	data map[int]string
	mu   sync.Mutex
	rwmu sync.RWMutex
}

func NewDataStore() *DataStore {
	return &DataStore{
		data: make(map[int]string),
	}
}

// 使用互斥锁的读写
func (ds *DataStore) ReadWithMutex(key int) (string, bool) {
	ds.mu.Lock()
	defer ds.mu.Unlock()

	time.Sleep(1 * time.Millisecond) // 模拟读耗时
	value, exists := ds.data[key]
	return value, exists
}

func (ds *DataStore) WriteWithMutex(key int, value string) {
	ds.mu.Lock()
	defer ds.mu.Unlock()

	time.Sleep(5 * time.Millisecond) // 模拟写耗时
	ds.data[key] = value
}

// 使用读写锁的读写
func (ds *DataStore) ReadWithRWMutex(key int) (string, bool) {
	ds.rwmu.RLock()
	defer ds.rwmu.RUnlock()

	time.Sleep(1 * time.Millisecond) // 模拟读耗时
	value, exists := ds.data[key]
	return value, exists
}

func (ds *DataStore) WriteWithRWMutex(key int, value string) {
	ds.rwmu.Lock()
	defer ds.rwmu.Unlock()

	time.Sleep(5 * time.Millisecond) // 模拟写耗时
	ds.data[key] = value
}

func benchmark(name string, readFunc func(), writeFunc func(), readers, writers int) {
	var wg sync.WaitGroup
	start := time.Now()

	// 启动读协程
	for i := 0; i < readers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := 0; j < 100; j++ {
				readFunc()
			}
		}()
	}

	// 启动写协程
	for i := 0; i < writers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := 0; j < 20; j++ {
				writeFunc()
			}
		}()
	}

	wg.Wait()
	elapsed := time.Since(start)

	fmt.Printf("%s:\n", name)
	fmt.Printf("  读者数: %d, 写着数: %d\n", readers, writers)
	fmt.Printf("  耗时: %v\n", elapsed)
	fmt.Printf("  总操作数: %d\n", readers*100+writers*20)
	fmt.Println()
}

func main() {
	ds := NewDataStore()

	fmt.Println("性能对比测试（读多写少场景）:")
	fmt.Println(strings.Repeat("=", 50))

	// 测试1：大量读，少量写
	benchmark("互斥锁",
		func() { ds.ReadWithMutex(1) },
		func() { ds.WriteWithMutex(1, "value") },
		100, 5)

	benchmark("读写锁",
		func() { ds.ReadWithRWMutex(1) },
		func() { ds.WriteWithRWMutex(1, "value") },
		100, 5)

	fmt.Println("性能对比测试（读写平衡场景）:")
	fmt.Println(strings.Repeat("=", 50))

	// 测试2：读写平衡
	benchmark("互斥锁",
		func() { ds.ReadWithMutex(2) },
		func() { ds.WriteWithMutex(2, "value") },
		50, 50)

	benchmark("读写锁",
		func() { ds.ReadWithRWMutex(2) },
		func() { ds.WriteWithRWMutex(2, "value") },
		50, 50)

	fmt.Println("性能对比测试（写多读少场景）:")
	fmt.Println(strings.Repeat("=", 50))

	// 测试3：大量写，少量读
	benchmark("互斥锁",
		func() { ds.ReadWithMutex(3) },
		func() { ds.WriteWithMutex(3, "value") },
		10, 90)

	benchmark("读写锁",
		func() { ds.ReadWithRWMutex(3) },
		func() { ds.WriteWithRWMutex(3, "value") },
		10, 90)
}

缓存系统

package main

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

type Cache struct {
	data     map[string]interface{}
	rwmu     sync.RWMutex
	stats    CacheStats
	statsMu  sync.Mutex // 单独的保护统计数据的锁
}

type CacheStats struct {
	hits   int
	misses int
	evicts int
}

func NewCache() *Cache {
	return &Cache{
		data: make(map[string]interface{}),
		stats: CacheStats{
			hits:   0,
			misses: 0,
			evicts: 0,
		},
	}
}

// Get 读取缓存（读锁）
func (c *Cache) Get(key string) (interface{}, bool) {
	c.rwmu.RLock()
	value, exists := c.data[key]
	c.rwmu.RUnlock()
	
	// 更新统计信息（需要互斥锁）
	c.statsMu.Lock()
	if exists {
		c.stats.hits++
		fmt.Printf("缓存命中: %s\n", key)
	} else {
		c.stats.misses++
		fmt.Printf("缓存未命中: %s\n", key)
	}
	c.statsMu.Unlock()
	
	return value, exists
}

// Set 设置缓存（写锁）
func (c *Cache) Set(key string, value interface{}) {
	c.rwmu.Lock()
	defer c.rwmu.Unlock()
	
	// 模拟设置耗时
	time.Sleep(15 * time.Millisecond)
	
	c.data[key] = value
	fmt.Printf("设置缓存: %s = %v\n", key, value)
}

// Delete 删除缓存（写锁）
func (c *Cache) Delete(key string) bool {
	c.rwmu.Lock()
	defer c.rwmu.Unlock()
	
	_, exists := c.data[key]
	if exists {
		delete(c.data, key)
		c.statsMu.Lock()
		c.stats.evicts++
		c.statsMu.Unlock()
		fmt.Printf("删除缓存: %s\n", key)
	}
	return exists
}

// GetStats 获取统计信息
func (c *Cache) GetStats() CacheStats {
	c.statsMu.Lock()
	defer c.statsMu.Unlock()
	return c.stats
}

// Clear 清空缓存
func (c *Cache) Clear() {
	c.rwmu.Lock()
	defer c.rwmu.Unlock()
	
	c.data = make(map[string]interface{})
	c.statsMu.Lock()
	c.stats.evicts = 0 // 重置驱逐计数
	c.statsMu.Unlock()
	fmt.Println("缓存已清空")
}

func main() {
	cache := NewCache()
	
	var wg sync.WaitGroup
	
	// 并发读取（大量读操作）
	for i := 1; i <= 15; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for j := 0; j < 10; j++ {
				key := fmt.Sprintf("key%d", (id+j)%10)
				cache.Get(key)
				time.Sleep(time.Millisecond * 5)
			}
		}(i)
	}
	
	// 并发写入（少量写操作）
	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for j := 0; j < 3; j++ {
				key := fmt.Sprintf("key%d", id*10+j)
				cache.Set(key, fmt.Sprintf("value%d", id*100+j))
				time.Sleep(time.Millisecond * 30)
			}
		}(i)
	}
	
	// 定时清理
	wg.Add(1)
	go func() {
		defer wg.Done()
		time.Sleep(300 * time.Millisecond)
		cache.Clear()
	}()
	
	wg.Wait()
	
	stats := cache.GetStats()
	fmt.Printf("\n缓存统计:\n")
	fmt.Printf("命中次数: %d\n", stats.hits)
	fmt.Printf("未命中次数: %d\n", stats.misses)
	fmt.Printf("驱逐次数: %d\n", stats.evicts)
}

死锁预防与检测

package main

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

// 错误示例：可能导致死锁
func deadlockExample() {
	var mu1, mu2 sync.Mutex
	
	go func() {
		mu1.Lock()
		fmt.Println("goroutine1 获取 mu1")
		time.Sleep(100 * time.Millisecond)
		
		mu2.Lock() // 这里会等待 mu2
		fmt.Println("goroutine1 获取 mu2")
		
		mu2.Unlock()
		mu1.Unlock()
	}()
	
	go func() {
		mu2.Lock()
		fmt.Println("goroutine2 获取 mu2")
		time.Sleep(100 * time.Millisecond)
		
		mu1.Lock() // 这里会等待 mu1
		fmt.Println("goroutine2 获取 mu1")
		
		mu1.Unlock()
		mu2.Unlock()
	}()
	
	time.Sleep(2 * time.Second)
	fmt.Println("程序结束（可能已经死锁）")
}

// 正确示例：使用固定顺序获取锁
func correctExample() {
	var mu1, mu2 sync.Mutex
	
	go func() {
		// 总是先获取 mu1，再获取 mu2
		mu1.Lock()
		fmt.Println("goroutine1 获取 mu1")
		time.Sleep(100 * time.Millisecond)
		
		mu2.Lock()
		fmt.Println("goroutine1 获取 mu2")
		
		mu2.Unlock()
		mu1.Unlock()
	}()
	
	go func() {
		// 同样先获取 mu1，再获取 mu2
		mu1.Lock()
		fmt.Println("goroutine2 获取 mu1")
		time.Sleep(100 * time.Millisecond)
		
		mu2.Lock()
		fmt.Println("goroutine2 获取 mu2")
		
		mu2.Unlock()
		mu1.Unlock()
	}()
	
	time.Sleep(2 * time.Second)
	fmt.Println("程序正常结束")
}

// 使用 defer 确保解锁
func deferExample() {
	var mu sync.Mutex
	
	mu.Lock()
	defer mu.Unlock() // 确保函数退出时解锁
	
	// 执行一些操作
	fmt.Println("临界区操作")
	// 即使这里 panic，锁也会被释放
	// panic("发生错误")
}

// 使用 TryLock 避免死锁
func tryLockExample() {
	var mu sync.Mutex
	
	// 尝试获取锁
	if mu.TryLock() {
		defer mu.Unlock()
		fmt.Println("成功获取锁")
	} else {
		fmt.Println("锁已被占用，执行其他操作")
		// 执行不需要锁的操作
	}
}

func main() {
	fmt.Println("=== 死锁示例 ===")
	go deadlockExample()
	time.Sleep(3 * time.Second)
	
	fmt.Println("\n=== 正确示例 ===")
	correctExample()
	
	fmt.Println("\n=== defer 示例 ===")
	deferExample()
	
	fmt.Println("\n=== TryLock 示例 ===")
	tryLockExample()
}

线程安全的队列

package main

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

type ThreadSafeQueue struct {
	items []interface{}
	mu    sync.RWMutex
	cond  *sync.Cond // 条件变量
}

func NewThreadSafeQueue() *ThreadSafeQueue {
	q := &ThreadSafeQueue{
		items: make([]interface{}, 0),
	}
	q.cond = sync.NewCond(&q.mu)
	return q
}

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

// Dequeue 出队（阻塞）
func (q *ThreadSafeQueue) Dequeue() interface{} {
	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("出队: %v, 队列长度: %d\n", item, len(q.items))
	
	return item
}

// TryDequeue 尝试出队（非阻塞）
func (q *ThreadSafeQueue) TryDequeue() (interface{}, bool) {
	q.mu.Lock()
	defer q.mu.Unlock()
	
	if len(q.items) == 0 {
		return nil, false
	}
	
	item := q.items[0]
	q.items = q.items[1:]
	fmt.Printf("尝试出队成功: %v\n", item)
	
	return item, true
}

// Peek 查看队首元素
func (q *ThreadSafeQueue) Peek() (interface{}, bool) {
	q.mu.RLock()
	defer q.mu.RUnlock()
	
	if len(q.items) == 0 {
		return nil, false
	}
	
	return q.items[0], true
}

// Size 获取队列大小
func (q *ThreadSafeQueue) Size() int {
	q.mu.RLock()
	defer q.mu.RUnlock()
	return len(q.items)
}

func main() {
	queue := NewThreadSafeQueue()
	
	var wg sync.WaitGroup
	
	// 生产者
	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for j := 1; j <= 5; j++ {
				item := fmt.Sprintf("生产者%d-项目%d", id, j)
				queue.Enqueue(item)
				time.Sleep(time.Millisecond * 50 * time.Duration(id))
			}
		}(i)
	}
	
	// 消费者（阻塞式）
	for i := 1; i <= 2; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for j := 1; j <= 8; j++ {
				item := queue.Dequeue()
				fmt.Printf("消费者%d 消费: %v\n", id, item)
				time.Sleep(time.Millisecond * 80)
			}
		}(i)
	}
	
	// 消费者（非阻塞式）
	wg.Add(1)
	go func() {
		defer wg.Done()
		for i := 0; i < 10; i++ {
			if item, ok := queue.TryDequeue(); ok {
				fmt.Printf("非阻塞消费者 消费: %v\n", item)
			} else {
				fmt.Println("非阻塞消费者: 队列为空")
			}
			time.Sleep(time.Millisecond * 100)
		}
	}()
	
	wg.Wait()
	
	fmt.Printf("\n最终队列大小: %d\n", queue.Size())
}
← 返回博客列表