FutureTask原理解析-java多线程(实现并行计算)

本文涉及的产品
全局流量管理 GTM,标准版 1个月
云解析 DNS,旗舰版 1个月
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
简介: FutureTask原理解析-java多线程(实现并行计算)
需求: 我们要计算1+…+10和20+…+30相加的结果,当然可以用a=1+…+10,b=20+…+30,之后resutl=a+b。 但实际上第一个任务A和第二个任务B互不影响, 我们可以使用多线程的方法,将任务A和任务B并行执行,最后将两个任务的执行结果相加。那这样怎么用java实现呢?

1.实现的多种方法

第一种方法

首先分析一下需求,实际上A、B任务没有任何关联,所以我们可以直接使用多线程的方式,只不过由于线程什么时候运行结束,我们并不知道,所以这里先简单采取sleep,但是大家需要知道,这样是不对的,因为这个时间并非固定的。

public class Test {
    public static void main(String[] args) {
         Task task = new Task(1, 10);
         task.start();
         Task task1 = new Task(11, 20);
         task1.start();
         
         try {
             Thread.sleep(1000);
         } catch (InterruptedException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         }
         System.out.println(task.result + task1.result);
    }
    public static class Task extends Thread {
         int start;
         int end;
         int result = 0;
         public Task(int start, int end) {
             this.start = start;
             this.end = end;
         }
         public void run() {
             int i = start;
             result = 0;
             while (i <= end) {
                 result += i;
                 i++;
             }
         }
    }
}

第二种方法

既然第一种实现方法里面,有弊端的地方就是sleep,因为我们不知道两个线程啥时候结束,大家记得多线程知识章节,我们讲到过CountDownLatch吗?CountDownLatch是AQS的一种共享锁的具体实现,可以让当前线程阻塞,等待N个线程运行完成之后,才可继续运行。

import java.util.concurrent.CountDownLatch;
public class Test {
    static int result = 0;
    static CountDownLatch countDownLatch = new CountDownLatch(10);
    public static void main(String[] args) {
         for (int i = 0; i < 10; i++) {
             Task task = new Task(i * 10 + 1, (i + 1) * 10);
             task.start();
         }
         try {
             countDownLatch.await();
         } catch (InterruptedException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         }
         System.out.println(result);
    }
    public static class Task extends Thread {
         int start;
         int end;
         public Task(int start, int end) {
             this.start = start;
             this.end = end;
         }
         public void run() {
             int i = start;
             while (i <= end) {
                 result += i;
                 i++;
             }
             countDownLatch.countDown();
         }
    }
}

第三种方法

其实上面说到底,是因为Runnable没有返回值导致的,如果线程本身有阻塞的功能,也就是说在运行完成之前,get是阻塞的,那么就可以达到目的了。
Java其实已经给我们提供了这样的API,就是FutureTask,可以看一下FutureTask的类继承和实现关系。

public class FutureTask<V> implements RunnableFuture<V> {
}
public interface RunnableFuture<V> extends Runnable, Future<V> {
}

从这里看出FutureTask实际上就是一个Runnable而已,只不过它的绝妙点在于,get是阻塞式的,也就是未运行完成之前,get会阻塞。稍后总结环节,我们再来看FutureTask如何做到这点的?我们先来看实际使用。

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
public class Test {
    public static void main(String[] args) throws Exception {
         System.out.println(Thread.currentThread().getName());
         int result = 0;
         for (int i = 0; i < 10; i++) {
             FutureTask<Integer> futureTask = new FutureTask<Integer>(
                      new Callabled(i * 10 + 1, (i + 1) * 10));
             new Thread(futureTask).start();
             result += futureTask.get();
         }
         System.out.println(result);
    }
    public static class Callabled implements Callable<Integer> {
         int start;
         int end;
         public Callabled(int start, int end) {
             this.start = start;
             this.end = end;
         }
         @Override
         public Integer call() throws Exception {
             System.out.println(Thread.currentThread().getName());
             int result = 0;
             while (start <= end) {
                 result += start;
                 start++;
             }
             return result;
         }
    }
}

运行结果为:

main
Thread-0
Thread-1
Thread-2
Thread-3
Thread-4
Thread-5
Thread-6
Thread-7
Thread-8
Thread-9
5050

