并发/异步代码实现的四种方式（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

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容