一个关于同步接口并发控制的简单demo

名字是乱打的

发布于 2022-10-28 11:20:05

2460

发布于 2022-10-28 11:20:05

文章被收录于专栏：软件工程软件工程

背景： 1.有个批量关单关单并且同步返回结果的需求

1.分析

需求

1.批量关单，且需要同步返回结果由于调用我方接口是一次传入很多任务，假设是N，串行关单必然超时，因此这里需要并发执行
2.由于公司不允许随便定义线程池（即使比较小的池）因此不能用线程池做并发控制，需要自定义线程控制
3.由于下游rpc给到我们这边的流量上限是有限的，假设是X,因此我们需要并发，但是又需要控制同时并发的数量

2.写代码

方案：线程池（实现并发）+CountDownLatch（实现阻塞主线程，并发结束返回结果）+Semaphore （实现并发控制）

需要注意的一点是 等待队列容量>=2*Semaphore，不然会有线程因为拿不到线程池资源不处理直接失败的（原因参考线程池执行流程），当然我们可以将这部分压根没处理的也一并同步返回出去，让调用侧重试（我方不重试，防止超时问题）

public class test {

    /**
     * 线程池
     * 等待队列容量>=2*Semaphore
     */
    static ExecutorService threadPool = new ThreadPoolExecutor(
            5,
            5,
            1,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(10),
            Executors.defaultThreadFactory()
            , new ThreadPoolExecutor.AbortPolicy());


    public static void main(String[] args) throws InterruptedException {

        List<Integer> successList=new CopyOnWriteArrayList<>();
        List<Integer> failList=new CopyOnWriteArrayList<>();
        List<Integer> errorList=new CopyOnWriteArrayList<>();
        
        CountDownLatch countDownLatch = new CountDownLatch(100);
        Semaphore semaphore=new Semaphore(10);
        for (int i = 0; i < 100; i++) {

            try {
                AtomicInteger self = new AtomicInteger(i);
                threadPool.execute(() -> {
                    try {
                        if ((self.get() & 1) == 1) {
                            int temp = (self.get()/0);
                        }
                        Thread.sleep(5000);
                        successList.add(self.get());
                        System.out.println(self.get() + "执行成功");
                    }catch (Exception e){
                        System.out.println(self.get() + "执行失败");
                        failList.add(self.get());
                    }finally {
                        semaphore.release();
                        countDownLatch.countDown();
                    }
                });
            } catch (Exception e) {
                errorList.add(i);
                System.out.println("biz  Exception");
                semaphore.release();
                countDownLatch.countDown();
            }
        }
        countDownLatch.await();
        threadPool.shutdown();
        System.out.println("成功的："+successList.size());
        successList.stream().sorted().forEach(item->{
            System.out.print(item+" ");
        });

        System.out.println();
        
        System.out.println("失败的："+failList.size());
        failList.stream().sorted().forEach(item->{
            System.out.print(item+" ");
        });
        System.out.println();

        System.out.println("错误的："+errorList.size());
        errorList.stream().sorted().forEach(item->{
            System.out.print(item+" ");
        });
        
        System.out.println("全部执行结束");
        threadPool.shutdown();
        
    }

}

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2022-10-14，如有侵权请联系 cloudcommunity@tencent.com 删除

rpc