消息传递与共享内存

概念定义

共享内存问的是“谁、何时可以碰这份数据”；消息传递问的是“谁拥有这份数据”——恰好一个 actor——其他人如何请求对它操作。所有权取代加锁成为正确性机制。

工作机制

actor 或 goroutine 拥有私有状态，经 channel 通信：发送、接收、select 多路。消息不可变或所有权转移防止共享访问。放大到分布式时同一形态变成队列、发布订阅与请求响应，叠加序列化与投递语义。

使用场景

所有权清晰的状态机、变换阶段流水线、以及“不加消息传递就得满地图加锁”的设计。消息保持小、所有权保持脆；真正的热读多数据仍可退回带同步的共享内存。

常见误区

消息传递不是消灭并发缺陷而是搬迁：死锁变成协议等待与邮箱溢出。可变载荷会把共享偷运回来——要么强制转移纪律要么深拷贝。channel 死锁与锁死锁同构但栈迹更糟，务必建模超时；所有操作挤进单一邮箱会把“并行系统”串行化，吞吐吃紧时按键分区 actor。大热状态逐消息传递的拷贝成本可能高于守卫式共享访问。

代码示例

#!/usr/bin/env bash
set -euo pipefail

lock_file=${1:-"${TMPDIR:-/tmp}/devexamples-backup.lock"}
max_wait=${LOCK_WAIT_SECONDS:-5}
work_file=""

cleanup() {
    local status=$?
    if [[ -n "$work_file" && -f "$work_file" ]]; then
        rm -f -- "$work_file"
    fi
    flock -u 9 2>/dev/null || true
    exit "$status"
}
trap cleanup EXIT

exec 9>"$lock_file"

if ! flock -w "$max_wait" 9; then
    printf 'could not acquire %s within %ss\n' "$lock_file" "$max_wait" >&2
    exit 75
fi

printf 'lock acquired by pid %d\n' "$"

work_file="$(mktemp "${TMPDIR:-/tmp}/backup.XXXXXX")"
printf 'job %d started at %s\n' "$" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" > "$work_file"

sleep "${JOB_SECONDS:-1}"
cat -- "$work_file"
printf 'job %d finished; lock released on exit\n' "$"

package main

import (
	"fmt"
	"sync"
)

type Broker struct {
	mu          sync.RWMutex
	subscribers map[int]chan string
	nextID      int
}

func NewBroker() *Broker {
	return &Broker{subscribers: make(map[int]chan string)}
}

func (b *Broker) Subscribe(buffer int) (int, <-chan string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	id := b.nextID
	b.nextID++
	channel := make(chan string, buffer)
	b.subscribers[id] = channel
	return id, channel
}

func (b *Broker) Unsubscribe(id int) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if channel, ok := b.subscribers[id]; ok {
		delete(b.subscribers, id)
		close(channel)
	}
}

func (b *Broker) Publish(event string) int {
	b.mu.RLock()
	defer b.mu.RUnlock()
	delivered := 0
	for _, channel := range b.subscribers {
		select {
		case channel <- event:
			delivered++
		default:
		}
	}
	return delivered
}

func main() {
	broker := NewBroker()
	fastID, fast := broker.Subscribe(4)
	defer broker.Unsubscribe(fastID)
	slowID, slow := broker.Subscribe(1)
	defer broker.Unsubscribe(slowID)

	delivered := []int{broker.Publish("a"), broker.Publish("b"), broker.Publish("c")}
	fmt.Println("delivered:", delivered)
	if delivered[0] != 2 || delivered[1] != 1 || delivered[2] != 1 {
		panic("unexpected delivery counts")
	}

	fastSeen := []string{<-fast, <-fast, <-fast}
	slowSeen := []string{<-slow}
	fmt.Println(fastSeen, slowSeen)
	if slowSeen[0] != "a" {
		panic("slow subscriber should keep only the first event")
	}

	broker.Unsubscribe(slowID)
	if broker.Publish("d") != 1 {
		panic("expected one subscriber after unsubscribe")
	}
	broker.Unsubscribe(fastID)
	if broker.Publish("e") != 0 {
		panic("expected no subscribers")
	}
}