多线程处理耗时任务的封装方式

 

工作中用到的处理多任务的多线程实现，以下仅为简略书写以便备忘、

 

package need.most.time.producer;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import need.most.time.ProcessorWorkerThread;

public class ProcessingThreadsDispatcher {
	protected int maxQueueSize = new Long(Runtime.getRuntime().maxMemory() / 400000L).intValue();
	int def_pool_size = 100;
	private String defPluginsThreadsPool = "default-threads-pool";
     //这里用一个map来封装，也就是可能有多种处理耗时任务的processor处理器，每一个处理器对应一个处理线程沲也就最合理
	private Map<String, ProcessingThreads<ProcessorWorkerThread>> workerThreads = new ConcurrentHashMap<String, ProcessingThreads<ProcessorWorkerThread>>(32);
	
	ProcessorWorkerThread worker;
	ProcessingThreads<ProcessorWorkerThread> pt;
	{
		worker = new ProcessorWorkerThread();
		pt = new ProcessingThreads<ProcessorWorkerThread>(worker, def_pool_size, maxQueueSize, defPluginsThreadsPool);
		workerThreads.put(defPluginsThreadsPool, pt);
	}
	
	
	//主要的方法入口
	private void walk(final Object arg) {
		//processor 为处理器，专处理耗时任务
		Object processor = new Object();
        String processorId=defPluginsThreadsPool;//应该每一个processor都提供一个唯一的name，这样就可以得到相对应processor
		ProcessingThreads<ProcessorWorkerThread> pt = workerThreads.get(processorId);

		if (pt == null) {
			pt = workerThreads.get(defPluginsThreadsPool);
		}
		if (pt.addItem(processor,arg)) {
			//记录哪个处理器处理的哪个对象等
		}
		
	}
}

 

 

 

package need.most.time;

// 通用的处理器实现
public class ProcessorWorkerThread extends WorkerThread {

	//一般这里就会执行真实代表的处理器实现方法
	@Override
	public void process(QueueItem item) {
		
		Object processor=item.getProcessor();
		Object arg=item.getArg();
		//processor拿着arg对象进行处理耗时任务。
		//processor.process(...)
		System.out.println("do finish");
	}

	@Override
	public WorkerThread getNewInstance() {
		ProcessorWorkerThread worker = new ProcessorWorkerThread();
		return worker;
	}

}

 

 

package need.most.time;

import java.util.concurrent.LinkedBlockingQueue;
//线程处理抽象类
public abstract class WorkerThread extends Thread {

	private LinkedBlockingQueue<QueueItem> queue = null;
	private boolean stopped = false;

	public abstract void process(QueueItem item);
	public abstract WorkerThread getNewInstance();
	public boolean offer(QueueItem item) {
		return queue.offer(item);
	}
	
	public void setQueueMaxSize(int maxSize) {
		LinkedBlockingQueue<QueueItem> oldQueue = queue;

		queue = new LinkedBlockingQueue<QueueItem>(maxSize);

		if (oldQueue != null) {
			queue.addAll(oldQueue);
		}
	}

	@Override
	public void run() {
		QueueItem item = null;

		while (!stopped) {
			try {
				item = queue.take();

				long start = System.currentTimeMillis();

				process(item);

				long end = System.currentTimeMillis() - start;
				System.out.println("do time=" + (end - start) + "ms");
			} catch (Exception e) {
			}

		}

	}
	
	
	

}

 

 

 

 

package need.most.time.producer;

import java.util.ArrayList;

import need.most.time.QueueItem;
import need.most.time.WorkerThread;
//处理线程沲类，些类维护多条处理线程
public class ProcessingThreads<E extends WorkerThread> {
	private int numWorkerThreads = 1;
	private ArrayList<E> workerThreads = null;
	private String name = null;

	public ProcessingThreads(E worker, int numWorkerThreads, int maxQueueSize, String name) {
		this.numWorkerThreads = numWorkerThreads;
		this.workerThreads = new ArrayList<E>(numWorkerThreads);
		this.name = name;
		// 生成指定数量的工作线程
		for (int j = 0; j < numWorkerThreads; j++) {
			WorkerThread t = worker.getNewInstance();

			t.setQueueMaxSize(maxQueueSize);
			t.setDaemon(true);
			t.setName(name + " Queue Worker " + j);
			t.start();
			workerThreads.add((E) t);
		}
	}

	public boolean addItem(Object processor, Object arg) {
		boolean ret = false;
		QueueItem item = new QueueItem(processor, arg);
		//这里应该指定哪一个用户他的唯一标识，这样可以每次处理同一个用户都是由同一个线程来执行
		ret = workerThreads.get(Math.abs(processor.hashCode()) % numWorkerThreads).offer(item);
		return ret;
	}

}

 

 

 

 

 

 

package need.most.time;
//封装一些处理参数bean
public class QueueItem {
	private Object arg;
	private Object processor;
	public QueueItem(Object arg, Object processor) {
		super();
		this.arg = arg;
		this.processor = processor;
	}
	public Object getArg() {
		return arg;
	}
	public void setArg(Object arg) {
		this.arg = arg;
	}
	public Object getProcessor() {
		return processor;
	}
	public void setProcessor(Object processor) {
		this.processor = processor;
	}
	

}