# 共享状态：锁、原子计数与局部归并

学习目标：安全地累计多任务计数，并判断什么时候根本不必共享可变状态。

前置知识：循环、可变变量、结果或异常；本篇会说明互斥锁和原子操作的最低必要语义。选择 Rust 与 Java，对照所有权约束下的作用域线程和 JVM 执行器；只使用标准库，不引入并行集合或第三方原子封装。

# 问题与边界

把两个分片分别交给工作者，每个分片处理 0～1000 个元素，处理一个元素就加一。输入长度至多为 2 是本教学入口的前置条件；负数和超过 1000 的分片在工作者中报告 invalid count。空输入结果为 0。

同一次工作分别记录三个计数：锁保护的共享计数、原子共享计数、工作者局部计数的最终合计。成功时三者都等于总元素数，[1000, 1000] 得到 2000,2000,2000。任意工作者失败时不返回计数快照，但必须先收束所有工作者。失败前已经更新的共享状态不回滚，因此它们只保存在本次调用内部，不泄露给外部调用方。

这三个计数同时出现仅用于比较机制，生产中应选一种，不要对每个元素重复做三份工作。这里不涉及进程间共享、持久化计数、多个字段组成的事务或锁自由算法。

# 关键差异

Rust 用作用域线程借用计数器，Java 用固定线程池；两者都先等待任务，再读取最终值。Rust 的 Relaxed 原子计数不负责发布其他数据，线程收束另行建立生命周期边界；Java 本例的 AtomicInteger 有更强的内存语义，但不能因此推导多个计数总能在中间时刻相等。

# 各语言实现

# Rust

thread::scope 允许借用非静态数据，并保证作用域退出前线程结束；Mutex 通过守卫生命周期解锁。

Rustuse std::sync::{atomic::{AtomicUsize, Ordering}, Mutex};
use std::thread;

fn count(parts: &[i32]) -> Result<(usize, usize, usize), &'static str> {
    let locked = Mutex::new(0usize);
    let atomic = AtomicUsize::new(0);
    let local_total = thread::scope(|scope| {
        let mut handles = Vec::with_capacity(parts.len());
        for &amount in parts {
            let locked = &locked;
            let atomic = &atomic;
            handles.push(scope.spawn(move || -> Result<usize, &'static str> {
                if !(0..=1000).contains(&amount) {
                    return Err("invalid count");
                }
                let mut local = 0;
                for _ in 0..amount {
                    *locked.lock().map_err(|_| "poisoned lock")? += 1;
                    atomic.fetch_add(1, Ordering::Relaxed);
                    local += 1;
                }
                Ok(local)
            }));
        }
        let mut total = 0;
        let mut failure = None;
        for handle in handles {
            match handle.join().map_err(|_| "worker panicked").and_then(|r| r) {
                Ok(value) => total += value,
                Err(error) => { if failure.is_none() { failure = Some(error); } }
            }
        }
        match failure {
            Some(error) => Err(error),
            None => Ok(total),
        }
    })?;
    let locked_total = *locked.lock().map_err(|_| "poisoned lock")?;
    Ok((locked_total, atomic.load(Ordering::Relaxed), local_total))
}

fn main() -> Result<(), &'static str> {
    assert_eq!(count(&[])?, (0, 0, 0));
    assert_eq!(count(&[0, 1])?, (1, 1, 1));
    let totals = count(&[1000, 1000])?;
    assert_eq!(totals, (2000, 2000, 2000));
    println!("{},{},{}", totals.0, totals.1, totals.2);
    for input in [[-1, 1000], [1001, 0]] {
        assert_eq!(count(&input), Err("invalid count"));
    }
    println!("invalid count");
    Ok(())
}

作用域借用避免为了这个短生命周期再加 Arc 引用计数。Mutex 的毒化失败与工作者 panic 被显式传播，不悄悄读取可能失去业务不变量的状态。原子顺序的依据见 Ordering::Relaxed：这里仅需计数原子性，不通过该计数告知其他内存已经初始化。

# Java

运行时使用 -ea 启用断言。JDK 21 的 ExecutorService.close 会关闭执行器并等待终止，适合用 try-with-resources 收束线程。

Javaimport java.util.ArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.Arrays;

public class Main {
    static int[] count(int[] parts) throws InterruptedException, ExecutionException {
        Object lock = new Object();
        int[] locked = {0};
        AtomicInteger atomic = new AtomicInteger();
        int total = 0;
        ExecutionException failure = null;
        try (var executor = Executors.newFixedThreadPool(2)) {
            var futures = new ArrayList<Future<Integer>>();
            for (int amount : parts) {
                futures.add(executor.submit(() -> {
                    if (amount < 0 || amount > 1000) {
                        throw new IllegalArgumentException("invalid count");
                    }
                    int local = 0;
                    for (int i = 0; i < amount; i++) {
                        synchronized (lock) { locked[0]++; }
                        atomic.incrementAndGet();
                        local++;
                    }
                    return local;
                }));
            }
            for (Future<Integer> future : futures) {
                try {
                    total += future.get();
                } catch (ExecutionException error) {
                    if (failure == null) failure = error;
                }
            }
        }
        if (failure != null) throw failure;
        return new int[] {locked[0], atomic.get(), total};
    }

    public static void main(String[] args) throws Exception {
        assert Arrays.equals(count(new int[] {}), new int[] {0, 0, 0});
        assert Arrays.equals(count(new int[] {0, 1}), new int[] {1, 1, 1});
        int[] totals = count(new int[] {1000, 1000});
        assert Arrays.equals(totals, new int[] {2000, 2000, 2000});
        System.out.println(totals[0] + "," + totals[1] + "," + totals[2]);
        for (int[] input : new int[][] {{-1, 1000}, {1001, 0}}) {
            try {
                count(input);
                throw new AssertionError("failure was lost");
            } catch (ExecutionException error) {
                assert error.getCause() instanceof IllegalArgumentException;
                assert error.getCause().getMessage().equals("invalid count");
            }
        }
        System.out.println("invalid count");
    }
}

Future.get 把任务失败包装为 ExecutionException，调用者检查真正原因；主线程被中断则向外传播 InterruptedException，资源块仍负责关闭线程池。普通 int[] 不是线程安全容器：安全性来自所有工作者都使用同一个锁，并且最终读取发生在工作结束后。单变量原子操作见 AtomicInteger.incrementAndGet。

# 迁移误区与适用限制

• counter = counter + 1 看似一行，却是读、计算、写；多个线程可能丢失更新。用 volatile 修饰也不会把复合操作自动变原子。
• 以为两个原子字段就能维护“已完成数等于成功加失败”：每个字段各自原子，不代表组合快照一致；此时需要一把锁或统一消息归并。
• 把锁跨越网络等待持有：其他工作者全部被阻挡。先在局部计算，再用短临界区更新，或直接返回局部结果。
• 默认原子一定更快：争用、缓存一致性与更新频率决定成本。本例不给性能排序，局部归并通常值得优先考虑，因为它从设计上减少共享写入。

# 运行与验证

两个完整程序预期输出均为：

纯文本2000,2000,2000
invalid count

断言覆盖空分片、零与一、上界 1000、越界 1001 和负数；任务失败后仍结束全部线程。没有计时或性能断言。版本见示例基线，实际执行状态见验证记录。

# 相关阅读

• 任务间传递数据：生产者与消费者通道：把可变状态留在唯一消费者中。
• CPU 并行：将局部结果归并用于计算任务。