【Java并发】JAVA并发编程实战-读书笔记10

如果你想Executor提交了一个批处理任务，希望获得结果，只能不断的调用timeout为零的get。幸运的是有一种更好的方法：完成服务（completion
service）

CompletionServie整合了Executor和BlockingQueue的功能。可以将Callable的任务提交给它去执行，然后使用类似队列中的take和poll方法，在结果完整可用时获得这个结果。ExecutorCompletionServie是实现CompletionService接口的一个类，并将计算任务委托给一个Executor。

使用CompletionService，我们可以从两个方面提高页面渲染的性能。我们可以每需要下载一个图像就创建一个独立的任务，并在线程池中执行他们，将顺序的下载过程转换为并行的。而且从CompletionService中获取结果，只要任何一个图像下载完成就立刻呈现。

public class Renderer{

private final ExecutorService executor;

Rnderer(ExecutorService executor){

this.executor=executor;

}

void renderPage(CharSequence source){

final List<ImageInfo> info=scanForImageInfo(source);

CompletionService<ImageData> completion=

new ExecutorCompletionService<ImageData>(executor);

for(final ImageInfo imageInfo:info){

completionService.submit(new Callable<ImageData>(){

public ImageData call(){

return imageInfo.downloadImage();

}

});

}

renderText(source);

try{

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();

}catch(ExecutionException e){

throw launderThrowable(e.getCause());

}

}

}
4000

多个ExecutorCompletionService可以共享单一的Executor，因此一个明智的做法是创建一个ExecutorCompletionService，他对于特定的计算服务是私有的，然后再共享一个公共的Executor。按照这种做法，CompletionService所扮演的批处理计算的句柄与Future所扮演的单一计算的句柄，在很大程度上是一样的。记录下提交给CompletionService的任务个数，然后计算出获得了多少个已完成的结果，这样即使你使用的是共享的Executor，也能知晓什么时候批处理任务的所有结果都已经全部获得。
如果有一个活动无法在指定时间完成，那就是失效了，此时应该放弃这个活动。Futrue.get的限时版本符合这个条件，如果超时了会抛出TimeoutException。

第二个问题是当任务超时的时候要能够停止他们。

Page renderPageWithAd() throws InterruptedException{

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,NANOSECONDS);

}catch(ExecutionException e){

ad=DEFAULT_AD;

}catch(TimeoutException e){

ad=DEFAULT_AD;

f.cancel(true);

}

page.setAd(ad);

return page;

}

cancel中参数true意味着任务线程可以在运行时被中断。
在预定时间内请求旅游报价

private class QuoteTask implements Callable<TravelQuote>{

private final TravelCompany company;

private final TravelInfo travelInfo;

public TravelQuote call()throws Exception{

return company.solicitQuote(travelInfo);

}

}

public List<TravelQuote> getRankedTravelQuotes(

TravelInfo travelInfo,Set<TravelCompany> companies,

Comparator<TravelQuote> ranking,long time,TimeUnit unit){

List<QuoteTask> tasks=new ArrayList<QuoteTask>();

for(TravelCompany company:companies){

tasks.add(new QuoteTask(company,travelInfo));

}

List<Future<TravelQuote>> futures = exec.invokeAll(task,time,unit);

List<TravelQuote> quotes = new ArrayList<TravelQuote>(tasks.size));

Iterator<QuoteTask> taskIter=tasks.iterator();

for(Future<TravelQuote> f:futures){

QuoteTask task=taskIter.next();

try{

quotes.add(f.get());

}catch(ExecutionException e){

quotes.add(task.getFailureQuote(e.getCause()));

}catch(CancellationException e){

quotes.add(task.getTimeoutQuote(e));

}

}

Collections.sore(quotes,ranking);

return quotes;

}

关于任务和线程的停止，Java没有提供任何机制来安全的迫使线程停止工作，她提供中断——一个协作机制使一个线程能够要求另一个线程停止当前的工作。

