asyncprocessor.java

来自「基于J2ME的手机地图客户端源码」· Java 代码 · 共 93 行

package org.sreid.j2me.util;

import java.util.*;

/**
 * AsyncProcessor for handing off tasks to asynchronous worker threads.
 * Construct one, then call enqueueJob(Runnable job) to run job in the background.
 */
public class AsyncProcessor {
	private final Vector queue = new Vector(); //XXX remove is O(n)
	private final Object lock = this;
	private boolean disposed = false;

	/**
	 * Constructs an AsyncProcessor and starts the specified number
	 * of worker threads.
	 */
	public AsyncProcessor(int numberOfWorkerThreads) {
		for (int i = 0; i < numberOfWorkerThreads; i++) {
			new Worker().start();
		}
	}

	/**
	 * Disposes of this AsyncProcessor. No new jobs can be added, and worker 
	 * threads will terminate once the current job queue is drained.
	 */
	public void dispose() {
		synchronized (lock) {
			disposed = true;
			lock.notifyAll();
		}
	}


	/**
	 * Adds a job who's run() method will be invoked asynchronously by a worker thread.
	 */
	public void enqueueJob(Runnable job) {
		if (job == null) throw new IllegalArgumentException("job is null");
		synchronized (lock) {
			if (disposed) {
				throw new IllegalStateException("Attempt to add job to disposed AsyncProcessor queue.");
			}
			queue.addElement(job);
			lock.notify();
		}
	}

	private class Worker extends Thread {

		Worker() {
			super();
		}

		public void run() {
			for (;;) {
				// Get a job from the queue
				Runnable job = null;
				synchronized (lock) {
					while (job == null) {
						if (queue.size() > 0) {
							job = (Runnable)queue.firstElement();
							queue.removeElementAt(0);
						}
						else if (disposed) {
							return;
						}
						else {
							try { lock.wait(); }
							catch (InterruptedException e) { }
						}
					}
				}
				// Run it
				boolean done = false;
				try {
					job.run();
					done = true;
				}
				finally {
					if (!done) {
						// Some kind of exception/error. Allow it to be caught by the default
						// exception handler, but start another worker to replace this one.
						new Worker().start();
					}
				}
			}
		}
	}
}