隊列送券的實際應用--ConcurrentLinkedQueue並發隊列

1、TicketQueue.java--隊列封裝類，負責如下職責：
a、把活動登記對象放入隊列中
b、從隊列中獲取活動登記對象，並派券

package com.datong.pear.ticket;

import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.commons.lang.math.NumberUtils;
import org.apache.log4j.Logger;

import com.datong.pear.system.common.Constants;
import com.datong.pear.system.common.Result;

/**
 * 活動登記隊列
 *
 * @author jqlin
 */
public class TicketQueue {
    private static ConcurrentLinkedQueue<ActivityRecordModel> linkedQueue = new ConcurrentLinkedQueue<ActivityRecordModel>();
    
    private static final Logger logger = Logger.getLogger(TicketQueue.class);
    
    public static boolean isRunning = false;
    
    /**
     * 活動登記對象放入隊列中
     * 
     * @param activityRecord
     * @author jqlin
     */
    public static void offer(ActivityRecordModel activityRecord) {
        if(activityRecord == null){
            logger.info(String.format("%s activityRecord is null", TicketQueue.class.getName()));
            return;
        }
        
        if(NumberUtils.toLong(activityRecord.getId(), 0) == 0L){
            logger.info(String.format("%s activityRecord.id is illegal，activityRecord.id=%s",
                    TicketQueue.class.getName(), activityRecord.getId()));
            return;
        }
        
        logger.info(String.format("%s 即將放入隊列的活動登記信息：%s", TicketQueue.class.getName(), activityRecord));
        if(activityRecord.getState() != Constants.ActivityRecordStatus.WPF){
            logger.info(String.format("%s activityRecordId=%s 狀態不是未派發，無法放入隊列", TicketQueue.class.getName(), activityRecord.getId()));
            return;
        }
        
        logger.info(String.format("%s activityRecordId=%s 准備放入隊列", TicketQueue.class.getName(), activityRecord.getId()));
        linkedQueue.offer(activityRecord);
        logger.info(String.format("----setActivityRecordId=%s *****", activityRecord.getId()));
        logger.info(String.format("%s activityRecordId=%s 放入隊列成功", TicketQueue.class.getName(), activityRecord.getId()));
        logger.info("**********************************************************************");
    }

    /**
     * 從隊列中獲取活動登記對象，並派券
     * 
     * @return
     * @author jqlin
     */
    public static synchronized void pollAndSendTicket(TicketService ticketService) {
        isRunning = true;
        
        logger.info("准備從隊列中獲取活動登記對象，並派券...");
        if(linkedQueue != null && !linkedQueue.isEmpty()){ 
            ActivityRecordModel activityRecordModel = null;
            while (true) {
                Iterator<ActivityRecordModel> arIterator = linkedQueue.iterator();
                if(!arIterator.hasNext()){
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        logger.info("隊列派券異常中斷");
                    }
                    
                    continue;
                } 
                
                activityRecordModel = arIterator.next();
                logger.info(String.format("----getActivityRecordId=%s， Thread=%s *****", activityRecordModel.getId(), Thread.currentThread().getId()));
                logger.info(String.format("%s 從隊列中獲取的活動登記信息：%s", TicketQueue.class.getName(), activityRecordModel));
                
                logger.info(String.format("%s activityRecordId=%s 准備移出隊列", TicketQueue.class.getName(), activityRecordModel.getId()));
                linkedQueue.remove(activityRecordModel);
                logger.info(String.format("%s activityRecordId=%s 移出隊列成功", TicketQueue.class.getName(), activityRecordModel.getId()));
                
                logger.info(String.format("%s activityRecordId=%s 隊列派券開始", TicketQueue.class.getName(), activityRecordModel.getId()));
                Result result = ticketService.sendTicket(activityRecordModel);
                logger.info(String.format("%s 活動登記派券接口返回信息：", TicketQueue.class.getName()));
                logger.info(result);
                logger.info(String.format("%s 活動登記派券結束", TicketQueue.class.getName()));
                
                logger.info("**********************************************************************");
            }
        } else {
            logger.info("隊列中活動登記對象為空，沒有數據可派券...");
        }
        
        isRunning = false;
    }
}

 

2、啟動線程執行隊列

    /**
     * 通過隊列派券
     * 
     * @param ar
     * @param ticketService
     * @author jqlin
     */
    private synchronized void sendTicketByQueue(ActivityRecordModel ar, final TicketService ticketService) {
        final ActivityRecordModel activityRecord = ar;
        TicketQueue.offer(activityRecord);
        //隊列為空，啟動一個線程
        if(!TicketQueue.isRunning) {
            new Thread(new Runnable(){
                @Override
                public void run() {
                    logger.info("啟動一個線程，放入隊列派券...");
                    TicketQueue.pollAndSendTicket(ticketService);
                }
            }).start();
 
        } 
    }