public class PrimeGenerator implments Runnable{

private final List<BigInteger> primes=new ArrayList<BigInteger>();

private volatile boolean cancelled;

public void run(){

BigInteger p=BigInteger.ONE;

while(!cancelled){

p=p.nextProbablePrime();

synchronized(this){

primes.add(p);

}

}

}

public void cancel(){

cancelled=true;

}

public synchronized List<BigInteger> get(){

return new ArrayList<BigInteger>(primes);

}

}

List<BigInteger> aSecondOfPrimes()throws InterruptedException{

PrimeGenerator generator=new PrimeGenerator();

new Thread(generator).start();

try{

TimeUnit.SECONDS.sleep(1);

}finally{

generator.cancel();

}

return generator.get();

}

不可靠的取消把生产者置于阻塞的操作中

class BrokenPrimeProducer extends Thread{

private final BlockingQueue<BigInteger> queue;

private volatile boolean cancelled=false;

BrokenPrimeProducer(BlockingQueue<BigInteger> queue){

this.queue=queue;

}

public void run(){

try{

BigInteger p=BigInteger.ONE;

while(!cancelled){

queue.put(p=p.nextProbablePrime());

}catch(InterruptedException consumed){

}

}

}

public void cancel(){

cancelled=true;

}

}

void consumePrimes()throws InterruptedException{

BlockingQueue<BigInteger> primes=...

BrokenPrimeProducer producer=new BrokenPrimeProducer(primes);

producer.start();

try{

while(needMorePrimes()){

consume(primes.take());

}

}finally{

producer.cancel();

}

}

上面的例子是一个错误的示范，如果生产者的速度超过了消费者，队列会被填满，put方法会被阻塞，当put方法被阻塞后，消费者如果试图去取消生产者的任务，会调用cancel方法设置cancelled标志，但是此时的生产者永远不会检查这个标志了，因为他已经被put方法阻塞住了。
我们需要制定中断策略，一个中断策略决定线程如何应对中断请求——当发现请求时会做什么，哪些工作单元对于终端来说是原子操作以及在多快的时间里响应中断。

Thread中静态的interrupted应该小心使用，因为他会清除并发线程的终端状态。

public Task getNextTask(BlockingQueue<Task> queue){

boolean interrupted=false;

try{

while(true){

try{

return queue.take();

}catch(InterruptedException e){

interrupted=true;

}

}

}finally{

if(interrupted){

Thread.currentThread().interrupt();

}

}

}

上面的例子演示了不可取消的任务在退出前保存中断。

private static final ScheduledExecutorService cancelExec=...

public static void timedRun(Runnable r,long timeout,TimeUnit unit){

final Thread taskThread=Thread.currentThread();

cancelExec.schedule(new Runnable(){

public void run(){

taskThread.interrupt();

}

});

r.run();

}

上面的例子是错误的演示，因为timedRun可以被任意一个线程调用，但是我们不可能了解调用线程的中断策略，如果任务在时限之前完成，执行中断一个线程可能会出现莫名的问题。另外如果任务不相应终端，timedRun将不会返回直到任务结束，这可能会超过期望的限定时间很久了。

public static void timeRun(final Runnable r,long timeout,TimeUnit unit){

class RethrowableTask implements Runnable{

private volatile Throwable t;

public void run(){

try{

r.run();

}catch(Throwable t){

this.t=t;

}

}

}

}

RethrowableTask task=new RethrowableTask();

final Thread taskThread=new Thread(task);

taskThread.start();

cancelExec.schedule(new Runnable(){

public void run(){

taskThread.interrupt();

}

},timeout,unit);

taskThread.join(unit.toMillis(timeout));

task.rethrow();

上面的例子在新创建的线程中执行带有时限的join方法，因此也受到join不足的影响，因为无论join是否成功完成，在Java存储模型中都会相应存在内存可见性，但是join本身不会返回表示他成功与否的任何状态。