多线程合并查询

package com.anyec.zk;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;

public class AsyncQueFutherTask {
	Logger logger=LoggerFactory.getLogger(AsyncQueFutherTask.class);
	Random r = new Random();
	private static BlockingQueue<Long> bq = new LinkedBlockingQueue<Long>(20);
//	private final static ThreadPoolExecutor ses = new ThreadPoolExecutor(20, 40, 1L, TimeUnit.MINUTES,
//			new ArrayBlockingQueue<Runnable>(100), new ThreadPoolExecutor.DiscardPolicy());
//	ExecutorService ses =new ThreadPoolExecutor(20, 40, 1L, TimeUnit.MINUTES,
//			new ArrayBlockingQueue<Runnable>(1), new ThreadPoolExecutor.DiscardOldestPolicy());
	ExecutorService ses=Executors.newCachedThreadPool();
//	ExecutorService ses=Executors.newFixedThreadPool(20);
//	ExecutorService ses=new ThreadPoolExecutor(0, 200,
//            60L, TimeUnit.SECONDS,
//            new SynchronousQueue<Runnable>());
	static ConcurrentMap<Long, FutureTask2<JSONObject>> orderFutureMap = new ConcurrentHashMap<Long, FutureTask2<JSONObject>>(
			64, 0.75f, 1);
	Timer timer = new Timer();
	static int count = 0;

	public void init() {
		timer.schedule(new TimerTask() {
			@Override
			public void run() {
				count++;
				// logger.info("第"+count+"次开始执行");
				Set<Long> orderIds = new HashSet<>();
//				Iterator<Long> it = bq.iterator();
//				while (!bq.isEmpty()) {
//					orderIds.add(bq.poll());
//				}
				bq.drainTo(orderIds,100);
				Map<Long, JSONObject> rmap = mapByIds(orderIds);
				for (Long orderId : orderIds) {
					FutureTask2<JSONObject> f = orderFutureMap.get(orderId);
					JSONObject jo = rmap.get(orderId);
					if (f != null) {
						f.set(jo);
					} else {
						logger.info(orderId + " FutureTask2 is null");
					}
				}
				rmap.clear();
				rmap = null;
				// logger.info("第"+count+"次执行结束"+orderIds.size());
			}
		}, 10 * 1000, 100);
	}

	public Map<Long, JSONObject> mapByIds(Set<Long> orderIds) {
		Map<Long, JSONObject> map = new HashMap<>();

		for (Long orderId : orderIds) {
			JSONObject jo = new JSONObject();
			jo.put("orderId", orderId);
			jo.put("orderNum", r.nextInt(200));
			map.put(orderId, jo);
		}
		return map;
	}

	public JSONObject queryOrder(Long orderId) {
		FutureTask2<JSONObject> ft = new FutureTask2<JSONObject>(new Callable<JSONObject>() {
			@Override
			public JSONObject call() throws Exception {
				JSONObject jo = orderFutureMap.get(orderId).get();
				orderFutureMap.remove(orderId);
				return jo;
			}
		});
		try {
			orderFutureMap.put(orderId, ft);
			bq.put(orderId);
			
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
		ft.setOrderId(orderId);
		Future<JSONObject> f = (Future<JSONObject>) ses.submit(ft);
		JSONObject jo = null;
		try {
			jo = ft.get();
			orderFutureMap.remove(orderId);
		} catch (InterruptedException | ExecutionException e) {
			e.printStackTrace();

		}
		if (jo == null) {
			jo = new JSONObject();
		}
		logger.info(orderId + " 执行结果 " + jo.toJSONString());
		return jo;
	}

	public static void main(String[] args) {
		int i = 1;
		AsyncQueFutherTask task = new AsyncQueFutherTask();
		task.init();
		try {
			Thread.sleep(10 * 1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		while (i < 100000) {
			int j = ++i;
			new Thread(new Runnable() {
				@Override
				public void run() {
					task.queryOrder(new Long(j));
				}
			}).start();
		}
	}
}

class FutureTask2<T> extends FutureTask<T> {
	private Long orderId;

	public FutureTask2(Callable<T> callable) {
		super(callable);
	}

	public FutureTask2(Runnable runnable, T result) {
		super(runnable, result);
	}

	public Long getOrderId() {
		return orderId;
	}

	public void setOrderId(Long orderId) {
		this.orderId = orderId;
	}

	public void set(T v) {
		super.set(v);
	}
}

class Param {

}