java - 两个线程在两个 LinkedConcurrentQueue 之间以双向方式传输数据会导致一个队列为空，而另一个队列则为 "steals"所有内容

Question

大家!

我编写了一个扩展Thread的类(InAndOut)。该类在构造函数中接收两个 LinkedConcurrentQueue、entrance 和 exit，并且我的 run 方法从 入口到导出。

在我的主方法中，我实例化了两个 LinkedConcurrentQueue、myQueue1 和 myQueue2，每个队列都有一些值。然后，我实例化了两个 InAndOut，一个接收 myQueue1 (入口)和 myQueue2 (导出)，另一个接收 myQueue2 (入口)和 myQueue1(退出)。然后，我调用两个实例的 start 方法。

经过一些迭代后，结果是将所有对象从一个队列转移到另一个队列，换句话说，myQueue1 变空，myQueue2“窃取”所有对象对象。但是，如果我在每次迭代中添加 sleep 调用(大约 100 毫秒)，那么行为就像我预期的那样(两个队列中的元素数量之间达到平衡)。

为什么会发生这种情况以及如何解决？有什么方法可以不在我的 run 方法中使用这个 sleep 调用？我做错了什么吗？

这是我的源代码:

import java.util.concurrent.ConcurrentLinkedQueue;

class InAndOut extends Thread {

    ConcurrentLinkedQueue<String> entrance;
    ConcurrentLinkedQueue<String> exit;
    String name;

    public InAndOut(String name, ConcurrentLinkedQueue<String> entrance, ConcurrentLinkedQueue<String> exit){
        this.entrance = entrance;
        this.exit = exit;
        this.name = name;
    }

    public void run() {
        int it = 0;
        while(it < 3000){
            String value = entrance.poll();

            if(value != null){
                exit.offer(value);
                System.err.println(this.name + " / entrance: " + entrance.size() + " / exit: " + exit.size());
            }

            //THIS IS THE SLEEP CALL THAT MAKES THE CODE WORK AS EXPECTED
            try{
                this.sleep(100);
            } catch (Exception ex){

            }
            it++;
        }
    }
}

public class Main {

    public static void main(String[] args) {

        ConcurrentLinkedQueue<String> myQueue1 = new ConcurrentLinkedQueue<String>();
        ConcurrentLinkedQueue<String> myQueue2 = new ConcurrentLinkedQueue<String>();

        myQueue1.offer("a");
        myQueue1.offer("b");
        myQueue1.offer("c");
        myQueue1.offer("d");
        myQueue1.offer("e");
        myQueue1.offer("f");
        myQueue1.offer("g");
        myQueue1.offer("h");
        myQueue1.offer("i");
        myQueue1.offer("j");
        myQueue1.offer("k");
        myQueue1.offer("l");

        myQueue2.offer("m");
        myQueue2.offer("n");
        myQueue2.offer("o");
        myQueue2.offer("p");
        myQueue2.offer("q");
        myQueue2.offer("r");
        myQueue2.offer("s");
        myQueue2.offer("t");
        myQueue2.offer("u");
        myQueue2.offer("v");
        myQueue2.offer("w");

        InAndOut es = new InAndOut("First", myQueue1, myQueue2);
        InAndOut es2 = new InAndOut("Second", myQueue2, myQueue1);

        es.start();
        es2.start();
    }
}

提前致谢!

Answer 1

即使线程调度是确定性的，观察到的行为仍然是合理的。只要两个线程执行相同的任务，它们就可能平衡运行，尽管您不能依赖。但一旦一个队列空了，任务就不再平衡了。比较:

1. 线程 1 从包含项目的队列中进行轮询。 poll 方法将修改源队列的状态以反射(reflect)删除，您的代码将接收到的项目插入到另一个队列中，创建一个内部列表节点对象并修改目标队列的状态以反射(reflect)插入。所有修改均以其他线程可见的方式执行。

2. 从空队列中线程化两个轮询。 poll 方法检查引用并找到 null 就这样了。不执行任何其他操作。

我认为很明显，一旦一个队列变空，一个线程就会比另一个线程有更多的事情要做。更准确地说，一个线程可以完成 3000 次循环迭代(甚至可以完成 300000 次)，而另一个线程甚至不足以执行一次迭代。

因此，一旦一个队列为空，一个线程几乎立即完成其循环，之后另一个线程会将所有项目从一个队列传输到另一个队列，然后也完成。

因此，即使具有几乎确定性的调度行为，一旦一个队列碰巧变空，平衡也始终会面临倾斜的风险。

<小时/>

您可以通过向队列添加更多项目来降低一个队列空载的可能性，从而提高平衡运行的机会。您可以提高迭代次数(远大于一百万)，以避免线程在队列为空时立即退出，或者仅在看到非 null 项时才增加计数器。您可以使用 CountDownLatch让两个线程在进入循环之前等待，补偿线程启动开销，使它们尽可能同步运行。

<小时/>

但是，请记住，它仍然是不确定的，并且轮询循环会浪费 CPU 资源。机器人尝试学习是可以的。