JAVA 已經給我們提供了比較好的佇列實現Queue，繼承於Collection。 本次我使用的是BlockingQueue，繼承於Queue。    

         在Concurrent包中，BlockingQueue很好的解決了多執行緒中，如何高效安全“傳輸”資料的問題。通過這些高效並且執行緒安全的佇列類，為我們快速搭建高質量的多執行緒程式帶來極大的便利。

         首先利用BlockingQueue封裝了一個佇列類。佇列裡存放Map物件，這個依專案需求而定，供參考。

import java.util.AbstractMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
/**
 * 單例的快取map
 */

public class CachePool<Key, Value> extends AbstractMap<Key, Value>{

		// 私有化快取物件例項
		private static CachePool cachePool = new CachePool();
		private int maxCount = 1000;
		private BlockingQueue<Entry> queue = new LinkedBlockingQueue<Entry>();
		/**
		 * private Constructor.
		 * @return 
		 */
		private CachePool() {
		}
		/**
		 * 開放一個公有方法，判斷是否已經存在例項，有返回，沒有新建一個在返回 
		 * @return
		 */
	    public static CachePool getInstance(){  
	        return cachePool;  
	    }  

		/**
		 * The Entry for this Map.
		 * @author  AnCan
		 *
		 */
		private class Entry implements Map.Entry<Key, Value>{
			private Key key;
			private Value value;

			public Entry(Key key, Value value){
				this.key = key;
				this.value = value;
			}

			@Override
			public String toString() {
				return key + "=" + value;
			}

			public Key getKey() {
				return key;
			}

			public Value getValue() {
				return value;
			}

			public Value setValue(Value value) {
				return this.value = value;
			}
		}



		/**
		 * Constructor.
		 * @param size the size of the pooled map;
		 */
		public CachePool(int size) {
			maxCount = size;
		}

		@Override
		public  Value put(Key key, Value value) {
			while(queue.size() >= maxCount){
				queue.remove();
			}
			queue.add(new Entry(key, value));
			return value;
		}

		@Override
		public Value get(Object key){
			for(Iterator<Entry> iter = queue.iterator();iter.hasNext();){
				Entry type = iter.next();
				if(type.key.equals(key)){
					queue.remove(type);
					queue.add(type);
					return type.value;
				}
			}
			return null;
		}

		@Override
		public Set<Map.Entry<Key, Value>> entrySet() {
			Set<Map.Entry<Key, Value>> set = new HashSet<Map.Entry<Key, Value>>();
			set.addAll(queue);
			return set;
		}

		@Override
		public void clear() {
			queue.clear();
		}

		@Override
		public Set<Key> keySet() {
			Set<Key> set = new HashSet<Key>();
			for(Entry e : queue){
				set.add(e.getKey());
			}
			return set;
		}

		@Override
		public Value remove(Object obj) {
			for(Entry e : queue){
				if(e.getKey().equals(obj)){
					queue.remove(e);
					return e.getValue();
				}
			}
			return null;
		}

		@Override
		public int size() {
            return queue.size();
        }
	}

 

            先看下消費者類，使用多執行緒來處理佇列中的內容：

import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;


/**
 * 操作業務類,通過引數中的方法引數進行具體的操作
 */
public class TicketTradeOper extends  HttpServlet
{
    /**
     * 快取物件 map
     */
    public static CachePool<String, Object> mapPool = CachePool.getInstance();

    private static final int NTHREADS=5;  
    // 使用執行緒池來避免 為每個請求建立一個執行緒。  
    private static final Executor threadPool=Executors.newFixedThreadPool(NTHREADS);  

    //業務操作
    IETicketTradeOper ticketTradeOper;

    @Override
    public void init() throws ServletException
    {
        Timer timer = new Timer();  
        timer.schedule(new TimerTask(){     
            @Override          
                public void run() {        
                    startThread(); 
                }     
            }, new Date(), 5000);//間隔5秒執行一次定時器任務
        super.init();
    }


    public void startThread(){
        threadPool.execute(new Runnable(){  
            public void run() {  
                executeCodeOper();  
             }  
        }); 
    }

    public  void executeCodeOper()
    {
        String key = "";
        Map param = null;
        synchronized (mapPool)
        {
            System.out.println(Thread.currentThread().getName() + "進來了。。。。");
            System.out.println("現在佇列中共有----"+mapPool.size()+"---條資料");

            Iterator it = mapPool.keySet().iterator();
            //快取不為空時,取出一個值
            while (it.hasNext())
            {
                key = (String) it.next();
                param = (Map) mapPool.get(key);
            }
            if (null != param)
            {
                  //為防止重複,將其移除
                  mapPool.remove(key);
            }
        }

        if (null != param)
        {
            boolean result =ticketTradeOperator(param);
            System.out.println("此條資料處理========"+result);
            if(!result){
                //若處理失敗,重新放回佇列
                mapPool.put(key, param);
            };
        }
    }


    public boolean ticketTradeOperator(Map<String, String> params)
    {
        //具體的處理工作
        return resultCode;
    }

    public IETicketTradeOper getTicketTradeOper()
    {
        return ticketTradeOper;
    }
    public void setTicketTradeOper(IETicketTradeOper ticketTradeOper)
    {
        this.ticketTradeOper = ticketTradeOper;
    }

}
 

     TicketTradeOper.mapPool.put(newParams.get("order_id"), newParams);

以上便是整個佇列生產消費的過程，有問題的歡迎交流。