Future与FutureTask源码解析,接口阻塞问题及解决方案

本文涉及的产品
云解析 DNS,旗舰版 1个月
全局流量管理 GTM,标准版 1个月
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
简介: 【11月更文挑战第5天】在Java开发中,多线程编程是提高系统并发性能和资源利用率的重要手段。然而,多线程编程也带来了诸如线程安全、死锁、接口阻塞等一系列复杂问题。本文将深度剖析多线程优化技巧、Future与FutureTask的源码、接口阻塞问题及解决方案,并通过具体业务场景和Java代码示例进行实战演示。

引言

在Java开发中,多线程编程是提高系统并发性能和资源利用率的重要手段。然而,多线程编程也带来了诸如线程安全、死锁、接口阻塞等一系列复杂问题。本文将深度剖析多线程优化技巧、Future与FutureTask的源码、接口阻塞问题及解决方案,并通过具体业务场景和Java代码示例进行实战演示。

一、多线程优化技巧

1.1 线程池的使用与调优

线程池是一种预先创建一定数量线程的容器,当有任务需要执行时,从线程池中获取可用线程来执行任务,任务完成后线程不会被销毁,而是返回线程池等待下一个任务。线程池的使用可以显著减少线程创建和销毁的开销,提高系统资源利用率和响应性。

1.1.1 线程池的配置参数

  • 核心线程数(corePoolSize):线程池中的核心线程数,即使在空闲状态下也不会被回收。
  • 最大线程数(maximumPoolSize):线程池允许的最大线程数,当任务队列满且核心线程都在忙时,会创建新的线程直到达到最大线程数。
  • 队列容量(queueCapacity):任务队列的长度,当线程池中的线程数达到核心线程数时,新任务会被放入队列中等待执行。
  • 存活时间(keepAliveTime):当线程池中的线程数超过核心线程数时,多余的空闲线程在存活时间到达后会被回收。

1.1.2 线程池的类型

Java提供了多种线程池实现,通过Executors工厂类可以方便地创建不同类型的线程池:

  • newFixedThreadPool:创建固定大小的线程池。
  • newCachedThreadPool:创建一个可缓存的线程池,线程数量不固定,会根据需要创建新线程或复用已有线程。
  • newSingleThreadExecutor:创建一个单线程的线程池,所有任务按顺序在一个线程中执行。
  • newScheduledThreadPool:创建一个支持定时和周期性任务执行的线程池。

1.1.3 线程池的调优

  • 考虑CPU核心数:线程数量通常与CPU核心数相关。对于计算密集型任务,线程数量可以设置为CPU核心数或CPU核心数+1,以充分利用CPU资源。
  • 考虑任务类型:对于I/O密集型任务,由于线程在等待I/O操作时会处于阻塞状态,线程数量可以适当增加,一般可以设置为2*CPU核心数。
  • 考虑系统资源:还需要考虑系统的内存、网络带宽等资源限制,避免创建过多线程导致系统资源不足。

1.2 分而治之思想

分而治之是一种将复杂问题分解为若干个子问题分别解决的思想。在多线程编程中,可以利用分而治之的思想将一个任务分解为若干个子任务,并指派专门的线程来并发执行这些子任务。

1.2.1 基于数据的分割

将程序的输入数据分解为若干个规模较小的数据,并利用若干个工作线程并发处理这些分解后的数据。例如,在处理大规模日志文件时,可以将日志文件按行分割,每个线程处理一部分数据。

1.2.2 基于任务的分割

将程序的处理任务分解为若干子任务,并分配若干子线程并发执行这些子任务。例如,在进行大规模计算时,可以将计算任务分解为多个子任务,每个子任务由一个线程执行。

二、Future与FutureTask源码解析

2.1 Future接口

Future表示一个异步计算的结果,并提供了一些方法来检测计算是否完成、等待完成以及获取结果。

java复制代码
public interface Future<V> {  
boolean cancel(boolean mayInterruptIfRunning);  
boolean isCancelled();  
boolean isDone();  
    V get() throws InterruptedException, ExecutionException;  
    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;  
}

2.2 FutureTask类

FutureTask是Future接口的一个实现类,它实现了RunnableFuture接口,该接口继承了Runnable和Future接口。因此,FutureTask既可以作为Runnable提交给Thread执行,又可以作为Future获取异步计算的结果。

