并发/异步代码实现的四种方式（ThreadPool ，Callable，CompletableFuture，CompletableFuture+阻塞队列+定时任务线程池）

ThreadPool##############################################

package com.csw.shuanfa.CodeImprove.CompleteFutureLinkedBlockingQueue;

import com.csw.shuanfa.utils.ThreadPoolUtil;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * @author chengshuaiwei
 */
public class ThreadPoolTest {

    private static final int num = 1000;

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

        ExecutorService threadPool = ThreadPoolUtil.getExecutorServiceCPU();

        List<CompletableFuture> completableFutureList = new ArrayList<>();
        //1、
        try {
            int mm = 1 / 0;
            long t1 = System.currentTimeMillis();
            AtomicInteger a = new AtomicInteger();
            CountDownLatch countDownLatch = new CountDownLatch(num);

            for (int i = 0; i < num; i++) {
                threadPool.execute(() -> {
                    //每个步骤都用调用线程池里面的线程来做
                    try {
                        Thread.sleep(10);
                        //int aa=1/0;
                    } catch (InterruptedException e) {
                        //2、错误时打印信息
                        e.printStackTrace();
                    } finally {
                        a.getAndIncrement();
                        countDownLatch.countDown();
                    }
                });
            }

            //2、保证上面每一个异步执行完，设置超时获取时间，跑批不需要
            boolean await = countDownLatch.await(3, TimeUnit.MILLISECONDS);


            CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
            }, threadPool);
            completableFutureList.add(future1);

            //批量阻塞完成
            // CompletableFuture.allOf(futureList.toArray(futureList.toArray(new CompletableFuture[0]))).get(1, TimeUnit.SECONDS);
            //CompletableFuture.allOf(future1, future2, future3, future4, future5, future6, future7).get(1, TimeUnit.SECONDS);
            long t2 = System.currentTimeMillis();
            System.out.println("总共耗时：" + (t2 - t1));
        } catch (Exception e) {
            //3、
            System.out.println(ThreadPoolUtil.getThreadPoolNameMethod());
            threadPool.shutdownNow();
            throw new RuntimeException(e);
        } finally {
            threadPool.shutdown();
        }
    }


}

Callable_1###############################################

package com.csw.shuanfa.CodeImprove.CompleteFutureLinkedBlockingQueue;

import com.csw.shuanfa.utils.ThreadPoolUtil;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

public class CallableTest {
    public static void main(String[] args) throws InterruptedException,
            ExecutionException {
        int num = 10;
        ExecutorService executorServiceCPU = ThreadPoolUtil.getExecutorServiceCPU();

        try {
            AtomicInteger a = new AtomicInteger();
            long t1 = System.currentTimeMillis();
            Callable<Object> call = getObjectCallable(a, executorServiceCPU);

            List<Future<Object>> mathContexts = new ArrayList<>();
            for (int i = 0; i < num; i++) {
                //调用每个步骤
                Future<Object> future = executorServiceCPU.submit(call);
                //封装future
                mathContexts.add(future);
            }
            //后续遍历获取
            //异步处理，每个步骤执行完获取
            for (int i = 0; i < mathContexts.size(); i++) {
                Future<Object> future = mathContexts.get(i);
                //设置超时获取时间
                Object obj = future.get(3000 * 1, TimeUnit.MILLISECONDS);
            }
            long t2 = System.currentTimeMillis();

            //t2 - t1);
            //a.get());
        } catch (TimeoutException ex) {
            //处理超时啦....");
            ex.printStackTrace();
        } catch (Exception e) {
            //处理失败.");
            e.printStackTrace();
        }

    }

    private static Callable<Object> getObjectCallable(AtomicInteger a, ExecutorService singleThreadPool) {
        Callable<Object> call = new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                //开始执行耗时操作
                Thread.sleep(2000);//每个任务的处理时间
                a.getAndIncrement();
                //Thread.currentThread().getName());
                return null;
            }
        };
        return call;
    }

}

Callable_2##############################################

package com.csw.shuanfa.CodeImprove.CompleteFutureLinkedBlockingQueue;

import com.csw.shuanfa.utils.ThreadPoolUtil;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

