6.1 在线程中执行任务
6.1.1 串行地执行任务
程序清单6-1 串行的web服务器
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
public class SingleThreadWebServer {
public static void main(String[] args) throws IOException {
ServerSocket socket = new ServerSocket(10000);
while(true) {
Socket connection = socket.accept();
handleRequest(connection);
}
}
}
6.1.2 显示地为任务创建线程
程序清单6-2 在web 服务器中为每一个请求启动一个新的线程（不要这么做）
/**
* 1、任务处理过程从主线程中分离出来使得主循环能够更快地重新等待下一个到来的连接。
* 这使得程序在完成前面的请求之前可以接收新的请求，提供相应性
* 2、任务可以并行处理，提高吞吐量
* 3、处理任务代码必须是线程安全的
*/
public class ThreadPerTaskWebServer {
public static void main(String[] args) throws IOException {
ServerSocket socket = new ServerSocket(11111);
while(true) {
final Socket connection = socket.accept();
Runnable task = new Runnable() {
@Override
public void run() {
handleRequest(connection);
Executor e;
}
};
new Thread(task).start();
}
}
}
6.1.3 无限制创建线程的不足
线程生命周期非常高
资源消耗
稳定性
6.2Executor框架
程序清单6-3 Executor接口
public interface Executor {
void execute(Runnable command);
}
6.2.1示例：基于Executor的Web服务器
程序清单6-4 基于线程池的Web服务器
public class TaskExecutionWebServer {
private final static int NTHREADS = 100;
private final static Executor exec = Executors.newFixedThreadPool(NTHREADS);
public static void main(String[] args) throws IOException {
ServerSocket serverSocket = new ServerSocket(1234);
while(true) {
final Socket connection = serverSocket.accept();
Runnable task = new Runnable() {
@Override
public void run() {
handleRequest(connection);
}
};
exec.execute(task);
}
}
}
程序清单6-5 为每个请求启动一个新的Executor
public class ThreadPerTaskExecutor implements Executor {
public void execute(Runnable r) {
new Thread(r).start();
}
}
程序清单6-6 在调用线程中以同步方式执行所有任务的Executor
public class WithinThreadExecutor implements Executor {
public void execute(Runnable r) {
r.run();
}
}
6.2.2 执行策略
6.2.3 线程池
6.2.34 线程生命周期
1、运行
2、关闭
3、停止
程序清单6-7 ExecutorService中的声明周期管理方法
public interface ExecutorService extends Executor {
void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
//其它用于任务提交的便利方法
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
//其它用于任务提交的便利方法
......
}
程序清单6-8 支持关闭操作的Web服务器
public class LifecycleWebServer {
private final ExecutorService exec = Executors.newCachedThreadPool();
public void start() throws IOException {
ServerSocket socket = new ServerSocket(1111);
//线程池的关闭即意味着Web服务器关闭，Web服务器逻辑不再执行
while(!exec.isShutdown()) {
try {
final Socket conn = socket.accept();
exec.execute(new Runnable() {
@Override
public void run() {
handleRequest(conn);
}
});
} catch(RejectedExecutionException e) {
if(!exec.isShutdown()) {
log("task submition rejected ",e);
}
}
}
}
/**关闭服务器*/
public void stop() {
exec.shutdown();//关闭线程池
}
/**处理请求*/
void handleRequest(Socket connection) {
Request req = readRequest(connection);
if(isShutdownRequest(connection))
stop();
else
dispatchRequest(req);
}
}
6.2.5 延迟任务与周期任务
ScheduledThreadPoolExecutor能正确处理Timer表现出错误行为的任务。如果要构建自己的调度服务，可以使用DelayQueue，他实现了BlockingQueue，并为ScheduledThreadPoolExecutor提供调度功能。DelayQueue管理着一组Delayed对象。每个Delayed对象都有一个相应的延迟时间：在DelayedQueue中，只有某个元素逾期后，才能从DelayQueue中执行take操作。从DelayQueue中返回的对象将根据他们的延迟时间进行排序。
6.3 找出可利用的并行性
Executor框架帮助指定执行策略，但如果使用Executor，必须将任务表述成一个Runnable。大多数服务器应用程序都存在一个明显的 任务边界：单个客户请求。
单有时候任务边界是模糊的，即使是服务器，单个客户请求仍然有可待发掘的并行性，eg：DB服务器。
程序清单6-9 错误的Timer行为
略
6.3.1 示例：串行的页面渲染器（浏览器，渲染HTML）
对HTML文档进行渲染渲染的2种串行处理方式
- 当遇到文本标签时，将其绘制到图片缓存中；当遇到图片标签时，先通过网络获取到它，然后将其放到图像缓存中。
- 先绘制文本元素，同时为图片预留出矩形占位空间，在处理完第一遍文本后，程序开始下载图像并将其绘制到相应的占位空间去。eg:6-10
程序清单6-10 串行地渲染页面元素
//render 英[ˈrendə(r)] v. 给予; 使成为; 递交; 表达;
public class SingleThreadRender {
void renderPage(CharSequence source) {
renderText(source);//渲染文本标签：eg:<input> <b><br>
List<ImageData> imageData = new ArrayList<>();
for(ImageInfo imageInfo : scanForImageInfo(source))
imageData.add(imageInfo.downloadImage()); //下载图片
for(ImageData image : imageData)
renderImage(image);//往Web浏览器上渲染图片
}
}
6.3.2 带结果的任务Callable与Future
- Executor接口的execute(...)执行的是Runnable，而Runnable没有返回值，所以是void execute（）；
- ExecutorServcie继承了Executor，拓展了生命周期方法以及一些提交的方法：submit(...)，它提交的是Callable()；即便是Runnable参数也会转化成Callable，所以submit(...)返回的都是Future<V>
- Executor执行的任务有4个生命周期阶段：创建、提交、开始、完成
- Future 表示一个任务的生命周期，并且提供了相应的方法来判断是否已经完成或取消，以及获取任务的最终结果和取消任务等。Future规范中隐含的含义是：任务的生命周期只能前进，不能后退，就像ExecutorServcie的生命周期一样。
程序清单6-11Callable和Future接口以及FutureTask
//任务：真正的业务逻辑
public interface Callable<V> {
V call() throws Exception;
}
//任务的生命周期以操作任务
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;
}
//结合了线程的任务定义
public interface RunnableFuture<V> extends Runnable, Future<V> {
void run();
}
结合了线程的任务周期的落地实现：FutureTask
/**
* FutureTask是Future以及Runnable的共同实现，
* 并在Runnable的run中封装了Callable逻辑的调用，以及对其返回结果进行描述的其它操作
* 这使得线程的调用可以有返回结果
*/
public class FutureTask<V> implements RunnableFuture<V> {
private Callable<V> callable; //正在的业务逻辑
private Object outcome; // 业务逻辑返回的结果封装对象
private volatile Thread runner;
//构造函数1 Runnable实现类中引进 业务逻辑
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
//构造函数2 ：封装了业务逻辑的Runnable实现类，转化为可返回结果的Callable
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
/*设置返回结果*/
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
/*
* Runnable实现类的真正业务逻辑运行处，此处封装返回结果，释放资源
* 真正的逻辑 来着从构造函数引进来的Callable接口实现的调用
*/
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 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的泛型*/
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
/*真正的返回值是类成员变量：outcome*/
private V report(int s) throws ExecutionException {
Object x = outcome;
if (s == NORMAL)
return (V)x;
if (s >= CANCELLED)
throw new CancellationException();
throw new ExecutionException((Throwable)x);
}
}
程序清单6-12：AbstractExecutorService中的newTaskFor的默认实现
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
待补充
6.3.3用Future实现页面渲染
为了使渲染器实现更高的并发性，将渲染过程分解为2个任务
- 渲染文本（CPU密集型）
- 渲染图片（IO密集型）
程序清单 6-13 使用Future等待图像被下载
/**
* 当主任务需要图像时，会等待Furture.get的结果，
* 如果幸运的话，开始请求时所有的图片就都下载完了，
* 即使没有，图片下载任务也提请开始了
*/
public class FutureRender {
private final ExecutorService exec = Executors.newFixedThreadPool(100);
void renderPage(CharSequence source) {
final List<ImageInfo> imageInfos = scanForImageInfo(source);
Callable<List<ImageData>> task = new Callable<List<ImageData>>() {
@Override
public List<ImageData> call() throws Exception {
List<ImageData> result = new ArrayList<>();
for(ImageInfo imageInfo : imageInfos)
result.add(imageInfo.downloadImage()); //将图片下载任务提到单独的线程逻辑中，且可返回下载的图像
return result;
}
};
Future<List<ImageData>> future = exec.submit(task); //提交图像下载任务
renderText(source); //文本标签渲染任务在主线程中执行，与图片下载隔离在不同的2个线程中
//将另外一个线程中的下载的图片异步加载并渲染到页面上
try {
List<ImageData> imageDataList = future.get();
for(ImageData image :imageDataList)
renderImage(image);
} catch (InterruptedException e) {
//重新设置线程的中断状态
Thread.currentThread().interrupt();
//由于不需要结果，因此取消任务
future.cancel(true);
} catch (ExecutionException e) {
e.printStackTrace();
}
}
void renderText(CharSequence source) {
System.out.println("渲染文本标签");
}
/**渲染图片*/
void renderImage(ImageData image) {
System.out.println("在浏览器上渲染图片");
}
/**扫描图片信息*/
List<ImageInfo> scanForImageInfo(CharSequence source) {
return new ArrayList<ImageInfo>(100);
}
}
class ImageInfo{
public ImageData downloadImage() {
return new ImageData();
}
}
class ImageData{}
6.3.4异构任务并行化中存在的局限
6.3.5CompletionService：Executor与BlockingQueue
CompletionService将Executor与BlockingQueue结合在一起。将Callable任务提交给它，然后由类似于队列的take pop poll等方法获取已完成的结果。
ExecutorCompletionService实现了CompletionService，并将计算部分委托给Executor。
程序清单 6-14 源码：
1、CompletionService:
package java.util.concurrent;
public interface CompletionService<V> {
Future<V> submit(Callable<V> task);
Future<V> submit(Runnable task, V result);
Future<V> take() throws InterruptedException;
Future<V> poll();
Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
}
2、ExecutorCompletionService:
public class ExecutorCompletionService<V> implements CompletionService<V> {
private final Executor executor;
private final AbstractExecutorService aes;
private final BlockingQueue<Future<V>> completionQueue;
//子类，提交时队列中存结果
private class QueueingFuture extends FutureTask<Void> {
QueueingFuture(RunnableFuture<V> task) {
super(task, null);
this.task = task;
}//子类，提交时队列中存结果
@Override
protected void done() { completionQueue.add(task); }
private final Future<V> task;
}
private RunnableFuture<V> newTaskFor(Callable<V> task) {
if (aes == null)
return new FutureTask<V>(task);
else
return aes.newTaskFor(task);
}
private RunnableFuture<V> newTaskFor(Runnable task, V result) {
if (aes == null)
return new FutureTask<V>(task, result);
else
return aes.newTaskFor(task, result);
}
public ExecutorCompletionService(Executor executor) {
if (executor == null)
throw new NullPointerException();
this.executor = executor;
this.aes = (executor instanceof AbstractExecutorService) ?
(AbstractExecutorService) executor : null;
this.completionQueue = new LinkedBlockingQueue<Future<V>>();
}
public ExecutorCompletionService(Executor executor,
BlockingQueue<Future<V>> completionQueue) {
if (executor == null || completionQueue == null)
throw new NullPointerException();
this.executor = executor;
this.aes = (executor instanceof AbstractExecutorService) ?
(AbstractExecutorService) executor : null;
this.completionQueue = completionQueue;
}
public Future<V> submit(Callable<V> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task);
executor.execute(new QueueingFuture(f));//子类，提交时队列中存结果
return f;
}
public Future<V> submit(Runnable task, V result) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task, result);
executor.execute(new QueueingFuture(f)); //子类，提交时队列中存结果
return f;
}
public Future<V> take() throws InterruptedException {
return completionQueue.take();
}
public Future<V> poll() { return completionQueue.poll();}
public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException {
return completionQueue.poll(timeout, unit);
}
}
示例：使用CompletionService实现页面渲染器
程序清单 6-15 使用CompletionService,使页面在下载完成后立即显示出来
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import com.sun.scenario.effect.ImageData;
public class Renderer {
private final ExecutorService executor;
public Renderer(ExecutorService executor) {
this.executor = executor;
}
public void renderPage(CharSequence source) {
List<ImageInfo> info = scanForImageInfo(executor);
CompletionService<ImageData> completionService = new ExecutorCompletionService<>(executor);
//遍历提交下载逻辑到线程池的所有线程，每个线程执行一个下载
for(final ImageInfo imageInfo : info) {
completionService.submit(new Callable<ImageData>() {
@Override
public ImageData call() throws Exception {
return imageInfo.downloadImage(); //每张图片并发下载
}
});
}
renderText(source); //渲染文本标签
try {
//每张图片分别get
for(int t = 0,n = info.size(); t < n; t++) {
Future<ImageData> f = completionService.take();
ImageData imageData = f.get(); //每下载完一张遍加载渲染一张到页面
renderImage(imageData); //渲染图片到页面
}
} catch(InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
方法一
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < 10000000; i++) {
Runnable task = ....;
exec.submit(task);
}
exec.shutdown();
while(true){
if(exec.isTerminated()){
System.out.println("所有的子线程都结束了！");
break;
}
Thread.sleep(1000);
}
方式二据说性能不好
final CountDownLatch endGate = new CountDownLatch(nThreads);
for(;;){endGate.countDown();}
endGate.await();
方式三
ExecutorService exec = Executors.newCachedThreadPool();
for(;;){exec.submit(task);}
exec.shutdown();
exec.awaitTermination(1, TimeUnit.HOURS);
6.3.7为任务设置时间
如果任务无法在指定时间内完成，将不再需要它的结果。此时可以放弃这个任务。eg:某web 应用从外部的广告服务器上获取广告信息，如果2秒内未得到，则放一个默认的广告上去。
当限时的Future超时时，会跑出一个TimeoutException。
package java.util.concurrent;
public interface Future<V> {
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
程序清单6-16 在指定的时间内获取广告信息
import lombok.Data;
//程序清单6-16 在指定的时间内获取广告信息
public class TimeOutFuture {
private ExecutorService exec = Executors.newFixedThreadPool(10);
private final static long TIME_BUDGET = 2_000_000L;
private final static Ad DEFATULT_AD = new Ad();
Page renderPageWithAd() {
long endNanos = System.nanoTime() + TIME_BUDGET;
Future<Ad> f = exec.submit(new FetchAdTask())
Page page = renderPageBody();
Ad ad;
try {
//只等待指定的时间长度
long timeLeft = endNanos - System.nanoTime();
ad = f.get(timeLeft, TimeUnit.NANOSECONDS);
} catch(ExecutionException e) {
ad = DEFATULT_AD; //默认的
} catch(TimeoutException e) {
ad = DEFATULT_AD; //默认广告
f.cancel(true); //取消任务
}
}
public Page renderPageBody() { return new Page(); }
}
@Data
class Page{
Ad ad;
}
class Ad{}
class FetchAdTask<V> extends FutureTask<V> {
public FetchAdTask(Callable<V> callable) {
super(callable);
}
}
6.3.8 示例：旅行预订门户网站（invokeAll）
程序清单6-17在预定时间内请求旅游报价
//quote 英[kwəʊt] 引用; 引述; 举例说明; 开价; 出价; 报价;
public class QuoteTask implements Callable<TravelQuote> {
private final static ExecutorService exec = Executors.newCachedThreadPool();
private final TravelCompany travelCompany;
private final TravelInfo travelInfo;
public QuoteTask(TravelCompany travelCompany, TravelInfo travelInfo) {
this.travelCompany = travelCompany;
this.travelInfo = travelInfo;
}
/**
* 去多家旅游公司匹配旅行信息
* 正在的运算：Callable的call是应用companies和travelInfo动态计算出来的。
* @param travelInfo 用户输入的旅行信息
* @param companies 旅游公司
* @param comparator 排序规则
* @param timeout 超时时间
* @param timeUnit 超时时间单位
* @return List<TravelQuote> 可供用户选择的旅行报价列表信息
* @throws InterruptedException
*/
public List<TravelQuote> getRankedTranvelQuotes(TravelInfo travelInfo,Set<TravelCompany> companies,
Comparator<TravelQuote> comparator,long timeout, TimeUnit timeUnit) throws InterruptedException {
List<QuoteTask> tasks = new ArrayList<>(); //任务列表
for(TravelCompany company: companies)
tasks.add(new QuoteTask(company, travelInfo)); //把每个公司都和旅行信息匹配组成一个任务并添加到任务集合
List<Future<TravelQuote>> futures = exec.invokeAll(tasks, timeout, timeUnit); //将任务集合统一提交并设置超时时间
List<TravelQuote> quotes = new ArrayList<>(tasks.size()); //可供用户选择的旅行报价列表信息
Iterator<QuoteTask> taskIter = tasks.iterator();
for(Future<TravelQuote> future : futures) {
QuoteTask task = taskIter.next(); //每个报价生成任务，用于下面的显示友好的错误信息报价的生成
try {
quotes.add(future.get());
} catch (ExecutionException e) {
quotes.add(task.getFailureQuote(e));
} catch (CancellationException e) {
quotes.add(task.getTimeOutQuote(e));
}
}
Collections.sort(quotes,comparator);
return quotes;
}
/**旅行公司根据旅行信息生成旅行报价*/
@Override
public TravelQuote call() throws Exception {
return travelCompany.solicitQuote(travelInfo);
}
/**失败的旅行报价：友好信息展示*/
public TravelQuote getFailureQuote(Throwable e) {
return new TravelQuote();
}
/**请求超时时的旅行报价：显示超时友好信息*/
public TravelQuote getTimeOutQuote(Throwable e) {
return new TravelQuote();
}
}
/**
* 旅游公司根据旅行信息发布旅行报价
*/
@Data
class TravelCompany {
private String name;
private String location;
//solicit 英[səˈlɪsɪt] 索求，请求…给予(援助、钱或信息); 征求; 筹集; 招徕(嫖客); 拉(客);
TravelQuote solicitQuote(TravelInfo travelInfo) {
return new TravelQuote();
}
}
/***
* 旅行信息
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
class TravelInfo {
private String from;
private String to;
private Date startDate;
private Date endDate;
private String remark;
}
@Data
@AllArgsConstructor
@NoArgsConstructor
class TravelQuote {
private BigDecimal price;
private String remark;
}
