Executors+CyclicBarrier實現的併發測試小例子

        幾年前寫的一個併發測試小工具，分享給你們。比較適合webAPI的測試，已經實現了集合點，響應時間，吞吐量，以及經過失敗的統計，而且實現了能夠實時監控執行進展。 

        裏面用到了 phoenixframework 自動化測試平臺的接口測試模塊phoenix_interface，建議用maven工程引入 phoenix_interface，固然也能夠不引入，這就須要您本身寫數據抓取的方法了，在如下代碼中已經標註了替換的地方。

package org.phoenix.cases.webservice;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.Callable;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

import org.phoenix.api.action.APIAction;
import org.phoenix.api.action.WebAPIAction;

import com.meterware.httpunit.WebResponse;
 
/**
 * Executors+CyclicBarrier實現的併發測試小例子<br>
 * 例子實現了併發測試中使用的集合點，集合點超時時間及思考時間等技術
 * @author mengfeiyang
 *
 */
public class FlushGenerator {
    private static volatile boolean RUNFLAG = true;
    private CyclicBarrier rendzvous;
    private static int threads;
    private static AtomicInteger totalCount = new AtomicInteger();
    private static AtomicInteger startedCount = new AtomicInteger();
    private static AtomicInteger finishCount = new AtomicInteger();
    private static AtomicInteger runCount = new AtomicInteger();
    private static AtomicInteger successCount = new AtomicInteger();
    private static AtomicInteger failCount = new AtomicInteger();
    private String url;
    private long rendzvousWaitTime = 0;
    private long thinkTime = 0;
    private int iteration = 0;
     
    /**
     * 初始值設置
     * @param url 被測url
     * @param threads 總線程數
     * @param iteration 每一個線程迭代次數
     * @param rendzvousWaitTime 集合點超時時間，若是不啓用超時時間，請將此值設置爲0.<br>
     *                            若是不啓用集合點，請將此值設置爲-1<br>
     *                            若是不啓用超時時間，則等待全部的線程所有到達後，纔會繼續往下執行<br>
     * @param thinkTime 思考時間，若是不啓用思考時間，請將此值設置爲0
     */
    public FlushGenerator(String url,int threads,int iteration,long rendzvousWaitTime,long thinkTime){
        totalCount.getAndSet(threads);
        FlushGenerator.threads = threads;
        this.url = url;
        this.iteration = iteration;
        this.rendzvousWaitTime = rendzvousWaitTime;
        this.thinkTime = thinkTime;
    }
 
    public static ThreadCount getThreadCount(){
        return new ThreadCount(threads,runCount.get(),startedCount.get(),finishCount.get(),successCount.get(),failCount.get());
    }
     
    public static boolean isRun(){
        return finishCount.get() != threads;
    }
     
    public synchronized static void stop(){
        RUNFLAG = false;
    }
     
    public void runTask(){
        List<Future<String>> resultList = new ArrayList<Future<String>>();
        ExecutorService exeService = Executors.newFixedThreadPool(threads);
        rendzvous = new CyclicBarrier(threads);//默認加載所有線程
        for(int i=0;i<threads;i++){
            resultList.add(exeService.submit(new TaskThread(i,url,iteration,rendzvousWaitTime,thinkTime)));
        }
        exeService.shutdown();
        for(int j=0;j<resultList.size();j++){
            try{
                System.out.println(resultList.get(j).get());
            }catch(Exception e){
                e.printStackTrace();
            }
        }
        stop();
    }
     
    static class ThreadCount {
        public final int runThreads;
        public final int startedThreads;
        public final int finishedThreads;
        public final int totalThreads;
        public final int successCount;
        public final int failCount;
         
         
        public ThreadCount(int totalThreads,int runThreads, int startedThreads, int finishedThreads,int successCount,int failCount) {
            this.totalThreads = totalThreads;
            this.runThreads = runThreads;
            this.startedThreads = startedThreads;
            this.finishedThreads = finishedThreads;
            this.successCount = successCount;
            this.failCount = failCount;
        }
    }
     
