java - 为可索引并发跳过列表实现交换方法

Question

我正在基于 Java 的 ConcurrentSkipListMap 实现并发跳跃列表映射，不同之处在于我希望列表允许重复，而且我还希望列表为 indexable (因此找到列表的第 N 个元素需要 O(lg(n)) 时间，而不是标准跳跃列表的 O(n) 时间)。这些修改不会造成问题。

此外，跳过列表的键是可变的。例如，如果列表元素是整数{0, 4, 7}，那么可以将中间元素的键更改为[0, 7]中的任何值，而不会提示更改列表结构；如果键更改为 (-inf, -1] 或 [8, +inf)，则删除并重新添加该元素以维持列表顺序。我没有将其实现为先删除后跟 O(lg(n)) 插入，而是将其实现为先删除后跟线性遍历后跟 O(1) 插入(预期运行时间为 O(1) - 99节点将与相邻节点交换的时间百分比)。

插入一个全新的节点是很少见的(在启动后)，而删除一个节点(没有立即重新添加它)永远不会发生；几乎所有的操作都是 elementAt(i) 以检索第 i 个索引处的元素，或者在修改键后交换节点的操作。

我遇到的问题是如何实现键修改类。从概念上讲，我想做类似的事情

public class Node implements Runnable {
    private int key;
    private Node prev, next;
    private BlockingQueue<Integer> queue;

    public void update(int i) {
        queue.offer(i);
    }

    public void run() {
        while(true) {
            int temp = queue.take();
            temp += key;
            if(prev.getKey() > temp) {
                // remove node, update key to temp, perform backward linear traversal, and insert
            } else if(next.getKey() < temp) {
                // remove node, update key to temp, perform forward linear traveral, and insert
            } else {
                key = temp; // node doesn't change position
            }
        }
    }
}

(从 run 调用的 insert 子方法使用 CAS 来处理两个节点试图同时插入同一位置的问题(类似于ConcurrentSkipListMap 处理冲突插入)——从概念上讲，这与第一个节点锁定与插入点相邻的节点相同，只是在没有冲突的情况下减少了开销。)

这样我可以确保列表始终按顺序排列(如果 key 更新有点延迟也没关系，因为我可以确定更新最终会发生；但是，如果列表变得无序然后事情可能会变得困惑)。问题在于，以这种方式实现列表会生成大量线程，每个 Node(列表中有数千个节点)一个线程 - 大多数线程会在任何给定时间点阻塞，但我担心数千个阻塞线程仍会导致过高的开销。

另一种选择是使 update 方法同步并从 Node 中删除 Runnable 接口(interface)，这样就不用两个线程排队更新在 Node 中，后者负责在其单独的线程上处理这些更新，这两个线程将轮流执行 Node#update 方法。问题是这可能会造成瓶颈；如果八个不同的线程都决定同时更新同一个节点，那么队列实现会很好地扩展，但是同步实现会阻塞八个线程中的七个(然后会阻塞六个线程，然后阻塞五个，等等)。

所以我的问题是，除了减少线程数之外，我将如何实现类似于队列实现的东西，否则我将如何实现类似于同步实现的东西，除非没有潜在的瓶颈问题。

Answer 1

我想我可以用 ThreadPoolExecutor 来解决这个问题，比如

public class Node {
    private int key;
    private Node prev, next;
    private ConcurrentLinkedQueue<Integer> queue;
    private AtomicBoolean lock = new AtomicBoolean(false);
    private ThreadPoolExecutor executor;
    private UpdateNode updater = new UpdateNode();

    public void update(int i) {
        queue.offer(i);
        if(lock.compareAndSet(false, true)) {
            executor.execute(updater);
        }
    }

    private class UpdateNode implements Runnable {
        public void run() {
            do {
                try {
                    int temp = key;
                    while(!queue.isEmpty()) {
                        temp += queue.poll();
                    }
                    if(prev.getKey() > temp) {
                        // remove node, update key to temp, perform backward linear traversal, and insert
                    } else if(next.getKey() < temp) {
                        // remove node, update key to temp, perform forward linear traveral, and insert
                    } else {
                        key = temp; // node doesn't change position
                    }
                } finally {
                    lock.set(false);
                }
            } while (!queue.isEmpty() && lock.compareAndSet(false, true));
        }
    }
}

通过这种方式，我可以利用队列方法的优势，而不会阻塞上千个线程 - 我每次需要更新节点时都会执行一个 UpdateNode(除非已经有一个 UpdateNode 在该 Node 上执行，因此 AtomicBoolean 充当锁)，并依赖 ThreadPoolExecutor 使其成本低廉创建数千个 runnable。