如何使用LinkedBlockingQueue运行任务

Question

我目前正在尝试让我的代码使用simpleTask中的代码打印20次simpleTesting方法中的内容。其思想是simpleTesting将20个simpleTask实例添加到一个队列中，然后从simplePoolThread中的队列中取出它们。应该发生的是，它输出测试消息20次，然后继续运行，同时从队列中寻找更多的内容(但没有)。相反，它目前只是不打印任何东西，并持续运行。下面是我的代码(其中很多是接口，我相信问题在于simpleThreadPool代码)：

package simpleThreadPool;

/**
 *   <<-- Pool Thread -->> 
 *
 *   It will be running continuously. It will try to retrieve new tasks when it is idle. 
 */
public interface ISimplePoolThread extends Runnable {
    /**
     *   Use an infinite loop to retrieve and perform tasks.
     */
    @Override
    public void run();
}

。

package simpleThreadPool;

/**
 *   <<-- Simple Task -->> 
 *
 *   ISimpleTask is to be performed by PoolThread. 
 */
public interface ISimpleTask{
    /**
     *   #1. Create a class to implement ISimpleTask, put content of the task to method run().
     */
    public void run();
}

。

    package simpleThreadPool;


    /**
     *   <<-- Thread Pool -->> 
     *   It manages a queue of tasks, starts some pool threads.
     *      
     *   #1. Create a task queue by using queue data structures, or designing your own data structure. 
     */
    public interface ISimpleThreadPool {

        /**
         *   #1. Initialize your queue (or do so in somewhere)
         *   #2. Starts some ISimplePoolThreads.
         */
        public void start();

        /**
         *   #1. Stops everything
         */
        public void stop();

        /**
         *   #1. Add a task to your queue.
         */
        public void addTask(ISimpleTask task);
    }

。

    package simpleThreadPool;

public class SimpleTask implements ISimpleTask {

    @Override
    public void run() {
        System.out.println("testing testing 1 2 3");
    }
}

。我认为问题在于这段代码，其中任务是从队列中提取的：

package simpleThreadPool;

import java.util.concurrent.LinkedBlockingQueue;

public class SimplePoolThread implements ISimplePoolThread, Runnable {


    private LinkedBlockingQueue<ISimpleTask> queue = new LinkedBlockingQueue<>();

    @Override
    public void run() {

        while(true) {
            System.out.println("Inserting Element: ");
            try {
                queue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


}

。

package simpleThreadPool;

import java.util.concurrent.LinkedBlockingQueue;

public class SimpleThreadPool implements ISimpleThreadPool {

    private LinkedBlockingQueue<ISimpleTask> queue = new LinkedBlockingQueue<>();

    @Override
    public void start() {
        (new Thread(new SimplePoolThread())).start();
    }

    @Override
    public void stop() {
        try {
            queue.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void addTask(ISimpleTask task) { 

        try {
            queue.put(task);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

。

测试文件：

package simpleThreadPool;

public class SimpleTesting implements ISimpleTask{

    private int i;

    public SimpleTesting(int i){
        this.i = i;
    }

    @Override
    public void run() {
        System.out.println(i);
    }

    public static void main(String args[]){
        // Initialize thread pool
        SimpleThreadPool pool = new SimpleThreadPool();
        pool.start();
        // Create 20 tasks
        for(int i = 1; i<=20; i++){
            pool.addTask(new SimpleTesting(i));
        }
    }
}

Answer 1

SimplePoolThread中的任务队列是阻塞队列。一旦启动，它就执行queue.take()。抓取是一种阻塞操作。线程一直处于等待状态，直到其他东西将任务添加到队列中。

你对问题地点的预感非常接近。问题是，SimplePoolThread中的队列和SimpleThreadPool中的队列并不相同；您有两个独立的队列。因此，当SimpleTesting添加任务时，它们进入池的队列，而不是线程的队列。所以，线将永远坐在那里，什么也不等待。您还忘了在SimplePoolThread中实际运行任务。

试一试下面的方法。

public class SimpleThreadPool implements ISimpleThreadPool {

private LinkedBlockingQueue<ISimpleTask> queue = new LinkedBlockingQueue<>();

@Override
public void start() {
    (new Thread(new SimplePoolThread(queue))).start();
}

注意，池中的队列被传递到线程中。然后，线程保存对此队列的引用。在线程的run()期间，它现在也实际运行任务。

public class SimplePoolThread implements ISimplePoolThread, Runnable {


private LinkedBlockingQueue<ISimpleTask> queue;

public SimplePoolThread(LinkedBlockingQueue<ISimpleTask> queue)
{
    this.queue = queue;
}

@Override
public void run() {

    while(true) {
        System.out.println("Inserting Element: ");
        try {
            ISimpleTask task = queue.take();
            task.run();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

这是输出：

Inserting Element: 
1
Inserting Element: 
2
Inserting Element: 
3

..etc..

我认为这是为了家庭作业，否则我会告诉您不要重新发明轮子，去使用Java的内置池服务。

http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/package-summary.html