JoyLau's Blog

@Data
@Configuration
@ConfigurationProperties(prefix = "traffic.executor")
public class Executor {
    private String name;

    private Integer corePoolSize;

    private Integer maxPoolSize;

    private Integer queueCapacity;

    private String threadNamePrefix;
}

@Configuration
@ConditionalOnBean(Executor.class)
public class ExecutorConfig {
    @Bean
    public ThreadPoolTaskExecutor trafficServiceExecutor(@Autowired Executor executor) {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(executor.getCorePoolSize());
        threadPoolTaskExecutor.setMaxPoolSize(executor.getMaxPoolSize());
        threadPoolTaskExecutor.setQueueCapacity(executor.getQueueCapacity());
        threadPoolTaskExecutor.setThreadNamePrefix(executor.getThreadNamePrefix());
        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }
}

@Slf4j
public class MyExecutor extends ExecutorConfigurationSupport
        implements AsyncListenableTaskExecutor, SchedulingTaskExecutor {

    private final Object poolSizeMonitor = new Object();

    private int corePoolSize = 1;

    private int maxPoolSize = Integer.MAX_VALUE;

    private int keepAliveSeconds = 60;

    private int queueCapacity = Integer.MAX_VALUE;

    private boolean allowCoreThreadTimeOut = false;

    @Nullable
    private TaskDecorator taskDecorator;

    @Nullable
    private ThreadPoolExecutor threadPoolExecutor;

    // Runnable decorator to user-level FutureTask, if different
    private final Map<Runnable, Object> decoratedTaskMap =
            new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);


    public void setCorePoolSize(int corePoolSize) {
        synchronized (this.poolSizeMonitor) {
            this.corePoolSize = corePoolSize;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.setCorePoolSize(corePoolSize);
            }
        }
    }

    public int getCorePoolSize() {
        synchronized (this.poolSizeMonitor) {
            return this.corePoolSize;
        }
    }

    public void setMaxPoolSize(int maxPoolSize) {
        synchronized (this.poolSizeMonitor) {
            this.maxPoolSize = maxPoolSize;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.setMaximumPoolSize(maxPoolSize);
            }
        }
    }

    public int getMaxPoolSize() {
        synchronized (this.poolSizeMonitor) {
            return this.maxPoolSize;
        }
    }

    public void setKeepAliveSeconds(int keepAliveSeconds) {
        synchronized (this.poolSizeMonitor) {
            this.keepAliveSeconds = keepAliveSeconds;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.setKeepAliveTime(keepAliveSeconds, TimeUnit.SECONDS);
            }
        }
    }

    public int getKeepAliveSeconds() {
        synchronized (this.poolSizeMonitor) {
            return this.keepAliveSeconds;
        }
    }

    public void setQueueCapacity(int queueCapacity) {
        this.queueCapacity = queueCapacity;
    }

    public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
        this.allowCoreThreadTimeOut = allowCoreThreadTimeOut;
    }

    public void setTaskDecorator(TaskDecorator taskDecorator) {
        this.taskDecorator = taskDecorator;
    }


    @Override
    protected ExecutorService initializeExecutor(
            ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

        BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);

        ThreadPoolExecutor executor;
        if (this.taskDecorator != null) {
            executor = new ThreadPoolExecutor(
                    this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
                    queue, threadFactory, rejectedExecutionHandler) {
                @Override
                public void execute(Runnable command) {
                    Runnable decorated = taskDecorator.decorate(command);
                    if (decorated != command) {
                        decoratedTaskMap.put(decorated, command);
                    }
                    super.execute(decorated);
                }

            };
        }
        else {
            executor = new ThreadPoolExecutor(
                    this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
                    queue, threadFactory, rejectedExecutionHandler){
                @Override
                public void beforeExecute(Thread t, Runnable r) {
//                    log.error("线程开始......");
//                    log.error("当前线程池的线程数量:{}",MyExecutor.this.getPoolSize());
//                    log.error("活动的线程的数量:{}",MyExecutor.this.getActiveCount());
//                    log.error("线程需要执行的任务个数:{}",getTaskCount());
//                    log.error("线程池在运行过程中已完成的任务数:{}",getCompletedTaskCount());
                }
                @Override
                public void afterExecute(Runnable r, Throwable t) {
                    log.error("线程池在运行过程中已完成的任务数:{}",getCompletedTaskCount());
                }
            };

        }

        if (this.allowCoreThreadTimeOut) {
            executor.allowCoreThreadTimeOut(true);
        }

        this.threadPoolExecutor = executor;
        return executor;
    }

    protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
        if (queueCapacity > 0) {
            return new LinkedBlockingQueue<>(queueCapacity);
        }
        else {
            return new SynchronousQueue<>();
        }
    }

    public ThreadPoolExecutor getThreadPoolExecutor() throws IllegalStateException {
        Assert.state(this.threadPoolExecutor != null, "ThreadPoolTaskExecutor not initialized");
        return this.threadPoolExecutor;
    }

    public int getPoolSize() {
        if (this.threadPoolExecutor == null) {
            // Not initialized yet: assume core pool size.
            return this.corePoolSize;
        }
        return this.threadPoolExecutor.getPoolSize();
    }

    public int getActiveCount() {
        if (this.threadPoolExecutor == null) {
            // Not initialized yet: assume no active threads.
            return 0;
        }
        return this.threadPoolExecutor.getActiveCount();
    }


    @Override
    public void execute(Runnable task) {
        Executor executor = getThreadPoolExecutor();
        try {
            executor.execute(task);
        }
        catch (RejectedExecutionException ex) {
            throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
        }
    }

    @Override
    public void execute(Runnable task, long startTimeout) {
        execute(task);
    }

    @Override
    public Future<?> submit(Runnable task) {
        ExecutorService executor = getThreadPoolExecutor();
        try {
            return executor.submit(task);
        }
        catch (RejectedExecutionException ex) {
            throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
        }
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        ExecutorService executor = getThreadPoolExecutor();
        try {
            return executor.submit(task);
        }
        catch (RejectedExecutionException ex) {
            throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
        }
    }

    @Override
    public ListenableFuture<?> submitListenable(Runnable task) {
        ExecutorService executor = getThreadPoolExecutor();
        try {
            ListenableFutureTask<Object> future = new ListenableFutureTask<>(task, null);
            executor.execute(future);
            return future;
        }
        catch (RejectedExecutionException ex) {
            throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
        }
    }

    @Override
    public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
        ExecutorService executor = getThreadPoolExecutor();
        try {
            ListenableFutureTask<T> future = new ListenableFutureTask<>(task);
            executor.execute(future);
            return future;
        }
        catch (RejectedExecutionException ex) {
            throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
        }
    }

    @Override
    protected void cancelRemainingTask(Runnable task) {
        super.cancelRemainingTask(task);
        // Cancel associated user-level Future handle as well
        Object original = this.decoratedTaskMap.get(task);
        if (original instanceof Future) {
            ((Future<?>) original).cancel(true);
        }
    }
}

