java - 为什么我的排队任务没有被线程池中的所有线程处理？

Question

我编写了一个简单的 Java 程序来帮助使用线程和线程池来完成某些任务。在我的程序中，有类 TheObject 的对象必须对它们进行某种处理(在本例中，只是 sleep 延迟并打印出它们的字段)。

TheObject 对象被放置在队列中，WorkerThread 从中提取并处理它们。

我创建了一个Manager类来初始化TheObjects和WorkerThread。当我运行它时，我得到奇怪的输出；一个线程正在处理所有事情，而不是由线程关闭工作!为什么会发生这种情况？如何让线程分担工作负载？

这是输出:

--------------------------------------------
alfred   a   0
Thread-0
--------------------------------------------
bob   b   1
Thread-0
--------------------------------------------
carl   c   2
Thread-0
--------------------------------------------
dave   d   3
Thread-0
--------------------------------------------
earl   e   4
Thread-0
--------------------------------------------
fred   f   5
Thread-0
--------------------------------------------
greg   g   6
Thread-0
--------------------------------------------
harry   h   7
Thread-0
--------------------------------------------
izzie   i   8
Thread-0
--------------------------------------------
jim   j   9
Thread-0
--------------------------------------------
kyle   k   0
Thread-0
--------------------------------------------
Larry   L   1
Thread-1
--------------------------------------------
Michael   M   2
Thread-1
--------------------------------------------
Ned   N   3
Thread-0
--------------------------------------------
Olaf   O   4
Thread-0
--------------------------------------------
Peter   P   5
Thread-0
--------------------------------------------
Quincy   Q   6
Thread-0
--------------------------------------------
Raphael   R   7
Thread-0
--------------------------------------------
Sam   S   8
Thread-0
--------------------------------------------
Trixie   T   9
Thread-0

<小时/>

代码:

类(class):

• 经理
• 对象
• 对象队列
• 工作线程

经理

public class Manager {

    public static void main(String[] args) throws InterruptedException {
        TheObjectQueue queue = new TheObjectQueue();
        //Create Objects
        int numberOfInitialObjects = 10;
        String[] arrayOfNames = {"alfred", "bob", "carl", "dave", "earl", "fred", "greg", "harry", "izzie", "jim"};
        //char[] arrayOfChars = {'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j'};
        //int[] arrayOfNums = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
        for (int i = 0; i < numberOfInitialObjects; i++){
                TheObject anObject = new TheObject(arrayOfNames[i], arrayOfNames[i].charAt(0), i);
                queue.addToQueue(anObject);

        }

        int numberOfThreads = 2;
        for (int i = 0; i < numberOfThreads; i++){
            WorkerThread workerThread = new WorkerThread(queue);
            workerThread.start();
        }

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }


        String[] arrayOfNames2 = {"kyle", "Larry", "Michael", "Ned", "Olaf", "Peter", "Quincy", "Raphael", "Sam", "Trixie"};
        for (int i = 0; i < numberOfInitialObjects; i++){
            TheObject anObject = new TheObject(arrayOfNames2[i], arrayOfNames2[i].charAt(0), i);
            queue.addToQueue(anObject);
        }

    }
}

对象

public class TheObject {
    private String someName;
    private char someChar;
    private int someNum;

    public TheObject(String someName, char someChar, int someNum) {
        super();
        this.someName = someName;
        this.someChar = someChar;
        this.someNum = someNum;
    }

    public String getSomeName() {
        return someName;
    }
    public char getSomeChar() {
        return someChar;
    }
    public int getSomeNum() {
        return someNum;
    }
}

对象队列

import java.util.LinkedList;

public class TheObjectQueue {

    private LinkedList<TheObject> objectQueue = null;

    public TheObjectQueue(){
        objectQueue = new LinkedList<TheObject>();
    }
    public LinkedList<TheObject> getQueue(){
        return objectQueue;
    }
    public void addToQueue(TheObject obj){
        synchronized (this) {
            objectQueue.addFirst(obj);
            this.notify();
        }

    }
    public TheObject removeFromQueue(){
        synchronized (this) {
            TheObject obj = objectQueue.removeLast();
            return obj;
        } 
    }
    public int getSize(){
        return objectQueue.size();
    }
    public boolean isEmpty(){
        return objectQueue.isEmpty();
    }
}

工作线程

import java.util.Random;

public class WorkerThread extends Thread{
    private TheObjectQueue queue;

    public WorkerThread(TheObjectQueue queue){
        this.queue = queue;
    }
    public void process(TheObject obj){
        //Stall for a random amount of time
        Random random = new Random();
        int randomInt = random.nextInt(3)*1000;
        try {
            Thread.sleep(randomInt);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

        //Begin Printing
        System.out.println("--------------------------------------------");
        System.out.print(obj.getSomeName());
        System.out.print("   ");
        System.out.print(obj.getSomeChar());
        System.out.print("   ");
        System.out.println(obj.getSomeNum());
        System.out.println(this.getName());
    }

    @Override
    public void run() {
        while(true){
            synchronized (queue) {
                while(queue.isEmpty()){
                    try {
                        queue.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                TheObject objToDealWith = queue.removeFromQueue();
                process(objToDealWith);
                super.run();
            }
        }
    }
}

Answer 1

WorkerThread.run 中(共享)队列的同步仅允许单个线程一次处理一个任务 - 效果实际上是一个工作池!在这种情况下，线程 0 在获得锁时大部分时间都“获胜”；同步锁获取是not guaranteed to be fair .

一个简单的修复方法是使用所需的线程安全构造从队列中获取任务，然后在同步部分的外部处理该任务。这允许工作人员同时处理被假定为独立的任务。

// e.g.
@Override
public void run() {
    while(true){
        TheObject objToDealWith; 
        synchronized (queue) {
            // Only synchronized on queue when fetching task
            objToDealWith = getTask(queue);
        }

        // Process task; independent of queue
        process(objToDealWith);
    }
}

由于其中一个线程将忙于处理任务，而另一个线程正在获取(或已获取)队列上的锁，因此工作分配将“公平”。