线程池的实现

 

我在修改方案中对两个方法加上了同步。我有一个疑义， 

private LinkedBlockingQueue<OjadbThread> ojadbThreads;
private BlockingQueue<OjadbTask> ojadbTasks

这两个队列本身是线程同步的，一定要再同步方法才能保证size取值是安全的么？

 

这么处理是一种线程实现策略，策略可以定义多种。我将在有空时研究下JDK的Fixed线程的实现策略。谢谢。

 

 result不会返回null，这里我默认返回值是String了，应该是不严谨的。我增加了try的处理。

 

这点我在设计时就想到了，但我很无奈。如果你有好的办法，希望不吝赐教！谢谢。

package ojadb.core.thread;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class OjadbThreadPool {

    private int minimum;
    private int maximum;
    private LinkedBlockingQueue<OjadbThread> ojadbThreads;
    private BlockingQueue<OjadbTask> ojadbTasks;

    public OjadbThreadPool(int min, int max) throws OjadbException {
        if (0 < min && min < max) {
            minimum = min;
            maximum = max;
            ojadbThreads = new LinkedBlockingQueue<OjadbThread>();
            ojadbTasks = new LinkedBlockingQueue<OjadbTask>();
            for (int i = 0; i < minimum; i++) {
                OjadbThread worker = new OjadbThread(this, "worker" + i);
                worker.start();
            }
        } else {
            throw new OjadbException("无法建立线程池: 最小值必须大于零,小于最大值");
        }
    }

    void addWorker(OjadbThread ojadbThreadWorker) {
        ojadbThreads.offer(ojadbThreadWorker);
    }

    public void excute(OjadbTask thread) {
        // 生产任务
        ojadbTasks.offer(thread);

        OjadbThread worker = ojadbThreads.remove();
        if (worker != null) {
            worker.work();
        } else if (ojadbThreads.size() < maximum) {
            OjadbThread newWorker = new OjadbThread(this, "worker" + ojadbThreads.size());
            newWorker.start();
        } else {
        }
        releaseThreads();
    }

    private synchronized void releaseThreads() {
        while (ojadbThreads.size() > minimum) {
            OjadbThread worker = ojadbThreads.remove();

            if (worker != null) {
                System.out.println(worker + "is killed");
                worker.kill();
            }
        }
    }

    public synchronized OjadbTask getTask() {
        if (ojadbTasks.size() == 0) {
            return null;
        }
        // 消费任务
        return ojadbTasks.remove();
    }
}

 

package ojadb.core.thread;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class OjadbThread extends Thread {

    private AtomicBoolean isActived;
    private ReentrantLock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();
    private OjadbThreadPool pool;

    public OjadbThread(OjadbThreadPool pool, String threadName) {
        super(threadName);
        this.pool = pool;
        isActived = new AtomicBoolean(true);
    }

    @Override
    public synchronized void start() {
        super.start();
    }

    @Override
    public void run() {
        try {
            lock.lock();

            while (isActived.get()) {
                OjadbTask task = pool.getTask();
                if (task != null) {
                    try {
                        String result = (String) task.excute();
                        if (result.length() > 0) {
                            pool.addWorker(this);
                            // 懒工人策略 -- 没人招呼工作，绝不睡醒
                            condition.await();
                        }
                    } catch (Exception e) {
                    }
                } else {
                    pool.addWorker(this);
                    condition.await();
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public void work() {
        condition.signal();
    }

    public void kill() {
        isActived.set(false);
        work();
    }

    @Override
    public String toString() {
        return getName();
    }
}

 

 

package ojadb.core.thread;

public class OjadbTask {
    private String threadName;

    public OjadbTask(String threadName) {
        this.threadName = threadName;
    }

    public Object excute() {
        System.out.println("excuting " + threadName);
        return "success";
    }

    public String getName() {
        return threadName;
    }

    public String toString() {
        return threadName;
    }
}

 

   对象本身同步并不代表对象与外部多个操作是同步的。其实啊，这是很多刚开始接触多线程编程的人的误解的，如果想了解更多，推荐你看一本书，电子工业出版社的《java并发编程实践》
倒不是固定数量的线程池是好的。作为一个池来说，伸缩性是非常重要的，但是另外一个关键就是伸缩性的效率的效果更加是应该考虑的，否则会适得其反。其实你看看我写的那个关于task/worker的帖子吧，其中worker就是一个工作线程池，它是如何根据request的情况来进行伸缩性的调整的。

   所以我用的是危险这个词。你如果做的是一个中间组件供大家使用。你就不会清楚知道调用者任务的具体实现是怎样。

   其实很容易，引入管理线程。这个管理线程可以管理一个控制变量，从而实现对其他子线程的控制。同时管理线程在所有子线程结束后再进行一些相关处理（比如将文件移入到指定文件夹，确保流关闭等等）。

这句话什么意思，偶程序没有看明白，能否解释的清楚一点儿？