/**
 * 自动完成提示
 * @param search search
 * @return MessageBody
 */
public MessageBody autoCompleteKnowledgeInfo(KnowledgeSearch search) {
    //设置搜索建议
    CompletionSuggestionBuilder infoTitleSuggestion = new CompletionSuggestionBuilder("infoTitle.suggest")
            .text(search.getQuery())
            .size(6);
    CompletionSuggestionBuilder infoKeywordsSuggestion = new CompletionSuggestionBuilder("infoKeywords.suggest")
            .text(search.getQuery())
            .size(6);
    CompletionSuggestionBuilder infoSummarySuggestion = new CompletionSuggestionBuilder("infoSummary.suggest")
            .text(search.getQuery())
            .size(6);
    CompletionSuggestionBuilder infoFileNameSuggestion = new CompletionSuggestionBuilder("infoFileName.suggest")
            .text(search.getQuery())
            .size(6);
    SuggestBuilder suggestBuilder = new SuggestBuilder()
            .addSuggestion("标题", infoTitleSuggestion)
            .addSuggestion("关键字", infoKeywordsSuggestion)
            .addSuggestion("摘要", infoSummarySuggestion)
            .addSuggestion("附件",infoFileNameSuggestion);
    SearchRequestBuilder searchRequest = client.prepareSearch(ES_KNOWLEDGE_INDEX)
            .setFetchSource(false)
            .suggest(suggestBuilder);
    List<JSONObject> list = new ArrayList<>();

    //查询结果
    SearchResponse searchResponse = searchRequest.get();

    /*没查到结果*/
    if (searchResponse.getSuggest() == null) {
        return MessageBody.success(list);
    }
    searchResponse.getSuggest().forEach(entries -> {
        String name = entries.getName();
        for (Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option> entry : entries) {
            for (Suggest.Suggestion.Entry.Option option : entry.getOptions()) {
                JSONObject object = new JSONObject();
                object.put("name",name);
                object.put("text",option.getText().string());
                list.add(object);
            }
        }
    });
    return MessageBody.success(list);
}

TimeValue keepAlive = TimeValue.timeValueMinutes(30);
    SearchRequestBuilder searchRequest = client.prepareSearch(ES_KNOWLEDGE_INDEX)
            .setScroll(keepAlive)
            .setSize(10000);
    SearchResponse searchResponse = searchRequest.get();
    do {
        //处理的业务 saveIds(searchResponse);
        searchResponse = client.prepareSearchScroll(searchResponse.getScrollId()).setScroll(keepAlive).execute()
                .actionGet();
    } while (searchResponse.getHits().getHits().length != 0);