胡思乱想JNI与线程池的维护

Posted 2021-01-15 longfurcat

JNI中，C/C++代码里创建的资源不由Java GC处理，故这里的资源必须由C/C++代码明确释放。在JNI中，C/C++回调Java的方法是调用一个CallXXMethod函数来实现的，如果回调的方法结束，C/C++执行下一行代码。

故猜测，由C/C++创建的OS线程应该会在运行完run方法后释放，不然好像也没有其他合适的时间点来对线程进行释放了。因为按照语义的话，既然线程的任务已经完成，那线程还留着干什么，就应该被释放。

有些时候，我们需要维护一个线程池来减少创建和释放线程的开销，让一些线程完成当前任务后被收回到线程池，等待接受下一个任务。根据前面的猜测，run方法结束后，OS线程将被释放。我们要维护线程池，就是不能让线程被释放。所以我们就要阻止run方法返回。当Thread.run把target.run执行完的时候，利用wait挂起线程。直到有新的target，才唤醒当前线程。

以下是我在旧版的《Thinking in Enterprise Java》中看到的线程池的实现。

public class Worker extends Thread { //工作者线程
    public static final Logger logger = Logger.setLogger("Worker"); //类日志
    private String workerId; //工作者ID
    private Runnable task;   //任务对象
    private ThreadPool threadPool;  //线程池引用，方便操作。

    static { //静态块，配置logger
        try {
            logger.setUseParentHandlers(false);
            FileHandler ferr = new FileHandler("WorkerErr.log");
            ferr.setFormatter(new SimpleFormatter());
            logger.addHandler(ferr);
        } catch(IOException e) {
            System.out.println("Logger not initialized.");
        }
    }

    public Worker(String id, ThreadPool pool) {
        workerId = id;
        threadPool = pool;
        start();    //创建即启动
    }

    public void setTask(Runnable t) { //这里放入新任务，并唤醒线程，进入就绪队列。
        task = t;
        synchronized(this) {
            notify(); //wait、notify方法都必须获取对应的锁
        }
    }

    public void run() {
        try {
            while(!threadPool.isStopped()) { //如果线程池未停止工作，此线程不释放
                synchronized(this) {
                    if(task != null) {
                        try {
                            task.run(); //执行任务
                        } catch(Exception e) {
                            logger.log(Level.SERVER, "Exception in source Runnable task", e);
                        }
                        threadPool.putWorker(this); //完成当前任务，回收到线程池
                    }
                    wait(); //完成任务或无任务时，挂起线程。免得空循环浪费时间片。

                }
            }
            //跳出循环，意味着线程池结束工作，此线程也将停止工作
            System.out.println(this + " Stopped");
        } catch(InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public String toString() {
        return "Worker: " + workerId;
    }

}

public class ThreadPool extends Thread { //线程池, 同样是一个线程，负责接收和分配任务
    private static final int DEFAULT_NUM_WORKERS = 5; //默认线程池大小为5
    private LinkedList workerPool = new LinkedList(); //空闲线程列表 
    private LinkedList taskQueue = new LinkedList();  //任务列表
    private boolean stopped = false; //线程池的工作状态
    
    public ThreadPool() { //默认构造方法，线程池大小默认
        this(DEFAULT_NUM_WORKERS);
    }

    public ThreadPool(int numOfWorkers) { //自定义线程池大小
        for(int i=0;i<numOfWorkers;i++){
            workerPool.add(new Worker("" + i, this));
        }
        start(); //创建即启动
    }

    public void run() { //分发任务
        try {
            while(!stopped) {
                if(taskQueue.isEmpty()) { //如果任务队列为空，挂起当前线程。 也就是暂停线程池的分发任务的工作
                    synchronized(taskQueue) {
                        taskQueue.wait(); //不管调用哪个对象的wait方法，都是挂起当前执行它的线程。
                    }
                } else if(workerPool.isEmpty()) { //如果没有空闲的线程，则暂停线程池的工作。
                    synchronized(workerPool) {
                        workerPool.wait();
                    }
                }
                //有任务，且有空闲线程的情况  => 从空闲线程中取出一个线程，让其负责任务队列中的一个任务
                getWorker().setTask((Runnable)taskQueue.removeLast());
            }
        } catch(InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public void addTask(Runnable task) {
        synchronized(taskQueue) {
            taskQueue.addFirst(task);
            taskQueue.notify(); //通知已有新任务，如果前面线程因无任务被挂起，这个操作将唤醒线程
        }
    }

    public void putWorker(Worker worker) {
        synchronized(workerPool) {
            workerPool.addFirst(worker);
            workerPool.notify(); //通知已有新空闲线程，如果前面线程因无空闲工作者线程被挂起，此操作将唤醒线程
        }
    }

    public Worker getWorker() {
        return (Worker) workerPool.removeLast(); //取出一个空闲线程，并从列表中移除。
    }

    public boolean isStopped() {
        return stopped;
    }

    public void stopThreads() { //关闭线程池
        stopped = true;
        Iterator it = workerPool.Iterator();
        //这里唤醒挂起的工作者线程，使得它醒来并发现ThreadPool已关闭，并结束run方法 => 释放OS线程
        while(it.hasNext()) { 
            Worker w = (Worker)it.next();
            synchronized(w) {
                w.notify();
            }
        }
    }

}

对这个代码的认识，在注释里已经表现的很清楚了。另外，我还觉得，ThreadPool和Woker的关系有点观察者模式的味道，Woker是观察者，ThreadPool是被观察者/主题。不过，与标准的观察者模式不同的是，ThreadPool接受到新任务（发生了变化），并没有通知所有Worker。