public class CallableTestShiZhan {
    public static void main(String[] args) throws InterruptedException,
            ExecutionException {
        int num = 10;
        ExecutorService threadPool = ThreadPoolUtil.getExecutorServiceCPU();

        try {
            long t1 = System.currentTimeMillis();
            AtomicInteger a = new AtomicInteger();
            Callable<Object> call = getObjectCallable(a, threadPool);
            //调用每个步骤
            Future<Object> future1 = threadPool.submit(call);
            Future<Object> future2 = threadPool.submit(call);
            Future<Object> future3 = threadPool.submit(call);


            Object obj1 = future1.get(3000 * 1, TimeUnit.MILLISECONDS);
            Object obj2 = future2.get(3000 * 1, TimeUnit.MILLISECONDS);
            Object obj3 = future3.get(3000 * 1, TimeUnit.MILLISECONDS);

            long t2 = System.currentTimeMillis();
            //t2 - t1);
            //a.get());
        } catch (TimeoutException ex) {
            //处理超时啦....");
            ex.printStackTrace();
        } catch (Exception e) {
            //处理失败.");
            e.printStackTrace();
        }
    }

    private static Callable<Object> getObjectCallable(AtomicInteger a, ExecutorService singleThreadPool) {
        Callable<Object> call = new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                //开始执行耗时操作
                Thread.sleep(2000);//每个任务的处理时间
                a.getAndIncrement();
                //Thread.currentThread().getName());
                return null;
            }
        };
        return call;
    }

}

CompletableFuture########################################

其他相关方法https://www.jianshu.com/p/8e4a21cb1ba5

package com.csw.shuanfa.CodeImprove.CompleteFutureLinkedBlockingQueue;

import com.csw.shuanfa.utils.ThreadPoolUtil;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author chengshuaiwei
 */
public class CompletableFutureTestShiZhan {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1、
        ExecutorService threadPool = ThreadPoolUtil.getExecutorServiceCPU();

        long t1 = System.currentTimeMillis();
        AtomicInteger a = new AtomicInteger();

        try {
            CompletableFuture<Long> future1 = CompletableFuture.supplyAsync(() -> {
                return doSomthing(a);
            }, threadPool);
            CompletableFuture<Long> future2 = CompletableFuture.supplyAsync(() -> {
                return doSomthing(a);
            }, threadPool);
            CompletableFuture<Long> future3 = CompletableFuture.supplyAsync(() -> {
                try {
                    int i = 1 / 0;
                } catch (Exception e) {
                    System.out.println(Thread.currentThread().getName());
                    e.printStackTrace();
                    throw new RuntimeException(e);
                }
                return doSomthing(a);
            }, threadPool);

            //2、
            long time1 = future1.get(1, TimeUnit.SECONDS);
            long time2 = future2.get();
            long time3 = future3.get();

            long t2 = System.currentTimeMillis();
            System.out.println("总共耗时：" + (t2 - t1));
        } catch (Exception e) {
            //3、
            System.out.println(ThreadPoolUtil.getThreadPoolNameMethod());
            threadPool.shutdown();

            throw new RuntimeException(e);
        }

    }

    private static long doSomthing(AtomicInteger a) {
        a.getAndIncrement();
        try {
            TimeUnit.SECONDS.sleep(2);
            //Thread.currentThread().getName());
        } catch (Exception e) {
            e.getMessage();
        }
        return System.currentTimeMillis();
    }


//    /**
//     *
//     * @throws Exception
//     * 无返回值
//     */
//    public static void runAsync() throws Exception {
//        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(1);
//            } catch (InterruptedException e) {
//            }
//            //run end ...");
//        });
//
//        future.get();
//    }
//
//    /**
//     *
//     * @throws Exception
//     * 有返回值
//     */
//    public static void supplyAsync() throws Exception {
//        CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {
//            try {
//                TimeUnit.SECONDS.sleep(1);
//            } catch (InterruptedException e) {
//            }
//            //run end ...");
//            return System.currentTimeMillis();
//        });
//
//        long time = future.get();
//        //time = " + time);
//    }
}

