第六章并发编程,异步执行框架executor


异步执行框架executor是一个接口,只有一个方法。接受一个Runnable做为参数,执行任务。 将任务的执行与提交解耦。

1:executor

package java.util.concurrent;

public interface Executor {
    void execute(Runnable var1);
}

2:使用executor 创建一个任务,并执行。

package chaptor06;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class TaskExecutionWebServer {
    private static final int NTHREADS=100;
    private static final Executor exec = Executors.newFixedThreadPool(NTHREADS);

    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(100);

        while (true) {
            final Socket connection=serverSocket.accept();
            Runnable task=new Runnable() {  //Runnable 不处理异常,callable 处理异常。
                @Override
                public void run() {
                    //handlerRequest (connection);
                }
            };

            exec.execute(task);  //将Runnable 交给executor
        }
    }
}

3:因为executor不能对线程进行进行操作,例如终止等。所以executorService继承了 executor。executorService 可以接收callable对象,callable能够处理异常。

     能够返回含有将来结果的 future等。能够使该框架执行的线程终止。

//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//

package java.util.concurrent;

import java.util.Collection;
import java.util.List;

public interface ExecutorService extends Executor {
    void shutdown();

    List<Runnable> shutdownNow();

    boolean isShutdown();

    boolean isTerminated();

    boolean awaitTermination(long var1, TimeUnit var3) throws InterruptedException;

    <T> Future<T> submit(Callable<T> var1);

    <T> Future<T> submit(Runnable var1, T var2);

    Future<?> submit(Runnable var1);

    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> var1) throws InterruptedException;

    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> var1, long var2, TimeUnit var4) throws InterruptedException;

    <T> T invokeAny(Collection<? extends Callable<T>> var1) throws InterruptedException, ExecutionException;

    <T> T invokeAny(Collection<? extends Callable<T>> var1, long var2, TimeUnit var4) throws InterruptedException, ExecutionException, TimeoutException;
}

4:completionService将executor和blockingqueue的功能融合在一起,可以将(一组)callable任务交给它,然后用take和poll方法获得结果future。

      

//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//

package java.util.concurrent;

public interface CompletionService<V> {
    Future<V> submit(Callable<V> var1);

    Future<V> submit(Runnable var1, V var2);  //提交任务

    Future<V> take() throws InterruptedException;  //获得结果

    Future<V> poll();

    Future<V> poll(long var1, TimeUnit var3) throws InterruptedException;
}

 5:为任务设置时限,

    当某个任务获得的结果超过一定时间时,就舍弃它,可以通过future.get方法实现。传递给一个时间限制。

    V get(long var1, TimeUnit var3) throws InterruptedException, ExecutionException, TimeoutException;

6:当一组任务提交,访问不同网站的数据库等,可以设置一个时限,当到达时限时,只要取到的结果,如果某些任务没有取到结果,那么将它们舍弃。可以用executorService中的invokeall方法。

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> var1, long var2, TimeUnit var4) throws InterruptedException;
ExecutorService 构建多线程生产者消费者

https://github.com/1367356/GradleTestUseSubModule/tree/master/javaConcurrency/src/main/java/MessageQueue/one

原文地址:https://www.cnblogs.com/liyafei/p/8758523.html