遇到問題：

最近做微信支付,專案上線一陣,發現一個問題。有一條訂單流水居然在資料庫的出現兩次。這個問題非常嚴重。

檢視微信回撥系統的介面程式碼發現程式碼是沒錯的(正常情況下),而這次遇到非正常情況了

原因：微信支付成功後回撥我們系統介面在極短時間回調了2次,微信官方文件說明了,是最短15s回撥一次。

前幾天微信支付抽風了,可能業務出現了波動。

簡單來說就是在併發情況下沒有做資料唯一性處理，不管怎麼樣這類併發情況都是有必要的處理。

解決方式：使用執行緒池+佇列

專案基於Spring,如果不用spring需要自己把

ThreadPoolManager.java

改成單例模式

1.寫一個Controller(Spring mvc)

/**
 * @author HeyS1
 * @date 2016/12/1
 * @description
 */
@Controller
public class ThreadPoolController {
    @Autowired
    ThreadPoolManager tpm;

    @RequestMapping("/pool")
    public
    @ResponseBody
    Object test() {
        for (int i = 0; i < 500; i++) {
            //模擬併發500條記錄
            tpm.processOrders
 
(Integer.toString(i));
        }

        return "ok";
    }
}

2.執行緒池管理

/**
 * @author HeyS1
 * @date 2016/12/1
 * @description threadPool訂單執行緒池, 處理訂單
 * scheduler 排程執行緒池 用於處理訂單執行緒池由於超出執行緒範圍和佇列容量而不能處理的訂單
 */
@Component
public class ThreadPoolManager implements BeanFactoryAware {
    private static Logger log = LoggerFactory.getLogger(ThreadPoolManager.class);
    private
 
 BeanFactory factory;//用於從IOC裡取物件
    // 執行緒池維護執行緒的最少數量
    private final static int CORE_POOL_SIZE = 2;
    // 執行緒池維護執行緒的最大數量
    private final static int MAX_POOL_SIZE = 10;
    // 執行緒池維護執行緒所允許的空閒時間
    private final static int KEEP_ALIVE_TIME = 0;
    // 執行緒池所使用的緩衝佇列大小
    private final static int WORK_QUEUE_SIZE = 50;
    // 訊息緩衝佇列
    Queue<Object> msgQueue = new LinkedList<Object>();

    //用於儲存在佇列中的訂單,防止重複提交
    Map<String, Object> cacheMap = new ConcurrentHashMap<>();

    //由於超出執行緒範圍和佇列容量而使執行被阻塞時所使用的處理程式
    final RejectedExecutionHandler handler = new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            //System.out.println("太忙了,把該訂單交給排程執行緒池逐一處理" + ((DBThread) r).getMsg());
            msgQueue.offer(((DBThread) r).getMsg());
        }
    };

    // 訂單執行緒池
    final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
            CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME,
            TimeUnit.SECONDS, new ArrayBlockingQueue(WORK_QUEUE_SIZE), this.handler);

    // 排程執行緒池。此執行緒池支援定時以及週期性執行任務的需求。
    final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5);

    // 訪問訊息快取的排程執行緒,每秒執行一次
    // 檢視是否有待定請求，如果有，則建立一個新的AccessDBThread，並新增到執行緒池中
    final ScheduledFuture taskHandler = scheduler.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            if (!msgQueue.isEmpty()) {
                if (threadPool.getQueue().size() < WORK_QUEUE_SIZE) {
                    System.out.print("排程：");
                    String orderId = (String) msgQueue.poll();
                    DBThread accessDBThread = (DBThread) factory.getBean("dBThread");
                    accessDBThread.setMsg(orderId);
                    threadPool.execute(accessDBThread);
                }
                // while (msgQueue.peek() != null) {
                // }
            }
        }
    }, 0, 1, TimeUnit.SECONDS);

    //終止訂單執行緒池+排程執行緒池
    public void shutdown() {
        //true表示如果定時任務在執行，立即中止，false則等待任務結束後再停止
        System.out.println(taskHandler.cancel(false));
        scheduler.shutdown();
        threadPool.shutdown();
    }

    public Queue<Object> getMsgQueue() {
        return msgQueue;
    }


    //將任務加入訂單執行緒池
    public void processOrders(String orderId) {
        if (cacheMap.get(orderId) == null) {
            cacheMap.put(orderId,new Object());
            DBThread accessDBThread = (DBThread) factory.getBean("dBThread");
            accessDBThread.setMsg(orderId);
            threadPool.execute(accessDBThread);
        }
    }


    //BeanFactoryAware
    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        factory = beanFactory;
    }
}

3.執行緒池中工作的執行緒

//執行緒池中工作的執行緒
@Component
@Scope("prototype")//spring 多例
public class DBThread implements Runnable {
    private String msg;
    private Logger log = LoggerFactory.getLogger(DBThread.class);

    @Autowired
    SystemLogService systemLogService;



    @Override
    public void run() {
        //模擬在資料庫插入資料
        Systemlog systemlog = new Systemlog();
        systemlog.setTime(new Date());
        systemlog.setLogdescribe(msg);
        //systemLogService.insert(systemlog);
        log.info("insert->" + msg);
    }


    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }
}

瀏覽器輸入地址127.0.0.1/pool

幾秒後關閉tomcat。

模擬500條資料，訂單執行緒池處理了117條。排程執行緒池處理5條

關閉tomcat，後還有378條未處理（這裡的實現需要用到spring監聽器）。加起來一共500

OK。完畢

spring監聽器，監聽tomcat關閉事件：

public class MyApplicationListener implements ApplicationListener<ApplicationEvent> {

    @Autowired
    ThreadPoolManager threadPoolManager;

    @Override
    public void onApplicationEvent(ApplicationEvent event) {

        if (event instanceof ContextClosedEvent) {
            XmlWebApplicationContext x = (XmlWebApplicationContext) event.getSource();
            //防止執行兩次。root application context 沒有parent，他就是老大
            if (x.getDisplayName().equals("Root WebApplicationContext")) {
                threadPoolManager.shutdown();
                Queue q = threadPoolManager.getMsgQueue();
                System.out.println("關閉了伺服器，還有未處理的資訊條數：" + q.size());
            }


        } else if (event instanceof ContextRefreshedEvent) {
//            System.out.println(event.getClass().getSimpleName()+" 事件已發生！");
        } else if (event instanceof ContextStartedEvent) {
//            System.out.println(event.getClass().getSimpleName()+" 事件已發生！");
        } else if (event instanceof ContextStoppedEvent) {
//            System.out.println(event.getClass().getSimpleName()+" 事件已發生！");
        } else {
//            System.out.println("有其它事件發生:"+event.getClass().getName());
        }
    }
}

spring配置一下

<bean id="springStartListener" class="com.temp.MyApplicationListener"></bean>