深入 Eclipse 多线程机制

Eclipse 提供了一套多线程类库（包括 Job 等）极大的方便了开发人员对多线程程序的处理。本文通过对 Eclipse 内核代码的研究，分析 Eclipse 多线程库的内部实现机制，特别是其内部线程池的实现方式，Job 的调度，线程同步机制等。读者通过阅读本文，可以深入了解 Eclipse 多线程机制，同时加深对 Java 线程的理解。这对于我们设计多任务系统，以及解决多线程问题将大有帮助。

简介

Eclipse 提供了一套多线程类库（包括 Job 等）极大的方便了开发人员对多线程程序的处理。本文通过对 Eclipse 内核代码的研究，分析 Eclipse 多线程库的内部实现机制，特别是其内部线程池的实现方式，Job 的调度，线程同步机制等。读者通过阅读本文，可以深入了解 Eclipse 多线程机制，同时加深对 Java 线程的理解。这对于我们设计多任务系统，以及解决多线程问题将大有帮助。建议读者在阅读本文前，先阅读本文作者的另一篇文章“ Eclipse 客户端程序中多线程的使用”（ http://www.ibm.com/developerworks/cn/opensource/os-cn-eclipse-multithrd/）对您理解本文大有裨益。

文章的组织结构如下：

• Jobs 的框架
• 并发模型－线程池机制
• 并发模型－ Job 的调度
• 并发模型－线程同步机制
• 小结

