Java多线程环境下如何实现存储类的细粒度线程安全？

问题分析 #

你希望用基于值对象的细粒度锁替代粗粒度的对象锁，避免不同值的add/delete/search操作互斥，提升并发性能。但这里存在一个容易忽略的关键漏洞：BST的操作会遍历从根节点到目标节点的路径，不同值的操作可能会修改路径上的同一个父节点。比如添加8时需要修改9的左指针，而删除9时需要修改9的父节点指针，这两个操作同时执行会导致BST结构损坏，仅对目标值加锁无法解决这个问题。

下面提供三种可行方案，覆盖不同场景需求：

方案一：基于值对象的锁（适配你的初始设想） #

如果业务场景可以接受BST结构的并发风险，或者仅需保证同一值的操作互斥，可采用此方案：

代码实现 #

import java.util.concurrent.ConcurrentHashMap;

class MyStorage<T extends Comparable<T>> {
    private BSTreeNode rootNode;
    // 线程安全的映射表，存储每个值对应的锁对象
    private final ConcurrentHashMap<T, Object> valueLocks = new ConcurrentHashMap<>();

    private class BSTreeNode {
        T value;
        int occurrence;
        BSTreeNode left;
        BSTreeNode right;

        BSTreeNode(T value) {
            this.value = value;
            this.occurrence = 1;
        }
    }

    // 原子化获取或创建值对应的锁
    private Object getValueLock(T value) {
        Object lock = valueLocks.putIfAbsent(value, new Object());
        return lock == null ? valueLocks.get(value) : lock;
    }

    public boolean add(T value) {
        Object lock = getValueLock(value);
        synchronized (lock) {
            if (rootNode == null) {
                rootNode = new BSTreeNode(value);
                return true;
            }
            BSTreeNode current = rootNode;
            BSTreeNode parent = null;
            while (current != null) {
                parent = current;
                int cmp = value.compareTo(current.value);
                if (cmp == 0) {
                    current.occurrence++;
                    return true;
                }
                current = cmp < 0 ? current.left : current.right;
            }
            // 添加新节点到对应位置
            if (value.compareTo(parent.value) < 0) {
                parent.left = new BSTreeNode(value);
            } else {
                parent.right = new BSTreeNode(value);
            }
            return true;
        }
    }

    public boolean delete(T value) {
        Object lock = getValueLock(value);
        synchronized (lock) {
            BSTreeNode parent = null;
            BSTreeNode current = rootNode;
            while (current != null && value.compareTo(current.value) != 0) {
                parent = current;
                current = value.compareTo(current.value) < 0 ? current.left : current.right;
            }
            if (current == null) return false;
            // 处理重复值
            if (current.occurrence > 1) {
                current.occurrence--;
                return true;
            }
            // 删除无后代的节点（复杂子节点迁移逻辑省略）
            if (current.left == null && current.right == null) {
                if (parent == null) rootNode = null;
                else if (parent.left == current) parent.left = null;
                else parent.right = null;
            }
            return true;
        }
    }

    public boolean search(T value) {
        Object lock = getValueLock(value);
        synchronized (lock) {
            BSTreeNode current = rootNode;
            while (current != null) {
                int cmp = value.compareTo(current.value);
                if (cmp == 0) return current.occurrence > 0;
                current = cmp < 0 ? current.left : current.right;
            }
            return false;
        }
    }
}

关键注意事项 #

• 值对象必须正确实现equals和hashCode：ConcurrentHashMap依赖这两个方法区分不同值，错误实现会导致锁混乱，破坏线程安全。
• 内存占用问题：valueLocks会保留所有曾添加过的值的锁，即使值被删除也不会自动释放。若内存压力大，可改用WeakHashMap配合同步块（需自行处理线程安全）。
• BST结构风险：不同值的操作可能修改同一父节点，导致结构不一致，此方案仅适合对结构一致性要求较低的场景。

方案二：节点级细粒度锁（保证BST结构安全） #

如果需要严格保证BST的并发结构安全，可采用节点级锁：遍历路径时按顺序锁定节点，避免死锁，同时保证细粒度并发。

代码示例（简化版） #

class MyStorage<T extends Comparable<T>> {
    private BSTreeNode rootNode;

    private class BSTreeNode {
        T value;
        int occurrence;
        BSTreeNode left;
        BSTreeNode right;
        // 每个节点自带锁对象
        final Object lock = new Object();

        BSTreeNode(T value) {
            this.value = value;
            this.occurrence = 1;
        }
    }

    public boolean add(T value) {
        // 处理根节点为空的情况
        if (rootNode == null) {
            synchronized (this) {
                if (rootNode == null) {
                    rootNode = new BSTreeNode(value);
                    return true;
                }
            }
        }
        BSTreeNode current = rootNode;
        synchronized (current.lock) {
            while (true) {
                int cmp = value.compareTo(current.value);
                if (cmp == 0) {
                    current.occurrence++;
                    return true;
                }
                BSTreeNode next = cmp < 0 ? current.left : current.right;
                if (next == null) {
                    // 添加新节点
                    BSTreeNode newNode = new BSTreeNode(value);
                    if (cmp < 0) current.left = newNode;
                    else current.right = newNode;
                    return true;
                }
                // 先锁子节点再释放父节点，避免死锁
                synchronized (next.lock) {
                    current.lock.notifyAll();
                    current = next;
                }
            }
        }
    }

    // delete、search方法遵循相同的节点锁顺序逻辑，此处省略
}

方案三：使用Java内置并发有序集合（最简方案） #

如果不需要手动实现BST，直接用ConcurrentSkipListMap是最优选择——它是线程安全的有序集合，基于跳表实现，并发性能优于手动实现的BST。

代码示例 #

import java.util.concurrent.ConcurrentSkipListMap;

class MyStorage<T extends Comparable<T>> {
    private final ConcurrentSkipListMap<T, Integer> map = new ConcurrentSkipListMap<>();

    public boolean add(T value) {
        map.merge(value, 1, Integer::sum);
        return true;
    }

    public boolean delete(T value) {
        return map.computeIfPresent(value, (k, v) -> v > 1 ? v - 1 : null) != null;
    }

    public boolean search(T value) {
        return map.containsKey(value);
    }
}

内容的提问来源于stack exchange，提问作者kesarling