2.FutureTask原理

2.1 异同对比

首先通过上面的需求实践,大家其实已经知道了我们平常经常用的Runnable与FutureTask的区别了。

  • 相同点:Runnable和FutureTask都是并发线程执行任务;
  • 不同点:

    1)FutureTask在线程运行完成之前(可以是结果返回、也可以是中断),是会一直阻塞当前线程的,只要您调用了它的get方法;

    2)Runnable实现接口,只是一个run方法, 并不会阻塞当前线程,而且您也不能获取线程的运行结果;

2.2 状态定义

FutureTask既然要做到运行完成之前,一旦调用get就会阻塞当前线程,那么必然内部肯定维护了状态。我们看一下状态定义源码

rivate static final int NEW = 0;

private static final int COMPLETING = 1;

private static final int NORMAL = 2;

private static final int EXCEPTIONAL = 3;

private static final int CANCELLED = 4;

private static final int INTERRUPTING = 5;

private static final int INTERRUPTED = 6;

《Java并发编程的艺术》一本书中,根据FutureTask.run()方法的执行的时机,把FutureTask的运行状态分为三个阶段

  • 未启动:FutureTask.run()方法还没有被执行之前,FutureTask处于未启动状态
  • 已启动:FutureTask.run()方法执行之后,处于已启动状态。
  • 已完成:FutureTask.run()方法执行结束,或者调用FutureTask.cancel(…)方法取消任务,或者在执行任务期间抛出异常,这些情况都称之为FutureTask的已完成状态。

FutureTask的类继承和实现关系。

public class FutureTask<V> implements RunnableFuture<V> {
}
public interface RunnableFuture<V> extends Runnable, Future<V> {
}

从这里看出FutureTask实际上就是一个Runnable而已,我们继续看一下它的构造函数

    //Callable是什么?有Callable源码可知,实际上就是一个同步返回的接口定义而已,这里为什么传一个Callable进来呢?稍后我们再来解答
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }

   //我们可以直接把runnable传入,希望得到的运行结果传入
    public FutureTask(Runnable runnable, V result) {
        //此处源码会将我们传入的runnable和result,组合为callable接口实现
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

//可以看到Callable只是一个同步返回的功能接口定义而已
public interface Callable<V> {
    V call() throws Exception;
}

2.3 run方法

我们接下来看一下,FutureTask的run方法

 public void run() {
         //如果state不是新建,也就是已启动或者已完成,则return
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            //callable我们传入的具体实现操作
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    //调用callable的call方法, 也就是调用我们外部定义的具体实现操作,获取结果
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
                    //如果运行完成,则设置结果
                    set(result);
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }

这里也许大家有点迷糊了,callable到底是个啥?实际上,细心的小伙伴已经发行了,callable不就是我们外部定义的具体线程执行操作吗?

//使用
FutureTask<Integer> futureTask = new FutureTask<Integer>(
                      new Callabled(start, end));

public static class Callabled implements Callable<Integer> {
         int start;
         int end;
         public Callabled(int start, int end) {
             this.start = start;
             this.end = end;
         }
         @Override
         public Integer call() throws Exception {
             System.out.println(Thread.currentThread().getName());
             int result = 0;
             while (start <= end) {
                 result += start;
                 start++;
             }
             return result;
         }
    }

从上面源码可以看到,FutureTask的run会将callable的运行结果保存下来,那么关键来了,get是如何阻塞的?

2.4 get方法

大家可以先想一下,如果自己实现,该如何做?

我的头脑风暴过程是这样的:

1)上面已经有了这个线程运行的几种状态的定义,那么在callable的call结果返回之前 or 中断异常抛出之前,那么状态就是COMPLETING之前的状态。那我自己get的时候,判断当前FutureTask的内部状态,如果<=COMPLETING,那么直接线程等待?

2)这里关键问题来了,线程如何等待?目前我自己想的话,是写一个死循环,每次sleep固定的时间,例如10ms,然后判断状态是否是完成了,如果没有,则继续sleep 10ms,每隔10ms线程重新进入就绪状态,排队获取cpu时间片,直到完成。