• Jobs 的框架

  Eclipse 在 org.eclipse.core.runtime.osgi 运行时插件里提供了 Jobs API 。 Jobs API 被广泛的应用到 Eclipse 平台中，用户所开发的 eclipse 插件里。 Job 是 Eclipse 运行时重要的组成部分（基于 equinox 的 OSGi 框架则是 Eclipse 运行时的最重要的组成部分）。 Job 可以理解成被平台调用异步运行的代码块。 Job 可以异步的执行，多个 Jobs 可以并发执行。那么读者会问了？为什么 eclipse 平台要提供这样的 API 出来，为什么不直接使用 java.lang.Thread 呢？

  原因有以下几点：

  1）性能更好：通过使用线程池实现了线程共享，减少了创建和销毁线程的开销，提高了性能。

  2）交互响应信息：Job 提供了一个框架，开发人员使用 Job 很容易实现与用户的交互，例如允许用户取消 Job 的执行或者显示 Job 。

  3）调度的灵活性：可以马上运行一个 Job，可以稍后运行一个 Job, 还可以反复的运行一个 Job

  4）Job 的监听机制：Job 监听器监听 Job 的状态信息，比如，知道一个 Job 何时开始运行以及何时结束运行等。

  5）优先级及互斥的灵活应用：Job 提供了多种方式来控制 Job 的调度，开发者可以设定 Job 的优先级（读者应注意这一点，JobManager 不保证优先级高的 Job 一定比优先级低的 Job 先被调度执行），也可以使用调度规则保证 Jobs 的同步与互斥。

  下面我们首先介绍一下 Jobs 的框架 , 如图 1 所示。其囊括了 org.eclipse.core.runtime.jobs 包内的所有接口和类。

  
  图 1. Jobs 框架
   
  

  IJobManager 是 Job 管理器类的接口，其中定义了一些对 Job 以及 JobFamily 操作的一些 API 。有关 Job 管理器的实现，我们将在线程池机制一节中作详细介绍。

  当并发执行多个 Jobs 的时候，可能会引发冲突（Conflict）。 Job 的框架则充分考虑到了这种情况，并提供了管理和避免冲突的工具。 ISchedulingRule 接口，是用来管理共享资源访问冲突的技术。它使得 IJobManager 能够识别出冲突的 Jobs，进而能保证这些不能在一起执行的 Jobs 不在同一时间被调度或者启动。 ISchedulingRule 接口的子类 MultiRule 表示一组固定的子调度规则，如果任何一个 MultiRule 的子调度规则和另一个调度规则相冲突，那么该 MultiRule 和另一个调度规则就会发生冲突。形式化的说，一个组合调度规则表示其所有的子调度规则关于 isConflicting 方法等价的逻辑交集。组合调度规则不会再包含另一个组合调度规则，如果你把一个组合规则作为子规则加入到另一个组合规则中，算法就是该组合规则的所有子规则被加入到另一个组合规则中去了。

  ILock ，锁是用来对排他性资源的访问控制的。锁是可以重入的，也就是说同一个线程在不释放已有锁的情况下，可多次获取该锁。当成功获取锁的次数和成功释放锁的次数相等时，锁才能被释放掉。通过实现释放等待策略，锁避免了循环等待的死锁。如果一组线程陷入一个死锁，一个线程将会失去它所拥有的锁，进而打破死锁，使得其它的线程能够继续执行。一旦线程获得了运行所需要的所有的锁，它将获得对锁控制的排他性资源的访问。

  当一个线程正等待一个 acquire() 方法调用的时候，才可能会失去锁 ( 参见本文的线程同步机制部分 ) 。程序应用队列（先来先得）管理不同的线程来获取锁。线程获取的锁一定要释放掉，一般在 finally 程序块内释放锁。例如：

  lock.acquire(); 
   try { 
       // ... 执行程序 ... 
   } finally { 
       lock.release(); 
   }

  
  

  IJobChangeListener 接口，监听到 Job 的状态信息，进而执行相应的逻辑操作。

  ProgressProvider 类，为正在运行的 jobs 向 Job 管理器提供进度控制器。任何时候，它只有一个实例存在。该类仅由平台相关的插件内部使用。

  下面我们进入到 org.eclipse.core.internal.jobs 包内继续对 eclipse 并发 API 的进行讲解。该包内的类是 org.eclipse.core.runtime.jobs 包内接口或抽象类的具体实现。

  
  

  
  

  
  
  

  并发模型－线程池机制

  为了更高效的运行 Job，Eclipse 实现了一套线程池。 Eclipse 的线程池的实现主要涉及到三个类：Worker，WorkPool 和 JobManager 。

  
  图 2. Worker 和 WorkPool
   
  

  Worker 类继承自 java.lang.Thread 类，它执行 WorkPool 所提供的 job 。

  WorkPool 顾名思义，维护了一个 worker 线程的池子（也就是线程的集合）。线程是根据需要延迟构造的，如果过一会不用的话，最终线程会被丢弃掉。实际上来讲，线程池为 JobManager 提供了创建和销毁线程的相应管理。 JobManager 是 eclipse 平台添加的用以管理 Jobs 的 API 。

  JobManager 对 job 和锁 (lock) 提供调度（schedule），检索 (query) 和维护 (maintain) 的相关支持。 Job 管理器主要提供如下的服务：

  1）维持一个等待运行的 Job 队列，可以使用 schedule 方法把一个 job 加入到该等待队列中。

  2）允许对 Job family 的操作。（Job family 就是一组相关的 job 集合）可以取消，休眠或自动的唤醒某个 job family， Job 管理器也提供对于给定的 job family 检索其中 job 的服务。

  3）允许监听器知晓当前运行的 job 的进度信息。当运行的 job 的状态信息发生改变时，监听器能监听到，进而做出相应的逻辑操作。

  4）提供一个用来创建锁对象的工厂。 Lock 对象有预防死锁的策略，可以看成是一个灵敏的监视器（Monitor）

  5）为正在等待某一运行的 job 或 job family 完成的客户提供反馈信息。

  下面我们用顺序图图 3 来说明，当一个 Job 调用 schedule 方法时，内部的实现逻辑是怎么样的。 Job 继承了 InternalJob（Job 的内部实现类），Job 的 schedule 方法直接调用了父类 InternalJob 的 schedule 方法。 InternalJob 则持有 JobManager 的实例，在其 schedule 方法里调用了 JobManager 的 schedule 方法。清单 1 给出了 JobManager 的 schedule 方法，读者可见，在方法的末尾处，JobManager 调用了线程池类的 jobQueued 方法，。线程池类的 jobQueued 方法是线程管理的关键方法，它首先检查是否有可用的 Worker（正在睡眠的线程），如果有它通过调用 notify（）来唤醒 Worker 开始执行 Job，如果没有可用的 Worker，它会创建一个新的 Worker（线程），并启动这个 Worker 执行。清单 2 给出了 jobQueued 方法的实现。这里需要注意的是 Job.schedule（）方法只是启动了一次 Job 的调度，它有可能触发创建一个新的 Worker，但是新创建的 Worker 并不一定会执行刚 schedule 的 Job，具体哪个 Job 会被执行是由调度算法决定的。我们稍后会讨论 Job 的调度。

  
  图 3. schedule 方法调用顺序图
   
  
  清单 1. JobManager 的 schedule 方法
  

  protected void schedule(InternalJob job, long delay) { 
  	 if (!active) 
  		 throw new IllegalStateException("Job manager has been shut down."); 
  	 Assert.isNotNull(job, "Job is null"); //$NON-NLS-1$ 
  	 //call hook method outside sync block to avoid deadlock 
  	 if (!job.shouldSchedule()) 
  		 return; 
  	 synchronized (lock) { 
  		 //can't schedule a job that is already waiting, sleeping, or running 
  		 if (job.getState() != Job.NONE) 
  			 return; 
  		 //if it's a decoration job, don't run it right now if the system is busy 
  		 if (job.getPriority() == Job.DECORATE) { 
  			 long minDelay = running.size() * 100; 
  			 delay = Math.max(delay, minDelay); 
  		 } 
  		 if (delay > 0) { 
  			 job.setStartTime(System.currentTimeMillis() + delay);
  			 changeState(job, Job.SLEEPING); 
  		 } else { 
  	 job.setStartTime(System.currentTimeMillis() + delayFor(job.getPriority()));
  	 changeState(job, Job.WAITING); 
  		 } 
  	 } 
  	 //notify listeners outside sync block 
  	 jobListeners.scheduled((Job) job, delay); 
  	 //call the pool outside sync block to avoid deadlock 
  	 pool.jobQueued(job); 
   }

  
  
  清单 2. WorkPool 的 jobQueued 方法
  

  protected synchronized void jobQueued(InternalJob job) { 
  	 //if there is a sleeping thread, wake it up 
  	 if (sleepingThreads > 0) { 
  		 if (JobManager.DEBUG) 
  			 JobManager.debug("notifiying a worker"); //$NON-NLS-1$
  			 notify();
  			 return; 
  	 } 
  	 int threadCount = numThreads; 
  	 //create a thread if all threads are busy and we're under the max size 
  	 //if the job is high priority, we start a thread no matter what 
  	 if (busyThreads >= threadCount &&
  	  (threadCount < MAX_THREADS || 
  	  (job != null && job.getPriority() == Job.INTERACTIVE))) { 
  		 Worker worker = new Worker(this); 
  		 add(worker); 
  		 if (JobManager.DEBUG) 
  			 JobManager.debug("worker added to pool: " + worker); 
  			 //$NON-NLS-1$
  			 worker.start();
  			 return; 
  	 } else if (threadCount > MAX_THREADS) { 
  	  String msg = "The job manager has stopped allocating worker threads 
  	   because too many background tasks are running."; //$NON-NLS-1$ 
        InternalPlatform.getDefault().
         log(new Status(IStatus.ERROR, IPlatform.PI_RUNTIME, 1, msg, null)); 
  	 } 
   }