0

每个人！

我写了一个InAndOut扩展类（）Thread。此类在构造函数中接收两个LinkedConcurrentQueue,entrance和exit, 我的run方法将对象从 转移entrance到exit。

在我的主要方法中，我实例化了两个LinkedConcurrentQueue,myQueue1和myQueue2, 每个都有一些值。然后，我实例化了两个 InAndOut，一个接收myQueue1（入口）和myQueue2（出口），另一个接收myQueue2（入口）和myQueue1（出口）。然后，我调用两个实例的 start 方法。

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

为什么会发生以及如何解决？有一些方法可以不在我的运行方法中使用这个睡眠调用吗？难道我做错了什么？

这是我的源代码：

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();
    }
}

提前致谢！

4

3 回答 3

1

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

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

  2. 从一个空队列线程两个轮询。该poll方法检查引用并找到null，仅此而已。不执行其他操作。

我认为很明显，一旦一个队列变空，一个线程比另一个线程要做的事情要多得多。更准确地说，一个线程可以完成其 3000 次循环迭代（它甚至可以完成 300000 次），而另一个线程甚至无法执行一次迭代。

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

因此，即使使用几乎确定性的调度行为，一旦一个队列碰巧变空，天平总是会承担倾斜的风险。

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

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

于 2013-11-11T22:15:07.467 回答
0

线程的执行顺序是未定义的，所以任何事情都可能发生。但是，由于您不会同时启动两个线程，因此您可以对可能发生的情况做出一些假设：

  1. es 首先启动，因此如果 CPU 足够快，它已经在 es2 启动之前将 queue1 中的所有内容推送到 queue2 中，然后进入睡眠状态。
  2. es2 开始并将队列 2 中的 1 个元素放回队列 1。
  3. es 同时唤醒并将元素放回原处。

由于两个线程应该以相同的速度“大约”工作，因此一个可能的结果是 in 中只有 1 个或没有元素，es而es2.

于 2013-11-11T19:31:23.070 回答
0

当 jtahlborn 说多线程是非确定性的时，他是完全正确的，因此我建议您更多地阅读您在此应用程序中的期望，因为它不是很清楚并且它按我的预期运行（基于如何它是编码的）。

话虽如此，您可能正在寻找 BlockingQueue 而不是 ConcurrentLinkedQueue。如果线程为空，阻塞队列将暂停线程，并等待它在继续之前有一个项目。用 LinkedBlockingQueue 交换 ConcurrentLinkedQueue。

两者之间的区别在于，如果 ConcurrentLinkedQueue 没有项目，它将快速返回一个null值，因此它可以非常快速地完成 3000 次迭代。

于 2013-11-11T19:38:19.037 回答