摘要：修改的實現(xiàn)，實現(xiàn)接口在改造筆記一整合到中修改實現(xiàn)，添加對的實現(xiàn)如下負(fù)載過大，處理不過來時，會回調(diào)該方法例如可以發(fā)生郵件通知相關(guān)人員改造筆記四解決中的調(diào)用兩次

在io.moquette.spi.impl.BrokerInterceptor的構(gòu)造函數(shù)中，新建了一個線程池，代碼如下：

private BrokerInterceptor(int poolSize, List handlers) {
        LOG.info("Initializing broker interceptor. InterceptorIds={}", getInterceptorIds(handlers));
        this.handlers = new HashMap<>();
        for (Class messageType : InterceptHandler.ALL_MESSAGE_TYPES) {
            this.handlers.put(messageType, new CopyOnWriteArrayList());
        }
        for (InterceptHandler handler : handlers) {
            this.addInterceptHandler(handler);
        }
        executor = Executors.newFixedThreadPool(poolSize);
    }

executor = Executors.newFixedThreadPool(poolSize);這句代碼雖然創(chuàng)建了一個固定線程數(shù)量的線程池，但是線程池的任務(wù)隊列并沒有做限制，一旦某個InterceptHandler中的某個方法進(jìn)行了耗時處理，在高并發(fā)的情況下，會很容易導(dǎo)致線程池的隊列堆積大量待處理的任務(wù)，進(jìn)而可能造成內(nèi)存溢出。

分別添加以下類和接口

public class ThreadPoolHelper {
    public static ExecutorService createFixedExecutor(int threadNum,int capacity,String threadFactoryName) {
        return new ThreadPoolExecutor(
                threadNum,
                threadNum,
                30,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque(capacity),
                new SimpleThreadFactory(threadFactoryName),
                new LogDiscardRejectPolicy()
        );
    }
}

public class SimpleThreadFactory implements ThreadFactory {
    private static final String NAME_FORMAT = "%s-%s";
    private String threadNamePrefix;

    public SimpleThreadFactory(String threadNamePrefix) {
        this.threadNamePrefix = threadNamePrefix;
    }

    @Override
    public Thread newThread(Runnable runnable) {
        Thread thread = new Thread(runnable);
        thread.setName(String.format(NAME_FORMAT, threadNamePrefix, System.currentTimeMillis()));
        return thread;
    }
}


public class LogDiscardRejectPolicy implements RejectedExecutionHandler {
    private static final Logger LOG = LoggerFactory.getLogger(LogDiscardRejectPolicy.class);

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        LOG.error("executor:{} task queue has full, runnable:{} discarded",executor,r);
        if (!(r instanceof PublishTask)) {
            return;
        }

        PublishTask publishTask = (PublishTask) r;
        InterceptHandler interceptHandler = publishTask.getInterceptHandler();
        if (!(interceptHandler instanceof RejectHandler)) {
            return;
        }

        ((RejectHandler)interceptHandler).rejectedExecution(r,executor);
    }
}


public interface RejectHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

BrokerInterceptor 創(chuàng)建線程池的邏輯改為

private BrokerInterceptor(int poolSize, int capacity, List handlers) {
        LOG.info("Initializing broker interceptor. InterceptorIds={}", getInterceptorIds(handlers));
        this.handlers = new HashMap<>();
        for (Class messageType : InterceptHandler.ALL_MESSAGE_TYPES) {
            this.handlers.put(messageType, new CopyOnWriteArrayList());
        }
        for (InterceptHandler handler : handlers) {
            this.addInterceptHandler(handler);
        }

        /** modify by liuhh */
        executor = ThreadPoolHelper.createFixedExecutor(poolSize, capacity, THREAD_POOL_NAME);
        //executor = Executors.newFixedThreadPool(poolSize);

    }

解釋：
（1）ThreadPoolHelper中的createFixedExecutor()方法為新建的線程池指定任務(wù)隊列大小和拒絕策略LogDiscardRejectPolicy
（2）在LogDiscardRejectPolicy中，首先將被拒絕的任務(wù)log一遍，對于PublishTask（moquette改造筆記(二)：優(yōu)化BrokerInterceptor notifyTopicPublished()邏輯）做特殊處理，會交給實現(xiàn)RejectHandler的InterceptHandler處理，由業(yè)務(wù)邏輯決定，出現(xiàn)任務(wù)太多處理不完被遺棄的任務(wù)該如何處理。

在 moquette改造筆記(一)：整合到SpringBoot 中修改SafetyInterceptHandler實現(xiàn)，添加對RejectHandler的實現(xiàn)如下

@Slf4j
@Component
public class SafetyInterceptHandler extends AbstractInterceptHandler{

    @Override
    public String getID() {
        return SafetyInterceptHandler.class.getName();
    }

    @Override
    public void onConnect(InterceptConnectMessage msg) {
        
    }

    @Override
    public void onConnectionLost(InterceptConnectionLostMessage msg) {
       
    }


    @Override
    public void onPublish(InterceptPublishMessage msg) {
       
    }


    @Override
    public void onMessageAcknowledged(InterceptAcknowledgedMessage msg) {
        
    }
    
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        /**MQTT SERVICE 負(fù)載過大，處理不過來時，會回調(diào)該方法*/
        //例如可以發(fā)生郵件通知相關(guān)人員
    }

}

moquette改造筆記(四)：解決InterceptHandler中的onConnectionLost()調(diào)用兩次