快捷導(dǎo)航

Netty進階之EventExecutorGroup源碼詳解

更新時間：2023年11月16日 10:44:19 作者：立小研先森

這篇文章主要介紹了Netty進階之EventExecutorGroup源碼詳解,EventExecutorGroup繼承了JDK的ScheduledExecutroService,那么它就擁有了執(zhí)行定時任務(wù),執(zhí)行提交的普通任務(wù),需要的朋友可以參考下

前言

EventExecutorGroup繼承了JDK的ScheduledExecutroService，那么它就擁有了執(zhí)行定時任務(wù)，執(zhí)行提交的普通任務(wù)；

EventExecutorGroup還繼承了JDK的Iterable接口，表示EventExecutorGroup是可遍歷的，它的遍歷對象是EventExecutor;

一、Iterable相關(guān)方法

EventExecutorGroup中有兩個Iterable相關(guān)的方法：

    //返回一個被當(dāng)前EventExecutorGroup管理的EventExecutor對象
		EventExecutor next();
    //返回一個EventExecutor類型的迭代器
    @Override
    Iterator<EventExecutor> iterator();

二、execute

接口Executor提供了一個提交執(zhí)行任務(wù)的方法execute

    //返回一個被當(dāng)前EventExecutorGroup管理的EventExecutor對象
		EventExecutor next();
    //返回一個EventExecutor類型的迭代器
    @Override
    Iterator<EventExecutor> iterator();

三、Executor的子接口ExecutorService

public interface ExecutorService extends Executor {

    /**
     * 啟動有序關(guān)閉，在關(guān)閉過程中會繼續(xù)執(zhí)行以前提交的任務(wù)，但是不接受新的任務(wù)
     */
    void shutdown();

    /**
     * 嘗試停止所有正在執(zhí)行的任務(wù)，停止處理等待的任務(wù)，并返回一個等待執(zhí)行的任務(wù)列表
     */
    List<Runnable> shutdownNow();

    /**
     * 如果此執(zhí)行程序已關(guān)閉，則返回true
     */
    boolean isShutdown();

    /**
     * 如果關(guān)閉后所有任務(wù)都已完成，則返回true
     * 只有首先調(diào)用了shutdown()、shutdownNow()方法才有可能返回true，否則一定為false
     */
    boolean isTerminated();

    /**
     * 阻塞，shutdown后所有任務(wù)執(zhí)行完成、超時、當(dāng)前線程被終端，那個先發(fā)生那個優(yōu)先；
     */
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;

    /**
     * 提交一個帶返回值的執(zhí)行任務(wù)，并返回一個Future代表將要返回的任務(wù)結(jié)果
     */
    <T> Future<T> submit(Callable<T> task);

    /**
     * 提交一個可運行任務(wù)以供執(zhí)行，并返回一個表示該任務(wù)的Future。Future的get方法將在成功完成后返回給定的結(jié)果。
     */
    <T> Future<T> submit(Runnable task, T result);

    /**
     * 提交一個可運行任務(wù)以供執(zhí)行，并返回一個表示該任務(wù)的Future。Future的get方法在成功完成后將返回null。
     */
    Future<?> submit(Runnable task);

    /**
     * 執(zhí)行給定的多個任務(wù)，返回一個持有執(zhí)行狀態(tài)和結(jié)果的Future列表。
     */
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;

    /**
     * 執(zhí)行給定的多個任務(wù)，返回一個持有執(zhí)行狀態(tài)和結(jié)果的Future列表。
     * 當(dāng)所有的任務(wù)完成或超時過期Future#isDone都將返回ture
     */
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

    /**
     * 執(zhí)行給定的多個任務(wù)，返回其中一個執(zhí)行成功的結(jié)果
     */
    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

    /**
     * 執(zhí)行給定的多個任務(wù)，返回其中一個執(zhí)行成功的結(jié)果
     */
    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

下面是一個網(wǎng)絡(luò)服務(wù)案例，其中線程池中的線程為傳入請求提供服務(wù)，它使用預(yù)先配置的Executors.newFixedThreadPool工廠方法：

class NetworkService implements Runnable {
    private final ServerSocket serverSocket;
    private final ExecutorService pool;

    public NetworkService(int port, int poolSize) throws IOException {
        serverSocket = new ServerSocket(port);
        pool = Executors.newFixedThreadPool(poolSize);
    }

    public void run() { // run the service      
        try {
            for (; ; ) {
                pool.execute(new Handler(serverSocket.accept()));
            }
        } catch (IOException ex) {
            pool.shutdown();
        }
    }
}

class Handler implements Runnable {
    private final Socket socket;

    Handler(Socket socket) {
        this.socket = socket;
    }

    public void run() {
        // read and service request on socket   
    }
}

下面的方法關(guān)閉ExecutorService分兩個階段，首先通過調(diào)用shutdown方法拒絕新任務(wù)進來，然后調(diào)用shutdownNow，如果有必要取消任何延遲任務(wù)：

    void shutdownAndAwaitTermination(ExecutorService pool) {
        pool.shutdown(); // Disable new tasks from being submitted    
        try {
            // Wait a while for existing tasks to terminate      
            if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
                pool.shutdownNow(); // Cancel currently executing tasks        
                // Wait a while for tasks to respond to being cancelled      
                if (!pool.awaitTermination(60, TimeUnit.SECONDS))
                    System.err.println("Pool did not terminate");
            }
        } catch (InterruptedException ie) {
            // (Re-)Cancel if current thread also interrupted      
            pool.shutdownNow();      // Preserve interrupt status     
            Thread.currentThread().interrupt();
        }
    }

四、ExecutorService的子接口ScheduledExecutorService

ScheduledExecutorService接口提供了四個新的方法schedule延遲指定的時間執(zhí)行任務(wù)，scheduleAtFixedRate方法延遲指定時間、按照固定的時間頻率執(zhí)行任務(wù)。

public interface ScheduledExecutorService extends ExecutorService {
    /**
     * 提交一個一次性任務(wù)，該任務(wù)在給定延遲后變?yōu)閱⒂脿顟B(tài)。
     *
     * @param command 要執(zhí)行的任務(wù)
     * @param delay 從現(xiàn)在開始延遲執(zhí)行的時間
     * @param unit delay參數(shù)的時間單位
     * @return 一個ScheduledFuture，標(biāo)識任務(wù)的掛起完成，其get方法將在完成時返回null
     */
    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay, TimeUnit unit);
    /**
     * 提交一個帶返回值的一次性任務(wù)，該任務(wù)在給定延遲后變?yōu)閱⒂脿顟B(tài)
     *
     * @param callable 要執(zhí)行的函數(shù)任務(wù)
     * @param delay 從現(xiàn)在開始延遲執(zhí)行的時間
     * @param unit delay參數(shù)的時間單位
     * @param <V> the type of the callable's result
     * @return 可用于提取結(jié)果或取消的ScheduledFuture
     */
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay, TimeUnit unit);
    /**
     * 提交一個可執(zhí)行任務(wù)，延遲指定時間，然后按照period時間間隔執(zhí)行任務(wù)
     */
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);
    /**
     * 提交一個可執(zhí)行任務(wù)，延遲指定時間，然后按照period時間間隔執(zhí)行任務(wù)
     */
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);
}