架构师训练营第 7 周作业——性能压测工具

public class Client {
	public static void main(String[] args) {
		int workers = 10;
		PressureContainer pc = new PressureContainer();
    // 10个线程并发请求1000次
		pc.start("https://www.baidu.com", 1000, workers);
	}
}

public class PressureContainer {
	private String name = "worker";
	private ExecutorService executor;
	private double topPart = 0.95; // 95%
	/**
	 * @param url: 压测地址
	 * @param totalRequests： 总请求数
	 * @param workers：并发线程数
	 */
	public void start(String url, int totalRequests, int workers) {
		AtomicInteger totalTimes = new AtomicInteger(totalRequests);
		AtomicInteger failCount = new AtomicInteger();
		AtomicLong costCount = new AtomicLong();
		try {
			// 初始化OkHttp，不然第1次访问会很慢
			OkHttpUtil.get(url);
		} catch (IOException e1) {
			e1.printStackTrace();
		}
		
		int initialCapacity = (int) (totalRequests * (1D - topPart));
    // 存最慢的5%请求耗时
    TopK topK = new TopK(initialCapacity);
		
		CountDownLatch countDownLatch = new CountDownLatch(workers);
		
		executor = new ThreadPoolExecutor(workers, workers, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(),
				new PressureThreadFactory(name));
		for (int i = 0; i < workers; i++) {
			PressureThread pThread = new PressureThread(
					new PressureRunner(url, totalTimes, costCount, failCount, countDownLatch, topK), url);
			executor.execute(pThread);
		}
		try {
			countDownLatch.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		long preCost = costCount.get() / totalRequests;
		System.out.println("Test:[" + url + "] Works:[" + workers + "] call [" + totalRequests + "] times, fail ["
				+ failCount.get() + "] times, pre_cost:[" + preCost + "]ms, " + (int) (topPart * 100) + "% request <= ["
				+ topK.first() + "]ms");
		executor.shutdown();
	}

/**
 * 压测线程工厂
 */
public class PressureThreadFactory implements ThreadFactory {
	public PressureThreadFactory(String name) {
		this.name = name;
	}
	@Override
	public Thread newThread(Runnable r) {
		return new PressureThread(r, name + "-" + created.incrementAndGet());
	}
	private final String name;
	private final AtomicInteger created = new AtomicInteger();
}

/**
 *	压测线程
 */
public class PressureThread extends Thread {
	public PressureThread(Runnable r, String name) {
		super(r, name);
		setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
			@Override
			public void uncaughtException(Thread t, Throwable e) {
				System.err.println("THREAD[" + t.getName() + "] hava uncaughtException");
				e.printStackTrace();
			}
		});
	}
}
public class PressureRunner implements Runnable {
	private String url;
	private AtomicInteger totalTimes;
	private AtomicInteger failCount;
	private AtomicLong costCount;
	private TopK topK;
	private CountDownLatch countDownLatch;
	/**
	 * @param url 
	 * @param totalTimes：执行次数
	 * @param costCount: 总耗时统计
	 * @param failCount: 失败次数统计
	 * @param countDownLatch: 
	 * @param topK 用于存响应时间Topb百分之几的数据
	 */
	public PressureRunner(String url, AtomicInteger totalTimes, AtomicLong costCount, AtomicInteger failCount,
			CountDownLatch countDownLatch, TopK topK) {
		this.url = url;
		this.totalTimes = totalTimes;
		this.costCount = costCount;
		this.failCount = failCount;
		this.countDownLatch = countDownLatch;
		this.topK = topK;
	}
	@Override
	public void run() {
		int times = totalTimes.decrementAndGet();
		while (times >= 0) {
			long startTime = System.currentTimeMillis();
			try {
				String result = OkHttpUtil.get(url);
				if (StringUtils.isEmpty(result)) {
					failCount.incrementAndGet();
				}
			} catch (Exception e) {
				failCount.incrementAndGet();
			}
			long cost = System.currentTimeMillis() - startTime;
			costCount.addAndGet(cost);
			topK.add(cost);
			System.out.println(Thread.currentThread().getName() + " " + times + "--->>> cost:" + cost + "ms");
			try {
				// sleep一下，让出cpu给其它线程执行
				Thread.sleep(50);
			} catch (InterruptedException e) {
			}
			times = totalTimes.decrementAndGet();
		}
		countDownLatch.countDown();
	}
}

/**
 *	http请求工具类
 */
public class OkHttpUtil {
 
	private final static OkHttpClient client = new OkHttpClient().newBuilder().retryOnConnectionFailure(true)
			.connectTimeout(10, TimeUnit.SECONDS).readTimeout(60, TimeUnit.SECONDS).writeTimeout(60, TimeUnit.SECONDS)
			.build();;
 
	public static String get(String url) throws IOException {
		Request request = new Request.Builder().url(url).build();
		try (Response response = client.newCall(request).execute();) {
			if (response.isSuccessful()) {
				return response.body().string();
			} else {
				return null;
			}
		}
	}
 
}

/**
 * 以小顶堆存储部分响应时间最大的数据
 */
public class TopK {
	private int initialCapacity;
	private PriorityQueue<Long> minHeap;
	public TopK(int initialCapacity) {
		this.initialCapacity = initialCapacity;
		// 小顶堆
		minHeap = new PriorityQueue<Long>(initialCapacity, new Comparator<Long>() {
			public int compare(Long i1, Long i2) {
				return (int) (i1 - i2);
			}
		});
	}
	public synchronized void add(Long i) {
		if (initialCapacity > minHeap.size()) {
			minHeap.add(i);
		} else {
			Long top = minHeap.peek();
			// 大数才加入
			if (top < i) {
				minHeap.add(i);
				minHeap.remove();
			}
		}
	}
	public Long first() {
		return minHeap.peek();
	}
}

Test:[https://www.baidu.com] Works:[10] call [1000] times, fail [0] times, pre_cost:[7]ms, 95% request <= [13]ms
Test:[https://www.baidu.com] Works:[10] call [10000] times, fail [0] times, pre_cost:[6]ms, 95% request <= [8]ms

Test:[https://time.geekbang.org/] Works:[10] call [1000] times, fail [0] times, pre_cost:[15]ms, 95% request <= [19]ms
Test:[https://time.geekbang.org/] Works:[10] call [10000] times, fail [0] times, pre_cost:[15]ms, 95% request <= [22]ms