java复制代码
public class FutureTask<V> implements RunnableFuture<V> {  
// 任务状态  
private volatile int state;  
private 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;  
// 下层Callable  
private Callable<V> callable;  
// 任务结果  
private Object outcome;  
// 运行线程  
private volatile Thread runner;  
// 等待节点  
private volatile WaitNode waiters;  
// 构造函数  
public FutureTask(Callable<V> callable) {  
if (callable == null) throw new NullPointerException();  
this.callable = callable;  
this.state = NEW;  
    }  
public FutureTask(Runnable runnable, V result) {  
this.callable = Executors.callable(runnable, result);  
this.state = NEW;  
    }  
// run方法  
public void run() {  
if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))  
return;  
try {  
            Callable<V> c = callable;  
if (c != null && state == NEW) {  
                V result;  
boolean ran;  
try {  
                    result = c.call();  
                    ran = true;  
                } catch (Throwable ex) {  
                    result = null;  
                    ran = false;  
                    setException(ex);  
                }  
if (ran) set(result);  
            }  
        } finally {  
            runner = null;  
int s = state;  
if (s >= INTERRUPTING)  
                handlePossibleCancellationInterrupt(s);  
        }  
    }  
// set方法  
protected void set(V v) {  
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {  
            outcome = v;  
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL);  
            finishCompletion();  
        }  
    }  
// finishCompletion方法  
private void finishCompletion() {  
for (WaitNode q; (q = waiters) != null;) {  
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {  
for (;;) {  
Thread t = q.thread;  
if (t != null) {  
                        q.thread = null;  
                        LockSupport.unpark(t);  
                    }  
WaitNode next = q.next;  
if (next == null) break;  
                    q.next = null;  
                    q = next;  
                }  
break;  
            }  
        }  
        done();  
        callable = null;  
    }  
// get方法  
public V get() throws InterruptedException, ExecutionException {  
int s = state;  
if (s <= COMPLETING)  
            s = awaitDone(false, 0L);  
return report(s);  
    }  
public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {  
if (unit == null) throw new NullPointerException();  
int s = state;  
if (s <= COMPLETING && (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)  
throw new TimeoutException();  
return report(s);  
    }  
// 其他方法...  
}

三、接口阻塞问题及解决方案

3.1 接口阻塞问题表现

在Java中,接口阻塞问题通常表现为请求发出后长时间无响应,导致用户体验不佳,同时会占用系统资源,影响其他接口的调用,甚至引发连锁反应,导致整个系统瘫痪。

3.2 接口阻塞问题原因

  • I/O操作:如网络请求、文件读写等,这些操作的执行时间不可控,容易导致阻塞。
  • 全局解释器锁(GIL):在Python中,GIL限制了多线程的并发性能,导致长时间操作会阻塞其他线程。
  • 设计不当:在应用架构中未能合理使用异步编程或多线程技术,导致请求处理不高效。

3.3 解决方案

3.3.1 多线程处理

通过多线程处理长时间的操作,可以在一定程度上减轻阻塞的问题。例如,在Java中可以使用Thread类或ExecutorService来创建和管理线程。

java复制代码
import java.util.concurrent.ExecutorService;  
import java.util.concurrent.Executors;  
public class TaskExecutor {  
private static final int THREAD_POOL_SIZE = 10;  
private ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);  
public void executeTask(Runnable task) {  
        executorService.execute(task);  
    }  
public void shutdown() {  
        executorService.shutdown();  
    }  
public static void main(String[] args) {  
TaskExecutor taskExecutor = new TaskExecutor();  
        taskExecutor.executeTask(() -> {  
// 模拟长时间运行的任务  
try {  
                Thread.sleep(5000);  
            } catch (InterruptedException e) {  
                e.printStackTrace();  
            }  
            System.out.println("Task completed");  
        });  
        taskExecutor.shutdown();  
    }  
}

3.3.2 异步编程

异步编程可以有效地处理非阻塞操作。在Java中,可以使用CompletableFuture来实现异步编程。

