概念定义
两个访问指向同一位置且其一为写即构成冲突；线程之间没有 happens-before 顺序时就是数据竞争。后果从读到陈旧值、32 位平台上 64 位撕裂写，到严格语言里整程序未定义行为不等。
工作机制
编译器把值缓存进寄存器、按单线程推理合并重排访问、删除它认为冗余的写；CPU 再叠加存储缓冲与宽松可见性。检测器（ThreadSanitizer 等）靠影子跟踪访问顺序标记“无序冲突对”——能抓住还没造成损坏的竞争。
使用场景
一切被回调、定时器、多请求触碰的共享可变状态。让共享显式——传所有权、走 channel、或用一把锁——并在运行时支持时把竞争检测器开进 CI。
常见误区
x86 强序上的“侥幸按序”会掩盖竞争；用原子类型修错变量，竞争仍在孪生字段上。有的修复引入死锁：同步了 A 字段却在 B 字段上等待。数据竞争清零后，check-then-act 这类高层竞态条件依然健在。单核抢占同样有竞争——上下文切换与编译器重排都不挑核数。
代码示例
package main
import (
"context"
"fmt"
"sync"
"time"
)
type WeightedSemaphore struct {
mu sync.Mutex
cond *sync.Cond
capacity int64
used int64
}
func NewWeightedSemaphore(capacity int64) (*WeightedSemaphore, error) {
if capacity < 1 {
return nil, fmt.Errorf("capacity must be positive")
}
sem := &WeightedSemaphore{capacity: capacity}
sem.cond = sync.NewCond(&sem.mu)
return sem, nil
}
func (s *WeightedSemaphore) Acquire(ctx context.Context, weight int64) error {
if weight < 1 || weight > s.capacity {
return fmt.Errorf("weight %d outside 1..%d", weight, s.capacity)
}
done := make(chan struct{})
go func() {
select {
case <-ctx.Done():
s.mu.Lock()
s.cond.Broadcast()
s.mu.Unlock()
case <-done:
}
}()
defer close(done)
s.mu.Lock()
defer s.mu.Unlock()
for s.used+weight > s.capacity {
if err := ctx.Err(); err != nil {
return err
}
s.cond.Wait()
}
if err := ctx.Err(); err != nil {
return err
}
s.used += weight
return nil
}
func (s *WeightedSemaphore) Release(weight int64) {
s.mu.Lock()
defer s.mu.Unlock()
if weight < 1 || weight > s.used {
panic(fmt.Sprintf("release %d with %d in use", weight, s.used))
}
s.used -= weight
s.cond.Broadcast()
}
func main() {
sem, err := NewWeightedSemaphore(3)
if err != nil {
fmt.Println(err)
return
}
jobs := []struct {
name string
weight int64
}{
{"small", 1},
{"medium", 2},
{"large", 3},
{"small-2", 1},
}
var wg sync.WaitGroup
for _, job := range jobs {
job := job
wg.Add(1)
go func() {
defer wg.Done()
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err := sem.Acquire(ctx, job.weight); err != nil {
fmt.Printf("%s: %v\n", job.name, err)
return
}
time.Sleep(20 * time.Millisecond)
sem.Release(job.weight)
fmt.Printf("%s finished (weight %d)\n", job.name, job.weight)
}()
}
wg.Wait()
fmt.Println("all jobs done")
}package main
import (
"context"
"fmt"
"sync"
"time"
)
type Job struct {
ID int
Value int
}
type Result struct {
JobID int
Out int
Err error
}
func worker(ctx context.Context, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
defer wg.Done()
for {
select {
case <-ctx.Done():
return
case job, ok := <-jobs:
if !ok {
return
}
if job.Value < 0 {
results <- Result{JobID: job.ID, Err: fmt.Errorf("job %d: negative value %d", job.ID, job.Value)}
continue
}
select {
case <-ctx.Done():
return
case results <- Result{JobID: job.ID, Out: job.Value * job.Value}:
}
}
}
}
func run(ctx context.Context, jobs []Job, workers int) ([]Result, error) {
if workers < 1 {
return nil, fmt.Errorf("workers must be at least 1, got %d", workers)
}
if len(jobs) == 0 {
return nil, nil
}
jobCh := make(chan Job)
resCh := make(chan Result, len(jobs))
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
go worker(ctx, jobCh, resCh, &wg)
}
go func() {
defer close(jobCh)
for _, job := range jobs {
select {
case <-ctx.Done():
return
case jobCh <- job:
}
}
}()
go func() {
wg.Wait()
close(resCh)
}()
results := make([]Result, 0, len(jobs))
for result := range resCh {
results = append(results, result)
}
if err := ctx.Err(); err != nil {
return results, err
}
return results, nil
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
jobs := []Job{
{ID: 1, Value: 4},
{ID: 2, Value: 9},
{ID: 3, Value: -1},
{ID: 4, Value: 16},
{ID: 5, Value: 25},
{ID: 6, Value: 36},
}
results, err := run(ctx, jobs, 3)
if err != nil {
fmt.Println("stopped early:", err)
}
failed := 0
for _, result := range results {
if result.Err != nil {
failed++
fmt.Println("error:", result.Err)
continue
}
fmt.Printf("job %d -> %d\n", result.JobID, result.Out)
}
fmt.Printf("%d results, %d failed\n", len(results), failed)
}