Android ThreadLocal+PriorityQueue構建多線程隊列

1、消息隊列

Android中使用了不少消息隊列，如Intent,Handler,BroadcastReceiver等。使用消息隊列的目的是防止線程阻塞而且將不能及時執行的任務緩存起來，從而提升系統的運行效率。

爲了使得消息隊列具備全局性，而且重用性，建議定義在Application或者單例對象的類中

public class QApplication extends Application{

   public static final ThreadLocal<PriorityQueue<TaskMessage>> massegQueue = new ThreadLocal<PriorityQueue<TaskMessage>>();
    @Override
    public void onCreate() {
	super.onCreate();
    }
}

TaskMessage是一個任務消息，例如以下，注意，必須實現Comparable接口

package com.tianwt.app;

import java.io.Serializable;


public class TaskMessage implements Serializable,Comparable
{
	private long id;
	private String name;
	private String type;
	
	
	
	public TaskMessage(long id, String name, String type) {
		super();
		this.id = id;
		this.name = name;
		this.type = type;
	}

	public void show()
	{
	     System.out.println("id="+id+" , name="+name+" , type="+type);
	}

	public long getId() {
		return id;
	}

	public void setId(long id) {
		this.id = id;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public String getType() {
		return type;
	}

	public void setType(String type) {
		this.type = type;
	}

	@Override
	public int compareTo(Object arg0) {
		if(TaskMessage.class.isInstance(arg0))
		{
			TaskMessage tm = (TaskMessage) arg0;
			return tm.type.compareTo(type);
		}
		return 0;
	}
	
	
}

ThreadLocal是一個處理高併發問題的以線程爲做用域的類。

在Android中，消息的隊列使用了ThreadLocal,讀者可自行查Handler類相關的源碼。

PriorityQueue是一個能夠調整優先級的消息隊列

使用這種隊列的好處是，n個線程中能夠出現n個消息隊列，這個Handler相似，另外ThreadLocal是靜態的，但這並不會影響同步問題，由於它是以線程爲做用域。

對於ThreadLocal,可參見下面的文章。

Java中高併發任務中的ThreadLocal的用法解析

對於隊列的用法，請參考：Java 隊列協做同步

2、代碼實戰

下面給一個簡單的例子

package com.tianwt.app;

import java.util.PriorityQueue;

public class ThreadLocalQueue {
	// 指定初始值
	public static final ThreadLocal<PriorityQueue<TaskMessage>> massegQueue = new ThreadLocal<PriorityQueue<TaskMessage>>();

	public static void main(String[] args) {

		TestThreadQueue t1 = new TestThreadQueue();
		TestThreadQueue t2 = new TestThreadQueue();
		TestThreadQueue t3 = new TestThreadQueue();
		t1.start();
		t2.start();
		t3.start();
		
		
		TestThreadQueue[] tset = {t1,t2,t3};
		for (int i = 0; i < 100; i++) {
			
			tset[i%3].addTask(new TaskMessage(i,(i%3+1)+ "號線程", Math.random()*10000+""));
			
		}
	}

	private static class TestThreadQueue extends Thread {

		boolean isStop;
		private PriorityQueue<TaskMessage> taskQueue = null;

		public TestThreadQueue() {

		}

		public void shutDown() {
			if (taskQueue != null) {
				isStop = true;
				synchronized (taskQueue) {
					taskQueue.notify();
				}
			}
		}

		public void addTask(TaskMessage msg) {
			if (taskQueue != null && !isStop) {
				synchronized (taskQueue) {
					taskQueue.add(msg);
					taskQueue.notify();
				}
			}
		}

		public void run() {
			PriorityQueue<TaskMessage> qm = ThreadLocalQueue.massegQueue.get();
			if (qm == null) {
				taskQueue = new PriorityQueue<TaskMessage>(5);
				ThreadLocalQueue.massegQueue.set(taskQueue);
				qm = taskQueue;
			}

			while (!isStop) {
				synchronized (qm) {
					try {
						if (qm.size() != 0) {
							TaskMessage msgTask = qm.poll();
							msgTask.show();

						} else {
							qm.wait();
						}
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		}
	}

}