在实际的项目开发中,很多情况下我们需要用到线程类来控制程序并发进行以提高效率,但同时也会涉及到线程安全问题。
简单的线程控制我们可以使用Executors类来简单实现,具体可参考
文章1: Java ScheduledThreadPoolExecutor延迟或周期性执行任务
文章2:使用Executors和ThreadPoolExecutor2 使用了队列
文章3:ForkJoin-任务分拆合并处理器 JDK 7中的新功能,可以将大任务拆分处理,再将结果合并
但是一些涉及业务比较多的线程控制仍然需要传统的方法实现,下面代码的背景:有大量的请求访问,请求参数有很多重复,加入缓存机制,保证线程安全
我们通过service类和Worker类来实现,具体代码如下:
package itour.cn.fare.gateway.service; import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class ObeQueryServiceImpl implements ObeQueryService { private static Logger log = LoggerFactory.getLogger(ObeQueryServiceImpl.class); private static ObeQueryServiceImpl service = new ObeQueryServiceImpl(); private Map<String, Integer> avhTask = new HashMap<String, Integer>();// 线程执行任务表,保证同样的任务只执行一次 private Map<String, Object> avhLock = new HashMap<String, Object>();// 锁表,保证同样的请求锁定状态下只进入一个 private void getAvh(String key, String org, String dis, String date, String airCompany) { synchronized (this) { if (avhLock.get(key) == null) {// 添加锁,同样的请求只添加一次 avhLock.put(key, new byte[0]); log.debug(String.format("AV create lock:%s ", key)); } } synchronized (avhLock.get(key)) { // 添加任务,保证同样的请求只执行一次 if (avhTask.get(key) == null) { avhTask.put(key, new Integer(1)); ObeAvhWorker worker = new ObeAvhWorker(key, org, dis, date, airCompany); log.debug(String.format("AV[%s] AvWorker Start ", key)); worker.start(); } try { log.debug(String.format("AV[%s] Waitting AvWorker ", key)); avhLock.get(key).wait();// 等待run方法执行完 } catch (InterruptedException e) { log.error(e.getMessage(), e); } log.debug(String.format("AV[%s] AvWorker Notified", key)); } } public Map<String, Integer> getAvhTask() { return avhTask; } public Map<String, Object> getAvhLock() { return avhLock; } }
package itour.cn.fare.gateway; import itour.cn.fare.gateway.service.ObeQueryServiceImpl; import java.io.Serializable; import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.travelsky.sbeclient.obe.exceptions.ObeException; import com.travelsky.sbeclient.obe.response.AVResponse; import com.travelsky.sbeclient.obe.response.AvItem; public class ObeAvhWorker extends Thread { private static Logger log = LoggerFactory.getLogger(ObeAvhWorker.class); private ObeQueryServiceImpl service = ObeQueryServiceImpl.getInstance(); private String key; private String org; private String dis; private String date; private String airCompany; public ObeAvhWorker(String key, String org, String dis, String date, String airCompany) { this.key = key; this.org = org; this.dis = dis; this.date = date; this.airCompany = airCompany; } @Override public void run() { log.debug("ObeAvhWorker Thread Start for " + key); try { worker(); }catch (Exception e) { log.error(e.getMessage(), e); } finally { synchronized (service.getAvhLock().get(key)) { service.getAvhLock().get(key).notifyAll(); } service.getAvhTask().remove(key); } } private void worker() throws Exception { ObeSession session = new ObeSession(); AVResponse avr = session.avh(org, dis, date, airCompany); List<AvItem> avItems = avr.getAvItems(); // 存入缓存 if (avItems != null && avItems.size() > 0) { if (log.isDebugEnabled()) log.info(String.format("Avh[%s] put into cachePool total %d", key, avItems.size())); CachePoolManager.getInstance().putAvCV(key, (Serializable) avItems); } } }
原文:http://blog.csdn.net/kuyuyingzi/article/details/18600161