countDownLatch Future 实践学习以及parallelStream方式实现并发

66 阅读 0 评论 44 点赞

我是靠谱客的博主迷路唇彩，最近开发中收集的这篇文章主要介绍countDownLatch Future 实践学习以及parallelStream方式实现并发，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

嘚不嘚：自从到了新公司好久都没有时间写博客了，最近在忙什么呢，忙着学习业务、技术、服务器环境等。不过公司用的技术还挺棒的，起码在这之前我没有接触过，之前只是听说过，但是没有用过。主要用到的技术dubbo、es、zoopker。大概了解了这些技术是干嘛的，还没有深入了解呢，dubbo我自己搭了一个简单的用了用，到底哪里好，等研究研究再说，es和zookeeper的应用都没有怎么看。现在在学一些其他的应用技术guava、lambda、redis。redis之前在spring的框架中使用过，其他的也没有什么了解，最近我也在看看redis相关的东西。接触的多了发现了以前的技术有多low，个人觉得还是去一些技术比较好的平台，第一是技术底蕴，第二是技术氛围也比较好。countDownLatch Future则两个类我以前自己看过，但是从来没有理解过，也不知道这东西到底在什么时候使用，这次见到实例了，自己用了用，感觉好像会用了。如有问题希望大神多多指教。

- - - - 概述
        代码
        lambda parallelStream方式实现：

概述

countDownLatch 和 Future是java.util.concurrent包下面的类，为并发所用。
- countDownLatch：利用它可以实现类似计数器的功能。（一个方法内如果现在存在1、2、3、4四个线程正在处理任务，countDownLatch的值是4，四个线程的执行时间分别是10毫秒依次递增，用countDownLatch时，会在四个线程都执行完之后返回结果，如果没有用countDownLatch，有可能指完成了部分的线程就返回结果了）
- Future：对于具体的Callable或者Runnable的任务执行结果进行取消、查询是否完成和获取结果。（对于线程可以通过Future对当前线程进行）
-业务场景：调用其他的接口返回了结果，结果被封装到Version中，需要将Version中的结果转化为List。

代码

@Data
public class Version<T> {

    private T data;

    private String code;

    private String message;

}

@Data
public class User {

    private String name;

    private int age;

    private String grade;

}
//线程类
public class Task implements Callable<User>{

    private Version<User> version;
    private CountDownLatch latch;

    public Task(Version<User> version, CountDownLatch latch) {
        this.version = version;
        this.latch = latch;
    }

    @Override
    public User call() throws Exception {
        User u;
        try {
            u = version.getData();
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }finally {
            latch.countDown();
        }
        return u;
    }
}

具体的应用实现


    /**
     *两种实现线程的方式，比较推荐第二种，自定义线程不容易对资源产生浪费，
     */
    //private static final ExecutorService executorService = Executors.newFixedThreadPool(3);
    private static final ExecutorService executorService = new ThreadPoolExecutor
                (5,10,30, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        List<Version<User>> list = params();
        List<User> result =  latchFuture(list);
        result.forEach((user) -> System.out.println(user.getAge()+" "+user.getGrade()+" "+user.getName()));
    }

    /**
     * 没有用CountDownLatch实现功能
     * @param userList
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static List<User> versionToUser(List<Version<User>> userList) throws ExecutionException, InterruptedException {

        List<Future<User>> result = Lists.newArrayList();


        for (Version<User> v : userList){
            Future<User> future  = executorService.submit(() -> v.getData());
            result.add(future);
        }
        List<User> list = Lists.newArrayList();
        for(Future<User> u : result){
            list.add(u.get());
        }
        executorService.shutdown();
        return  list;
    }

    public static List<User> latchFuture(List<Version<User>> userList) throws ExecutionException, InterruptedException {

        List<Future<User>> result = Lists.newArrayList();

        /**
         * 线程执行完成后返回结果
         */

        final CountDownLatch latch = new CountDownLatch(3);

        for (Version<User> v : userList){
            Future<User> future  = executorService.submit(new Task(v,latch));
            result.add(future);
        }
        List<User> list = Lists.newArrayList();
        for(Future<User> u : result){
            list.add(u.get());
        }
        return  list;
    }
    private static List<Version<User>> params(){
        User u1 = new User();
        u1.setAge(11);
        u1.setGrade("88");
        u1.setName("lucas");
        User u2 = new User();
        u2.setAge(21);
        u2.setGrade("90");
        u2.setName("xiaoming");
        User u3 = new User();
        u3.setAge(31);
        u3.setGrade("94");
        u3.setName("xiaohong");

        List<Version<User>> list = Lists.newArrayList();
        Version<User> v1 = new Version<>();
        v1.setData(u1);
        Version<User> v2 = new Version<>();
        v2.setData(u2);
        Version<User> v3 = new Version<>();
        v3.setData(u3);

        list.add(v1);
        list.add(v2);
        list.add(v3);
        return list;
    }

lambda parallelStream方式实现：

  public static List<User>  transfer(List<Version<User>> version){
        List<User> users = version.parallelStream()
                .map(Version::getData)
                .collect(Collectors.toList());
        return null;
    }