我们接下来看一下关键FutureTask的get方法如何实现的。

  public V get() throws InterruptedException, ExecutionException {
        int s = state;
          //如果状态小于等于COMPLETING,则进入awaitDone
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    }

 private int awaitDone(boolean timed, long nanos)
         //timeed上面传入为false,nanos为0
        throws InterruptedException {
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        boolean queued = false;
        for (;;) {
            //...省略若干
            else
                //代码会走到这里
                LockSupport.park(this);
        }
    }

LockSupport.park(this);这里我们看到它的wait实现调用了这个方法,这个是干嘛的?和sleep的区别是?我们来通过一张图回顾一下线程的几种状态。
在这里插入图片描述

这里篇幅所限,就不详细展开了,各位有兴趣的小伙伴,可以自行搜索,或者看一下文章末尾的一篇文章链接。

LockSupport.park()休眠线程,LockSupport.unpark()唤醒线程,两个方法配合使用。也可以通过LockSupport.parkNanos()指定休眠时间后,自动唤醒。LockSupport.park()不会释放monitor锁。LockSupport.park()可以响应中断。

LockSupport.park()会是线程进入WAITING状态,而LockSupport.parkNanos(long nanos) 会进入TIMED_WAITING状态。

我们看一下,LockSupport.unpark()唤醒线程是在哪里做的?

      //run中运行结果获取之后,会调用set方法
    protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }
    //取消方法
    public boolean cancel(boolean mayInterruptIfRunning) {
       //...
        try {    // in case call to interrupt throws exception
           //...
        } finally {
            finishCompletion();
        }
        return true;
    }
    private void finishCompletion() {
        // assert state > COMPLETING;
        //...省略若干
                        LockSupport.unpark(t);
                  //...
    }

大家发现没有,LockSupport很像wait+sleep达到的效果,因为它不会释放锁,同时又必须成对去唤醒才可以。

课后作业

提一个需求,大家不要参考任何代码的前提下,自己实现一下,多线程下载一个文件,可以指定开启多少个线程,简单一点,对于客户来说,调用的接口就是输入两个参数,一个下载文件的url,一个是多少个线程。还有就是需要可以注册下载进度回调监听。

Java中的线程休眠大法系列(三)LockSupport.park()

对于Java线程中断的理解,哪种情况下会响应中断?哪种情况下不响应中断?