第四种########################################################

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class RequestMergeTest {
    /**
     * LinkedBlockingQueue是一个阻塞的队列,内部采用链表的结果,通过两个ReenTrantLock来保证线程安全
     * LinkedBlockingQueue与ArrayBlockingQueue的区别
     * ArrayBlockingQueue默认指定了长度,而LinkedBlockingQueue的默认长度是Integer.MAX_VALUE,也就是无界队列,在移除的速度小于添加的速度时，容易造成OOM。
     * ArrayBlockingQueue的存储容器是数组,而LinkedBlockingQueue是存储容器是链表
     * 两者的实现队列添加或移除的锁不一样，ArrayBlockingQueue实现的队列中的锁是没有分离的，即添加操作和移除操作采用的同一个ReenterLock锁，而LinkedBlockingQueue实现的队列中的锁是分离的，其添加采用的是putLock，移除采用的则是takeLock，这样能大大提高队列的吞吐量，也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。
     */
    static LinkedBlockingQueue<Request1> queue = new LinkedBlockingQueue();   //这里因为是测试,所以使用的是无界队列

    public static void main(String[] args) throws InterruptedException {
        int num = 100;
        init();
        //CountDownLatch来让主线程等待
        CountDownLatch countDownLatch = new CountDownLatch(num);
        for (int i = 0; i < num; i++) {
            //模拟0.1秒一个请求
            TimeUnit.MILLISECONDS.sleep(100);
            final String code = "code" + i;
            //模拟客户端瞬间涌入情况，不用线程池
            Thread thread = new Thread(() -> {
                try {
                    Map<String, Object> map = queryCommodity("000" + code);
                    System.out.println(Thread.currentThread().getName() + "的查询结果是:" + map);
                } catch (Exception e) {
                    System.out.println(Thread.currentThread().getName() + "出现异常:" + e.getMessage());
                    e.printStackTrace();
                }
                countDownLatch.countDown();
            });
            thread.setName("price-thread-" + code);
            thread.start();
        }
        countDownLatch.await();
    }

    public static Map<String, Object> queryCommodity(String code) throws ExecutionException, InterruptedException {
        Request1 request = new Request1();
        request.code = code;
        CompletableFuture<Map<String, Object>> future = new CompletableFuture<>();
        request.completableFuture = future;
        //将对象（请求参数）传入队列
        queue.add(request);
        //如果这时候没完成赋值,那么就会阻塞,知道能够拿到值
        return future.get();
    }
    
    //开发中可以用这个做自动调用，我这边是为了把他整合到一个文件，直接调用的
    //@PostConstruct
    public static void init() {
        //定时任务线程池,创建一个支持定时、周期性或延时任务的限定线程数目(这里传入的是1)的线程池
        //scheduleAtFixedRate是周期性执行 schedule是延迟执行 initialDelay是初始延迟 period是周期间隔 后面是单位
        //这里我写的是周期性执行10毫秒执行一次
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            int size = queue.size();
            //如果队列没数据,表示这段时间没有请求,直接返回
            if (size == 0) {
                return;
            }
            List<Request1> list = new ArrayList<>();
            System.out.println("合并了" + size + "个请求");
            //将队列的请求消费到一个集合保存
            for (int i = 0; i < size; i++) {
                list.add(queue.poll());
            }
            //拿到我们需要去数据库查询的特征,保存为集合
            List<String> commodityCodes = new ArrayList<>();
            for (Request1 request : list) {
                commodityCodes.add(request.code);
            }
            //将参数传入service处理
            Map<String, HashMap<String, Object>> response = null;
            try {
                response = queryCommodityByCodeBatch(commodityCodes);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //将处理结果返回各自的请求
            for (Request1 request : list) {
                Map<String, Object> result = response.get(request.code);
                request.completableFuture.complete(result);    //completableFuture.complete方法完成赋值,这一步执行完毕,阻塞的请求可以继续执行了
            }
        }, 0, 1000, TimeUnit.MILLISECONDS);
    }

    /**
     * 模拟从数据库查询
     *
     * @param codes
     * @return
     */
    public static Map<String, HashMap<String, Object>> queryCommodityByCodeBatch(List<String> codes) throws InterruptedException {
        Map<String, HashMap<String, Object>> result = new HashMap();
        for (String code : codes) {
            HashMap<String, Object> hashMap = new HashMap<>();
            hashMap.put("commodityId", new Random().nextInt(999999999));
            hashMap.put("code", code);
            hashMap.put("phone", "huawei");
            hashMap.put("isOk", "true");
            hashMap.put("price", "4000");
            result.put(code, hashMap);
        }
        //TimeUnit.MILLISECONDS.sleep(100);
        return result;
    }

}

class Request1 {
    String code;
    CompletableFuture completableFuture;
}

线程池创建工具

https://www.jianshu.com/p/e2f6e44a51f0

最后编辑于：2024.11.02 17:54:28

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 213,254评论 6赞 492
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 90,875评论 3赞 387
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 158,682评论 0赞 348
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 56,896评论 1赞 285
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 66,015评论 6赞 385
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 50,152评论 1赞 291
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 39,208评论 3赞 412
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 37,962评论 0赞 268
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 44,388评论 1赞 304
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 36,700评论 2赞 327
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,867评论 1赞 341
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 34,551评论 4赞 335
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 40,186评论 3赞 317
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,901评论 0赞 21
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 32,142评论 1赞 267
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 46,689评论 2赞 362
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 43,757评论 2赞 351

并发/异步代码实现的四种方式（ThreadPool ，Callable，CompletableFuture，CompletableFuture+阻塞队列+定时任务线程池）

ThreadPool##############################################

Callable_1###############################################

Callable_2##############################################

CompletableFuture########################################

第四种########################################################

线程池创建工具

推荐阅读更多精彩内容