java复制代码
import java.util.concurrent.CompletableFuture;  
public class AsyncTask {  
public static void main(String[] args) {  
        CompletableFuture.runAsync(() -> {  
// 模拟长时间运行的任务  
try {  
                Thread.sleep(5000);  
            } catch (InterruptedException e) {  
                e.printStackTrace();  
            }  
            System.out.println("Async task completed");  
        });  
        System.out.println("Main thread continues to execute");  
    }  
}

3.3.3 消息队列

将长时间的任务放入消息队列(如RabbitMQ、Celery)中异步处理,可以极大地改善接口阻塞的问题。以下是一个使用Celery处理异步任务的简单示例。

python复制代码
from celery import Celery  
import time  
app = Celery('tasks', broker='redis://localhost:6379/0')  
@app.task  
def long_running_task():  
    time.sleep(10)  
return "Task completed"
# 在Flask应用中调用Celery任务  
from flask import Flask, jsonify  
from tasks import long_running_task  
app = Flask(__name__)  
@app.route('/task', methods=['POST'])  
def task():  
    long_running_task.delay()  
return jsonify({"status": "Task received, processing..."})  
@app.route('/status', methods=['GET'])  
def status():  
return jsonify({"status": "Processing tasks..."})  
if __name__ == '__main__':  
    app.run(threaded=True)

四、实战示例:多线程文件下载系统

假设我们要开发一个文件下载系统,需要同时处理多个文件的下载任务。下载任务是I/O密集型操作,需要较多的线程来提高并发度。

4.1 系统设计

  • 线程池:使用固定大小的线程池来管理下载任务。
  • 任务队列:使用有界队列来存放待下载的任务。
  • 下载逻辑:每个下载任务由一个线程执行,从指定的URL下载文件并保存到本地。

4.2 代码实现

java复制代码
import java.io.FileOutputStream;  
import java.io.InputStream;  
import java.net.HttpURLConnection;  
import java.net.URL;  
import java.util.concurrent.ArrayBlockingQueue;  
import java.util.concurrent.ThreadPoolExecutor;  
import java.util.concurrent.TimeUnit;  
public class FileDownloader {  
private static final int CORE_POOL_SIZE = 10;  
private static final int MAX_POOL_SIZE = 20;  
private static final int QUEUE_CAPACITY = 50;  
private static final long KEEP_ALIVE_TIME = 60;  
private ThreadPoolExecutor executor;  
public FileDownloader() {  
        executor = new ThreadPoolExecutor(  
                CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME,  
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(QUEUE_CAPACITY),  
new ThreadPoolExecutor.CallerRunsPolicy()  
        );  
    }  
public void downloadFile(String fileUrl, String savePath) {  
        executor.execute(() -> {  
try {  
URL url = new URL(fileUrl);  
HttpURLConnection connection = (HttpURLConnection) url.openConnection();  
                connection.setRequestMethod("GET");  
try (InputStream inputStream = connection.getInputStream();  
FileOutputStream outputStream = new FileOutputStream(savePath)) {  
byte[] buffer = new byte[1024];  
int bytesRead;  
while ((bytesRead = inputStream.read(buffer)) != -1) {  
                        outputStream.write(buffer, 0, bytesRead);  
                    }  
                }  
                System.out.println("File downloaded: " + savePath);  
            } catch (Exception e) {  
                e.printStackTrace();  
            }  
        });  
    }  
public static void main(String[] args) {  
FileDownloader downloader = new FileDownloader();  
        downloader.downloadFile("http://example.com/file1.zip", "file1.zip");  
        downloader.downloadFile("http://example.com/file2.zip", "file2.zip");  
    }  
}

4.3 系统测试

通过模拟多个文件下载任务,测试系统的并发处理能力和响应性。可以使用JUnit编写测试用例来验证系统的正确性。