目录
相关文章
|
1月前
|
存储 缓存 算法
HashMap深度解析:从原理到实战
HashMap,作为Java集合框架中的一个核心组件,以其高效的键值对存储和检索机制,在软件开发中扮演着举足轻重的角色。作为一名资深的AI工程师,深入理解HashMap的原理、历史、业务场景以及实战应用,对于提升数据处理和算法实现的效率至关重要。本文将通过手绘结构图、流程图,结合Java代码示例,全方位解析HashMap,帮助读者从理论到实践全面掌握这一关键技术。
88 13
|
8天前
|
机器学习/深度学习 自然语言处理 搜索推荐
自注意力机制全解析:从原理到计算细节,一文尽览!
自注意力机制(Self-Attention)最早可追溯至20世纪70年代的神经网络研究,但直到2017年Google Brain团队提出Transformer架构后才广泛应用于深度学习。它通过计算序列内部元素间的相关性,捕捉复杂依赖关系,并支持并行化训练,显著提升了处理长文本和序列数据的能力。相比传统的RNN、LSTM和GRU,自注意力机制在自然语言处理(NLP)、计算机视觉、语音识别及推荐系统等领域展现出卓越性能。其核心步骤包括生成查询(Q)、键(K)和值(V)向量,计算缩放点积注意力得分,应用Softmax归一化,以及加权求和生成输出。自注意力机制提高了模型的表达能力,带来了更精准的服务。
|
20天前
|
人工智能 自然语言处理 Java
FastExcel:开源的 JAVA 解析 Excel 工具,集成 AI 通过自然语言处理 Excel 文件,完全兼容 EasyExcel
FastExcel 是一款基于 Java 的高性能 Excel 处理工具,专注于优化大规模数据处理,提供简洁易用的 API 和流式操作能力,支持从 EasyExcel 无缝迁移。
94 9
FastExcel:开源的 JAVA 解析 Excel 工具,集成 AI 通过自然语言处理 Excel 文件,完全兼容 EasyExcel
|
6天前
|
SQL Java 数据库连接
如何在 Java 代码中使用 JSqlParser 解析复杂的 SQL 语句?
大家好,我是 V 哥。JSqlParser 是一个用于解析 SQL 语句的 Java 库,可将 SQL 解析为 Java 对象树,支持多种 SQL 类型(如 `SELECT`、`INSERT` 等)。它适用于 SQL 分析、修改、生成和验证等场景。通过 Maven 或 Gradle 安装后,可以方便地在 Java 代码中使用。
96 11
|
5天前
|
存储 分布式计算 Hadoop
基于Java的Hadoop文件处理系统:高效分布式数据解析与存储
本文介绍了如何借鉴Hadoop的设计思想,使用Java实现其核心功能MapReduce,解决海量数据处理问题。通过类比图书馆管理系统,详细解释了Hadoop的两大组件:HDFS(分布式文件系统)和MapReduce(分布式计算模型)。具体实现了单词统计任务,并扩展支持CSV和JSON格式的数据解析。为了提升性能,引入了Combiner减少中间数据传输,以及自定义Partitioner解决数据倾斜问题。最后总结了Hadoop在大数据处理中的重要性,鼓励Java开发者学习Hadoop以拓展技术边界。
30 7
|
6天前
|
Java Linux 调度
硬核揭秘:线程与进程的底层原理,面试高分必备!
嘿,大家好!我是小米,29岁的技术爱好者。今天来聊聊线程和进程的区别。进程是操作系统中运行的程序实例,有独立内存空间;线程是进程内的最小执行单元,共享内存。创建进程开销大但更安全,线程轻量高效但易引发数据竞争。面试时可强调:进程是资源分配单位,线程是CPU调度单位。根据不同场景选择合适的并发模型,如高并发用线程池。希望这篇文章能帮你更好地理解并回答面试中的相关问题,祝你早日拿下心仪的offer!
24 6
|
27天前
|
存储 缓存 Java
Java 并发编程——volatile 关键字解析
本文介绍了Java线程中的`volatile`关键字及其与`synchronized`锁的区别。`volatile`保证了变量的可见性和一定的有序性,但不能保证原子性。它通过内存屏障实现,避免指令重排序,确保线程间数据一致。相比`synchronized`,`volatile`性能更优,适用于简单状态标记和某些特定场景,如单例模式中的双重检查锁定。文中还解释了Java内存模型的基本概念,包括主内存、工作内存及并发编程中的原子性、可见性和有序性。
Java 并发编程——volatile 关键字解析
|
19天前
|
存储 物联网 大数据
探索阿里云 Flink 物化表:原理、优势与应用场景全解析
阿里云Flink的物化表是流批一体化平台中的关键特性,支持低延迟实时更新、灵活查询性能、无缝流批处理和高容错性。它广泛应用于电商、物联网和金融等领域,助力企业高效处理实时数据,提升业务决策能力。实践案例表明,物化表显著提高了交易欺诈损失率的控制和信贷审批效率,推动企业在数字化转型中取得竞争优势。
75 14
|
25天前
|
Java 数据库连接 Spring
反射-----浅解析(Java)
在java中,我们可以通过反射机制,知道任何一个类的成员变量(成员属性)和成员方法,也可以堆任何一个对象,调用这个对象的任何属性和方法,更进一步我们还可以修改部分信息和。
|
28天前
|
网络协议 安全 网络安全
探索网络模型与协议:从OSI到HTTPs的原理解析
OSI七层网络模型和TCP/IP四层模型是理解和设计计算机网络的框架。OSI模型包括物理层、数据链路层、网络层、传输层、会话层、表示层和应用层,而TCP/IP模型则简化为链路层、网络层、传输层和 HTTPS协议基于HTTP并通过TLS/SSL加密数据,确保安全传输。其连接过程涉及TCP三次握手、SSL证书验证、对称密钥交换等步骤,以保障通信的安全性和完整性。数字信封技术使用非对称加密和数字证书确保数据的机密性和身份认证。 浏览器通过Https访问网站的过程包括输入网址、DNS解析、建立TCP连接、发送HTTPS请求、接收响应、验证证书和解析网页内容等步骤,确保用户与服务器之间的安全通信。
102 1

热门文章

最新文章

推荐镜像

更多