模仿Tomcat的BIO,NIO线程模型 顶

package com.guanjian;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by Administrator on 2018/7/10.
 */
public class ThreadPool {
    private ExecutorService service;
    private List<MessageTask> tasks;
    private int fixedThreadNum = 0;
    private List<String> messages;
    private MessageHandler messageHandler;
    public ThreadPool(int fixedThreadNum,List<String> messages,MessageHandler messageHandler) {
        this.fixedThreadNum = fixedThreadNum;
        this.messages = messages;
        this.messageHandler = messageHandler;
        service = Executors.newFixedThreadPool(fixedThreadNum);
        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                shutdownGracefully(service);
            }
        });
    }
    public void shutdownGracefully(ExecutorService ThreadPool) {
        ShutdownPool.shutdownThreadPool(ThreadPool, "main-pool");
    }

    public void startup() {
        tasks = new ArrayList<>();
        MessageTask messageTask = (fixedThreadNum == 0 ? new SequentialMessageTask(messageHandler,messages) : new ConcurrentMessageTask(messageHandler,messages));
        for (String message:messages) {
            tasks.add(messageTask);
            service.execute(messageTask);
        }
    }

}

package com.guanjian;

/**
 * Created by Administrator on 2018/7/10.
 */
public interface MessageHandler {
    public void execute(String message);
}

package com.guanjian;

/**
 * Created by Administrator on 2018/7/10.
 */
public class MessageHandlerImpl implements MessageHandler {
    @Override
    public void execute(String message) {
        System.out.println(message);
    }
}

package com.guanjian;

import java.util.List;

/**
 * Created by Administrator on 2018/7/10.
 */
public abstract class MessageTask implements Runnable {
    protected MessageHandler messageHandler;
    protected  List<String> messages;

    MessageTask(MessageHandler messageHandler,List<String> messages) {
        this.messageHandler = messageHandler;
        this.messages = messages;
    }
    @Override
    public void run() {
        for (String message:messages) {
            handlerMessage(message);
        }
    }
    protected abstract void handlerMessage(String message);
}

package com.guanjian;

import java.util.List;

/**
 * Created by Administrator on 2018/7/10.
 */
public class SequentialMessageTask extends MessageTask {
    SequentialMessageTask(MessageHandler messageHandler, List<String> messages) {
        super(messageHandler, messages);
    }

    @Override
    protected void handlerMessage(String message) {
        messageHandler.execute(message);
    }
}

package com.guanjian;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by Administrator on 2018/7/10.
 */
public class ConcurrentMessageTask extends MessageTask {
    private ExecutorService asyncService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
    ConcurrentMessageTask(MessageHandler messageHandler, List<String> messages) {
        super(messageHandler, messages);
    }

    @Override
    protected void handlerMessage(String message) {
        asyncService.submit(new Runnable() {
            @Override
            public void run() {
                messageHandler.execute(message);
            }
        });
    }
    protected void shutdown() {
        ShutdownPool.shutdownThreadPool(asyncService,"async-pool-" + Thread.currentThread().getId());
    }
}

package com.guanjian;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * Created by Administrator on 2018/7/10.
 */
public class ShutdownPool {
    private static Logger log = LoggerFactory.getLogger(ThreadPool.class);
    /**
     * 优雅关闭线程池
     * @param threadPool
     * @param alias
     */
    public static void shutdownThreadPool(ExecutorService threadPool, String alias) {
        log.info("Start to shutdown the thead pool: {}", alias);

        threadPool.shutdown(); // 使新任务无法提交.
        try {
            // 等待未完成任务结束
            if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
                threadPool.shutdownNow(); // 取消当前执行的任务
                log.warn("Interrupt the worker, which may cause some task inconsistent. Please check the biz logs.");

                // 等待任务取消的响应
                if (!threadPool.awaitTermination(60, TimeUnit.SECONDS))
                    log.error("Thread pool can't be shutdown even with interrupting worker threads, which may cause some task inconsistent. Please check the biz logs.");
            }
        } catch (InterruptedException ie) {
            // 重新取消当前线程进行中断
            threadPool.shutdownNow();
            log.error("The current server thread is interrupted when it is trying to stop the worker threads. This may leave an inconcistent state. Please check the biz logs.");

            // 保留中断状态
            Thread.currentThread().interrupt();
        }

        log.info("Finally shutdown the thead pool: {}", alias);
    }
}