java复制代码
import org.junit.Test;  
import static org.junit.Assert.*;  
public class FileDownloaderTest {  
@Test
public void testDownloadFile() {  
FileDownloader downloader = new FileDownloader();  
        downloader.downloadFile("http://example.com/file1.zip", "file1.zip");  
        downloader.downloadFile("http://example.com/file2.zip", "file2.zip");  
// 等待一段时间以确保下载任务完成  
try {  
            Thread.sleep(10000);  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
// 验证文件是否存在  
        assertTrue(new java.io.File("file1.zip").exists());  
        assertTrue(new java.io.File("file2.zip").exists());  
    }  
}

五、总结

本文通过深度剖析多线程优化技巧、Future与FutureTask源码、接口阻塞问题及解决方案,并通过具体业务场景和Java代码示例进行了实战演示。作为资深Java架构师,掌握这些技术和方法对于提高系统并发性能和资源利用率具有重要意义。希望本文能够为读者在实际开发中提供一些有价值的参考和启示。

相关文章
|
17天前
|
监控 Java 应用服务中间件
高级java面试---spring.factories文件的解析源码API机制
【11月更文挑战第20天】Spring Boot是一个用于快速构建基于Spring框架的应用程序的开源框架。它通过自动配置、起步依赖和内嵌服务器等特性,极大地简化了Spring应用的开发和部署过程。本文将深入探讨Spring Boot的背景历史、业务场景、功能点以及底层原理,并通过Java代码手写模拟Spring Boot的启动过程,特别是spring.factories文件的解析源码API机制。
49 2
|
18天前
|
存储 安全 Linux
Golang的GMP调度模型与源码解析
【11月更文挑战第11天】GMP 调度模型是 Go 语言运行时系统的核心部分,用于高效管理和调度大量协程(goroutine)。它通过少量的操作系统线程(M)和逻辑处理器(P)来调度大量的轻量级协程(G),从而实现高性能的并发处理。GMP 模型通过本地队列和全局队列来减少锁竞争,提高调度效率。在 Go 源码中,`runtime.h` 文件定义了关键数据结构,`schedule()` 和 `findrunnable()` 函数实现了核心调度逻辑。通过深入研究 GMP 模型,可以更好地理解 Go 语言的并发机制。
|
21天前
|
监控 关系型数据库 MySQL
MySQL自增ID耗尽应对策略:技术解决方案全解析
在数据库管理中,MySQL的自增ID(AUTO_INCREMENT)属性为表中的每一行提供了一个唯一的标识符。然而,当自增ID达到其最大值时,如何处理这一情况成为了数据库管理员和开发者必须面对的问题。本文将探讨MySQL自增ID耗尽的原因、影响以及有效的应对策略。
69 3
|
25天前
|
存储 人工智能 自然语言处理
高效档案管理案例介绍:文档内容批量结构化解决方案解析
档案文件内容丰富多样,传统人工管理耗时低效。思通数科AI平台通过自动布局分析、段落与标题检测、表格结构识别、嵌套内容还原及元数据生成等功能,实现档案的高精度分块处理和结构化存储,大幅提升管理和检索效率。某历史档案馆通过该平台完成了500万页档案的数字化,信息检索效率提升60%。
|
24天前
|
存储
文件太大不能拷贝到U盘怎么办?实用解决方案全解析
当我们试图将一个大文件拷贝到U盘时,却突然跳出提示“对于目标文件系统目标文件过大”。这种情况让人感到迷茫,尤其是在急需备份或传输数据的时候。那么,文件太大为什么会无法拷贝到U盘?又该如何解决?本文将详细分析这背后的原因,并提供几个实用的方法,帮助你顺利将文件传输到U盘。
|
2月前
|
缓存 Java 程序员
Map - LinkedHashSet&Map源码解析
Map - LinkedHashSet&Map源码解析
70 0
|
2月前
|
算法 Java 容器
Map - HashSet & HashMap 源码解析
Map - HashSet & HashMap 源码解析
57 0
|
2月前
|
存储 Java C++
Collection-PriorityQueue源码解析
Collection-PriorityQueue源码解析
64 0
|
2月前
|
安全 Java 程序员
Collection-Stack&Queue源码解析
Collection-Stack&Queue源码解析
85 0
|
2月前
|
存储
让星星⭐月亮告诉你,HashMap的put方法源码解析及其中两种会触发扩容的场景(足够详尽,有问题欢迎指正~)
`HashMap`的`put`方法通过调用`putVal`实现,主要涉及两个场景下的扩容操作:1. 初始化时,链表数组的初始容量设为16,阈值设为12;2. 当存储的元素个数超过阈值时,链表数组的容量和阈值均翻倍。`putVal`方法处理键值对的插入,包括链表和红黑树的转换,确保高效的数据存取。
59 5

推荐镜像

更多