Java基于CountDownLatch的持续并发测试工具

た入场券 2022-01-30 02:21 221阅读 0赞

可以设置并发回合，每回合并发随机次数，持续测试并发。[点这里查看单回合并发测试][Link 1]

import java.util.Random;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicLong;
    
    public abstract class AbstractConcurrentControl { 
    
        public AtomicLong longCounter = new AtomicLong(0);
    
        public static final int DEFAULT_CONCURRENT_CONTROL = 160;
    
        public Random random = new Random();
    
        private int times;
    
        private CountDownLatch latch;
    
        private CountDownLatch stopLatch;
    
        private ThreadPoolExecutor threadPool;
    
        private BlockingQueue blockingQueue;
    
        /** * 单位：ms */
        private int interval;
    
        public AbstractConcurrentControl(int times, int interval) { 
            this.times = times;
            this.interval = interval;
            blockingQueue = new LinkedBlockingQueue<Runnable>();
            threadPool = new ThreadPoolExecutor(DEFAULT_CONCURRENT_CONTROL, DEFAULT_CONCURRENT_CONTROL, 100000, TimeUnit.MILLISECONDS, blockingQueue);
        }
    
        public AbstractConcurrentControl(int times) { 
            this(times, 1000);
        }
    
        /** * 并发执行线程 * * @Title: process * @date 2018年12月26日 上午11:19:20 * @author yz */
        public final void process() { 
    
            // 总计并发执行的执行次数
            int exeTotal = 0;
    
            // 持续并发次数
            int i = 0;
    
            int[] perTimes = new int[times];
    
            for (int j = 0; j < times; j++) { 
                perTimes[j] = random.nextInt(DEFAULT_CONCURRENT_CONTROL);
                exeTotal = exeTotal + perTimes[j];
            }
            stopLatch = new CountDownLatch(exeTotal);
    
            while ((times--) > 0) { 
                int perTime = perTimes[i];
                i++;
                System.out.println("当前次数： " + (i) + " 剩余次数： " + (times) + " ， 随机推送次数：" + perTime);
                latch = new CountDownLatch(perTime);
                for (int j = 0; j < perTime; j++) { 
                    threadPool.submit(new Runnable() { 
                        @Override
                        public void run() { 
                            try { 
                                latch.await();
                                concurrentCode();
                                stopLatch.countDown();
                            } catch (InterruptedException e) { 
                                e.printStackTrace();
                            }
                        }
                    });
                    latch.countDown();
                }
    
                try { 
                    TimeUnit.MILLISECONDS.sleep(interval);
                } catch (InterruptedException e) { 
    
                }
            }
            System.out.println("总次数： " + exeTotal);
            blockingMainThread();
        }
    
        /** * 并发代码 * * @Title: concurrentCode * @date 2018年12月26日 下午2:05:25 * @author yz */
        protected abstract void concurrentCode();
    
        /** * 并发数据 * * @return * @Title: encapsulatingData * @date 2018年12月26日 下午2:06:14 * @author yz */
        protected <T> T encapsulatingData() { 
            return null;
        }
    
        /** * 阻塞主线程，防止JVM关闭，不建议使用Xxx.class.wait，可以使用TimeUnit.Seconds.sleep(200); * 如果使用TimeUnit.Seconds.sleep(200)，可能会出现异常，因为JVM已经关闭，而测试的线程可能没有执行完成 * * @Title: blockingMainThread * @date 2018年12月26日 下午6:55:03 * @author yz */
        protected void blockingMainThread() { 
            if (this.threadPool == null) { 
                return;
            }
            try { 
                stopLatch.await();
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } finally { 
                System.out.println("关闭线程池");
                this.threadPool.shutdown();
            }
        }
    }

public class HttpPushService { 
        public void delayBatchPushJson(String a, Integer b, Integer data) { 
            System.out.println(a + " " + b + " " + data);
        }
    }

import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.Random;
    
    public class HttpPushTest extends AbstractConcurrentControl { 
    
        static Random random = new Random();
    
        static List<Integer> appId = new ArrayList<>();
        static Map<Integer, Integer> pushTypes = new HashMap<>();
    
        static { 
            appId.add(12301);
            appId.add(12302);
            appId.add(12303);
            pushTypes.put(12301, 0);
            pushTypes.put(12302, 0);
            pushTypes.put(12303, 0);
        }
    
        public HttpPushTest() { 
            super(3,5000);
        }
    
        @Override
        protected void concurrentCode() { 
            try { 
                List<String> data = encapsulatingData();
                httpPushService.delayBatchPushJson(data.get(0), Integer.valueOf(data.get(1)), Integer.valueOf(data.get(2)));
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    
        @Override
        protected <T> T encapsulatingData() { 
            List<String> data = new ArrayList<>(3);
            String json = "{\"encapsulatingData\":" + longCounter.getAndIncrement() + "" + "}";
            data.add(json);
            int i = random.nextInt(3);
            data.add(appId.get(i) + "");
            data.add(pushTypes.get(appId.get(i)) + "");
            return (T) data;
        }
    
        private HttpPushService httpPushService = new HttpPushService();
    
        public static void main(String[] args) { 
            HttpPushTest httpPushTest = new HttpPushTest();
            httpPushTest.process();
        }
    
    }

[Link 1]: https://blog.csdn.net/qq_30038111/article/details/89229330