    private class TaskThread implements Callable<String> {
        private String url;
        private long rendzvousWaitTime = 0;
        private long thinkTime = 0;
        private int iteration = 0;
        private int iterCount = 0;
        private int taskId;
         
        /**
         * 任務執行者屬性設置
         * @param taskId 任務id號
         * @param url 被測url
         * @param iteration 迭代次數，若是一直執行則需將此值設置爲0
         * @param rendzvousWaitTime 集合點超時時間，若是不須要設置時間，則將此值設置爲0。若是不須要設置集合點，則將此值設置爲-1
         * @param thinkTime 思考時間，若是不須要設置思考時間，則將此值設置爲0
         */
        public TaskThread(int taskId,String url,int iteration, long rendzvousWaitTime,long thinkTime){
            this.taskId = taskId;
            this.url = url;
            this.rendzvousWaitTime = rendzvousWaitTime;
            this.thinkTime = thinkTime;
            this.iteration = iteration;
        }
        @Override
        public String call() throws Exception{
            startedCount.getAndIncrement();
            runCount.getAndIncrement();
            APIAction webAPI = new WebAPIAction();
            while(RUNFLAG && iterCount<iteration){
                if(iteration != 0)iterCount++;
                try{
                        if(rendzvousWaitTime > 0){
                            try{
                                System.out.println("任務：task-"+taskId+" 已到達集合點...等待其餘線程,集合點等待超時時間爲："+rendzvousWaitTime);
                                rendzvous.await(rendzvousWaitTime,TimeUnit.MILLISECONDS);
                            } catch (InterruptedException e) {
                            } catch (BrokenBarrierException e) {
                                System.out.println("task-"+taskId+" 等待時間已超過集合點超時時間："+rendzvousWaitTime+" ms,將開始執行任務....");
                            } catch (TimeoutException e) {
                            }
                        } else if(rendzvousWaitTime == 0){
                            try{
                                System.out.println("任務：task-"+taskId+" 已到達集合點...等待其餘線程");
                                rendzvous.await();
                            } catch (InterruptedException e) {
                            } catch (BrokenBarrierException e) {
                            }
                        }
                        //須要定製的部分，便可以用您本身的數據包抓取方法替換這段代碼
                    WebResponse wr = webAPI.getResponseByGet(url);
                    //WebResponse wr = new WebConversation().getResponse(new GetMethodWebRequest(url));
                    System.out.println("線程：task-"+taskId+" 獲取到的資源大小："+wr.getText().length()+",狀態碼："+wr.getResponseCode());
                    successCount.getAndIncrement();
                    if(thinkTime !=0){
                        System.out.println("task-"+taskId+" 距下次啓動時間："+thinkTime);
                        Thread.sleep(thinkTime);
                    }
                }catch(Exception e){
                    failCount.getAndIncrement();
                }
            }
            finishCount.getAndIncrement();
            runCount.decrementAndGet();
            return Thread.currentThread().getName()+" 執行完成！";
        }
    }
     
    public static void main(String[] args) {
        new Thread(){
            public void run(){
                new FlushGenerator("http://10.108.79.8:8080/hh.php",5,3,0,200).runTask();
            }
        }.start();
         
        
        //監控執行過程
        new Thread(){
            public void run(){
                while(true){
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("isRun:"+FlushGenerator.isRun());
                    System.out.println("totalThreads:"+FlushGenerator.getThreadCount().totalThreads);
                    System.out.println("startedThreads:"+FlushGenerator.getThreadCount().startedThreads);
                    System.out.println("runThreads:"+FlushGenerator.getThreadCount().runThreads);
                    System.out.println("finishedThread:"+FlushGenerator.getThreadCount().finishedThreads);
                    System.out.println("successCount:"+FlushGenerator.getThreadCount().successCount);
                    System.out.println("failCount:"+FlushGenerator.getThreadCount().failCount);
                    System.out.println();
                }
            }
        }.start();
    }
}