msb_6557
/
opsli-boot
mirror of https://github.com/hiparker/opsli-boot.git
1
1
Code Issues Projects Releases Wiki Activity

perf: 优化线程池处理工具

Browse Source
pull/9/head
Carina 3 years ago
parent 66f151ccb8
commit c2138f379c
11 changed files with 573 additions and 227 deletions
Show Stats Download Patch File Download Diff File

25
opsli-api/src/main/java/org/opsli/api/web/gentest/carinfo/TestCarRestApi.java
Unescape View File

@ -16,10 +16,7 @@
package org.opsli.api.web.gentest.carinfo;
import org.opsli.api.base.result.ResultVo;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartHttpServletRequest;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
@ -44,13 +41,15 @@ public interface TestCarRestApi {
String TITLE = "汽车信息管理";
/** 子标题 */
String SUB_TITLE = "汽车信息";
/** URL 前缀 */
String URL_PREFIX = "/gentest/carinfo/{version}";
/**
* 汽车信息 查一条
* @param model 模型
* @return ResultVo
*/
@GetMapping("/get")
@GetMapping(URL_PREFIX + "/get")
ResultVo<TestCarModel> get(TestCarModel model);
/**
@ -60,7 +59,7 @@ public interface TestCarRestApi {
* @param request request
* @return ResultVo
*/
@GetMapping("/findPage")
@GetMapping(URL_PREFIX + "/findPage")
ResultVo<?> findPage(
@RequestParam(name = "pageNo", defaultValue = "1") Integer pageNo,
@RequestParam(name = "pageSize", defaultValue = "10") Integer pageSize,
@ -72,7 +71,7 @@ public interface TestCarRestApi {
* @param model 模型
* @return ResultVo
*/
@PostMapping("/insert")
@PostMapping(URL_PREFIX + "/insert")
ResultVo<?> insert(@RequestBody TestCarModel model);
/**
@ -80,7 +79,7 @@ public interface TestCarRestApi {
* @param model 模型
* @return ResultVo
*/
@PostMapping("/update")
@PostMapping(URL_PREFIX + "/update")
ResultVo<?> update(@RequestBody TestCarModel model);
/**
@ -88,7 +87,7 @@ public interface TestCarRestApi {
* @param id ID
* @return ResultVo
*/
@PostMapping("/del")
@PostMapping(URL_PREFIX + "/del")
ResultVo<?> del(String id);
/**
@ -96,7 +95,7 @@ public interface TestCarRestApi {
* @param ids ID 数组
* @return ResultVo
*/
@PostMapping("/delAll")
@PostMapping(URL_PREFIX + "/delAll")
ResultVo<?> delAll(String ids);
/**
@ -112,7 +111,7 @@ public interface TestCarRestApi {
* @param request request
* @param response response
*/
@GetMapping("/exportExcel")
@GetMapping(URL_PREFIX + "/exportExcel")
void exportExcel(HttpServletRequest request, HttpServletResponse response);
/**
@ -120,14 +119,14 @@ public interface TestCarRestApi {
* @param request 文件流 request
* @return ResultVo
*/
@PostMapping("/importExcel")
@PostMapping(URL_PREFIX + "/importExcel")
ResultVo<?> importExcel(MultipartHttpServletRequest request);
/**
* 汽车信息 Excel 下载导入模版
* @param response response
*/
@GetMapping("/importExcel/template")
@GetMapping(URL_PREFIX + "/importExcel/template")
void importTemplate(HttpServletResponse response);
}

18
opsli-base-support/opsli-common/src/main/java/org/opsli/common/annotation/ApiVersion.java
Unescape View File

@ -0,0 +1,18 @@
package org.opsli.common.annotation;
import java.lang.annotation.*;
/**
* 标识版本号，从1开始
*
* @author Parker
* @date 2021年10月27日12:36:55
*/
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface ApiVersion {
int value() default 1;
}

22
opsli-base-support/opsli-common/src/main/java/org/opsli/common/thread/AsyncProcessExecutor.java
Unescape View File

@ -1,3 +1,18 @@
/**
* Copyright 2020 OPSLI 快速开发平台 https://www.opsli.com
* <p>
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
* use this file except in compliance with the License. You may obtain a copy of
* the License at
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations under
* the License.
*/
package org.opsli.common.thread;
import java.util.function.Function;
@ -24,11 +39,4 @@ public interface AsyncProcessExecutor {
*/
boolean execute();
/**
* 执行异常回调
* @param callback 回调
* @return boolean
*/
boolean executeErrorCallback(Function<Runnable, Void> callback);
}

23
opsli-base-support/opsli-common/src/main/java/org/opsli/common/thread/AsyncProcessExecutorByNormal.java
Unescape View File

@ -94,26 +94,6 @@ public class AsyncProcessExecutorByNormal implements AsyncProcessExecutor{
return true;
}
@Override
public boolean executeErrorCallback(Function<Runnable, Void> callback) {
if(CollUtil.isEmpty(this.taskList)){
return true;
}
for (Runnable task : this.taskList) {
// 多线程执行任务
boolean execute = this.execute(task);
// 执行任务被拒绝 门闩减1 计数器不动 End
if(!execute){
// 线程池失败后 返回该 Runnable
callback.apply(task);
}
}
return false;
}
// ====================================
/**
@ -134,7 +114,8 @@ public class AsyncProcessExecutorByNormal implements AsyncProcessExecutor{
private synchronized static AsyncProcessor getProcessor(String key){
AsyncProcessor asyncProcessor = EXECUTOR_MAP.get(key);
if(null == asyncProcessor){
asyncProcessor = new AsyncProcessor(key);
asyncProcessor = new AsyncProcessor();
asyncProcessor.init(key);
EXECUTOR_MAP.put(key, asyncProcessor);
}
return asyncProcessor;

172
opsli-base-support/opsli-common/src/main/java/org/opsli/common/thread/AsyncProcessExecutorByWait.java
Unescape View File

@ -16,21 +16,21 @@
package org.opsli.common.thread;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Maps;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
/**
* 多线程锁执行器
* 用于当前方法中复杂业务多线程处理，等待线程执行完毕后 获得统一结果
* 2021年11月2日14:07:54 重构 多线程异步等待执行器
*
* @author 周鹏程
* @date 2020-12-10 10:36
@ -44,19 +44,23 @@ public class AsyncProcessExecutorByWait implements AsyncProcessExecutor {
/** 线程Key */
private final String key;
/** 任务执行计数器 */
private AtomicInteger count;
/** 任务队列 */
private final List<Runnable> taskList;
private final List<Callable<Object>> taskList;
/** 执行器 */
private final AsyncProcessor processor;
/**
* 构造函数
*/
public AsyncProcessExecutorByWait(){
this.key = "def";
taskList = new ArrayList<>();
processor = AsyncProcessExecutorByWait.getProcessor(this.key);
processor = getProcessor(this.key);
}
/**
@ -66,17 +70,18 @@ public class AsyncProcessExecutorByWait implements AsyncProcessExecutor {
public AsyncProcessExecutorByWait(String key){
this.key = key;
taskList = new ArrayList<>();
processor = AsyncProcessExecutorByWait.getProcessor(this.key);
processor = getProcessor(this.key);
}
/**
* 执行
* 放入执行任务
* 特殊处理 Runnable 转换为 Callable
* @param task 任务
*/
@Override
public AsyncProcessExecutorByWait put(final Runnable task){
taskList.add(task);
public AsyncProcessExecutor put(final Runnable task){
taskList.add(Executors.callable(task));
return this;
}
@ -89,58 +94,25 @@ public class AsyncProcessExecutorByWait implements AsyncProcessExecutor {
return true;
}
// 锁
AsyncWaitLock lock = new AsyncWaitLock(this.taskList.size());
for (Runnable task : this.taskList) {
// 多线程执行任务
boolean execute = this.execute(task, lock);
// 执行任务被拒绝 门闩减1 计数器不动 End
if(!execute){
lock.getLatch().countDown();
}
}
// 线程锁 等待查询结果 结果完成后继续执行
try {
lock.getLatch().await();
}catch (Exception e){
log.error(e.getMessage(), e);
}finally {
this.taskList.clear();
}
// 返回执行结果
return lock.getCount().get() == 0;
}
/**
* 执行 线程锁 等待查询结果 结果完成后继续执行
*/
@Override
public boolean executeErrorCallback(Function<Runnable, Void> callback){
if(CollUtil.isEmpty(this.taskList)){
return true;
}
// 锁
AsyncWaitLock lock = new AsyncWaitLock(this.taskList.size());
for (Runnable task : this.taskList) {
// 多线程执行任务
boolean execute = this.execute(task, lock);
// 执行任务被拒绝 门闩减1 计数器不动 End
if(!execute){
// 线程池失败后 返回该 Runnable
callback.apply(task);
// 初始化锁参数
count = new AtomicInteger(this.taskList.size());
// 门闩 线程锁
CountDownLatch latch = new CountDownLatch(this.taskList.size());
lock.getLatch().countDown();
}
for (Callable<Object> task : this.taskList) {
// 回调减 门闩
processor.executeTaskAndCallback(task, (result)->{
if(result.getSuccess()){
count.decrementAndGet();
}
latch.countDown();
return null;
});
}
// 线程锁 等待查询结果 结果完成后继续执行
try {
lock.getLatch().await();
latch.await();
}catch (Exception e){
log.error(e.getMessage(), e);
}finally {
@ -148,18 +120,7 @@ public class AsyncProcessExecutorByWait implements AsyncProcessExecutor {
}
// 返回执行结果
return lock.getCount().get() == 0;
}
/**
* 执行指定的任务
*
* @param task 任务
* @return boolean
*/
private boolean execute(final Runnable task, final AsyncWaitLock lock) {
return processor.executeTask(new TaskWrapper(task, lock));
return count.get() == 0;
}
/**
@ -170,82 +131,11 @@ public class AsyncProcessExecutorByWait implements AsyncProcessExecutor {
private synchronized static AsyncProcessor getProcessor(String key){
AsyncProcessor asyncProcessor = EXECUTOR_MAP.get(key);
if(null == asyncProcessor){
asyncProcessor = new AsyncProcessor(key);
asyncProcessor = new AsyncProcessor();
asyncProcessor.init(key);
EXECUTOR_MAP.put(key, asyncProcessor);
}
return asyncProcessor;
}
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
/**
* 线程锁对象
*
* @author Parker
* @date 2020-10-08 10:24
*/
@Getter
public static class AsyncWaitLock {
/** 门闩 */
private final CountDownLatch latch;
/** 计数器 */
private final AtomicInteger count;
public AsyncWaitLock(int count){
// 初始化锁参数
this.count = new AtomicInteger(count);
// 门闩 线程锁
this.latch = new CountDownLatch(count);
}
}
/**
* Task 包装类<br>
* 此类型的意义是记录可能会被 Executor 吃掉的异常<br>
*/
public static class TaskWrapper implements Runnable {
private final Runnable gift;
private final CountDownLatch latch;
private final AtomicInteger count;
public TaskWrapper(final Runnable target) {
this.gift = target;
this.count = null;
this.latch = null;
}
public TaskWrapper(final Runnable target, final AsyncWaitLock lock) {
if (lock == null) {
this.gift = null;
this.count = null;
this.latch = null;
return;
}
this.gift = target;
this.count = lock.getCount();
this.latch = lock.getLatch();
}
@Override
public void run() {
// 捕获异常，避免在 Executor 里面被吞掉了
if (gift != null) {
try {
gift.run();
// 标示已执行
count.decrementAndGet();
} catch (Exception e) {
String errMsg = StrUtil.format("线程池-包装的目标执行异常: {}", e.getMessage());
log.error(errMsg, e);
} finally {
latch.countDown();
}
}
}
}
}

214
opsli-base-support/opsli-common/src/main/java/org/opsli/common/thread/AsyncProcessExecutorOldByWait.java
Unescape View File

@ -0,0 +1,214 @@
/**
* Copyright 2020 OPSLI 快速开发平台 https://www.opsli.com
* <p>
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
* use this file except in compliance with the License. You may obtain a copy of
* the License at
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations under
* the License.
*/
package org.opsli.common.thread;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Maps;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 多线程锁执行器
* 用于当前方法中复杂业务多线程处理，等待线程执行完毕后 获得统一结果
*
* @author 周鹏程
* @date 2020-12-10 10:36
*/
@Slf4j
@Deprecated
public class AsyncProcessExecutorOldByWait implements AsyncProcessExecutor {
/** 线程池字典 */
private static final Map<String, AsyncProcessor> EXECUTOR_MAP = Maps.newConcurrentMap();
/** 线程Key */
private final String key;
/** 任务队列 */
private final List<Runnable> taskList;
/** 执行器 */
private final AsyncProcessor processor;
/**
* 构造函数
*/
public AsyncProcessExecutorOldByWait(){
this.key = "def";
taskList = new ArrayList<>();
processor = getProcessor(this.key);
}
/**
* 构造函数
* @param key 线程池唯一Key
*/
public AsyncProcessExecutorOldByWait(String key){
this.key = key;
taskList = new ArrayList<>();
processor = getProcessor(this.key);
}
/**
* 执行
* @param task 任务
*/
@Override
public AsyncProcessExecutor put(final Runnable task){
taskList.add(task);
return this;
}
/**
* 执行 线程锁 等待查询结果 结果完成后继续执行
*/
@Override
public boolean execute(){
if(CollUtil.isEmpty(this.taskList)){
return true;
}
// 锁
AsyncWaitLock lock = new AsyncWaitLock(this.taskList.size());
for (Runnable task : this.taskList) {
// 多线程执行任务
boolean execute = this.execute(task, lock);
// 执行任务被拒绝 门闩减1 计数器不动 End
if(!execute){
lock.getLatch().countDown();
}
}
// 线程锁 等待查询结果 结果完成后继续执行
try {
lock.getLatch().await();
}catch (Exception e){
log.error(e.getMessage(), e);
}finally {
this.taskList.clear();
}
// 返回执行结果
return lock.getCount().get() == 0;
}
/**
* 执行指定的任务
*
* @param task 任务
* @return boolean
*/
private boolean execute(final Runnable task, final AsyncWaitLock lock) {
return processor.executeTask(new TaskWrapper(task, lock));
}
/**
* 获得执行器
* @param key Key
* @return AsyncProcessor
*/
private synchronized static AsyncProcessor getProcessor(String key){
AsyncProcessor asyncProcessor = EXECUTOR_MAP.get(key);
if(null == asyncProcessor){
asyncProcessor = new AsyncProcessor();
asyncProcessor.init(key);
EXECUTOR_MAP.put(key, asyncProcessor);
}
return asyncProcessor;
}
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
/**
* 线程锁对象
*
* @author Parker
* @date 2020-10-08 10:24
*/
@Getter
public static class AsyncWaitLock {
/** 门闩 */
private final CountDownLatch latch;
/** 计数器 */
private final AtomicInteger count;
public AsyncWaitLock(int count){
// 初始化锁参数
this.count = new AtomicInteger(count);
// 门闩 线程锁
this.latch = new CountDownLatch(count);
}
}
/**
* Task 包装类<br>
* 此类型的意义是记录可能会被 Executor 吃掉的异常<br>
*/
public static class TaskWrapper implements Runnable {
private final Runnable gift;
private final CountDownLatch latch;
private final AtomicInteger count;
public TaskWrapper(final Runnable target) {
this.gift = target;
this.count = null;
this.latch = null;
}
public TaskWrapper(final Runnable target, final AsyncWaitLock lock) {
if (lock == null) {
this.gift = null;
this.count = null;
this.latch = null;
return;
}
this.gift = target;
this.count = lock.getCount();
this.latch = lock.getLatch();
}
@Override
public void run() {
// 捕获异常，避免在 Executor 里面被吞掉了
if (gift != null) {
try {
gift.run();
// 标示已执行
count.decrementAndGet();
} catch (Exception e) {
String errMsg = StrUtil.format("线程池-包装的目标执行异常: {}", e.getMessage());
log.error(errMsg, e);
} finally {
latch.countDown();
}
}
}
}
}

104
opsli-base-support/opsli-common/src/main/java/org/opsli/common/thread/AsyncProcessor.java
Unescape View File

@ -1,11 +1,16 @@
package org.opsli.common.thread;
import cn.hutool.core.util.StrUtil;
import com.google.common.util.concurrent.*;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.opsli.common.thread.ThreadPoolFactory;
import java.util.concurrent.*;
import java.util.concurrent.Callable;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
/**
* 自定义线程执行器 - 等待线程执行完毕不拒绝
@ -16,20 +21,10 @@ import java.util.concurrent.*;
@Slf4j
public class AsyncProcessor {
/**
* 默认最大并发数<br>
*/
private static final int DEFAULT_MAX_CONCURRENT = Runtime.getRuntime().availableProcessors() * 2;
/**
* 线程池名称格式
*/
private static final String THREAD_POOL_NAME = "AsyncProcessPool-{}-%d";
/**
* 默认队列大小
*/
private static final int DEFAULT_SIZE = 1024;
private static final String THREAD_POOL_NAME = "AsyncProcessorWaitPool-{}-%d";
/**
* 默认线程池关闭等待时间 秒
@ -37,59 +32,51 @@ public class AsyncProcessor {
private static final int DEFAULT_WAIT_TIME = 10;
/**
* 默认线程存活时间
* 线程池监听执行器
*/
private static final long DEFAULT_KEEP_ALIVE = 60L;
/**
* 执行器
*/
private ExecutorService execute;
private ListeningExecutorService execute;
/**
* 初始化
* @param key 线程池标识
*/
public AsyncProcessor(String key){
public void init(String key){
if(StringUtils.isBlank(key)){
return;
}
// 线程工厂名称
String formatThreadPoolName = StrUtil.format(THREAD_POOL_NAME, key);
BasicThreadFactory basicThreadFactory = new BasicThreadFactory.Builder()
.namingPattern(formatThreadPoolName)
.daemon(true).build();
// 创建 Executor
// 此处默认最大值改为处理器数量的 4 倍
try {
// 执行队列
BlockingQueue<Runnable> executorQueue = new ArrayBlockingQueue<>(DEFAULT_SIZE);
execute = new ThreadPoolExecutor(DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_CONCURRENT * 4, DEFAULT_KEEP_ALIVE,
TimeUnit.SECONDS, executorQueue, basicThreadFactory);
// 监听执行器
execute = MoreExecutors.listeningDecorator(
ThreadPoolFactory.createDefThreadPool(formatThreadPoolName));
// 这里不会自动关闭线程， 当线程超过阈值时 抛异常
// 关闭事件的挂钩
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
log.info("AsyncProcessorWait 异步处理器关闭");
log.info("ProcessorWait 异步处理器关闭");
execute.shutdown();
try {
// 等待1秒执行关闭
if (!execute.awaitTermination(DEFAULT_WAIT_TIME, TimeUnit.SECONDS)) {
log.error("AsyncProcessorWait 由于等待超时，异步处理器立即关闭");
log.error("ProcessorWait 由于等待超时，异步处理器立即关闭");
execute.shutdownNow();
}
} catch (InterruptedException e) {
log.error("AsyncProcessorWait 异步处理器关闭中断");
log.error("ProcessorWait 异步处理器关闭中断");
execute.shutdownNow();
}
log.info("AsyncProcessorWait 异步处理器关闭完成");
log.info("ProcessorWait 异步处理器关闭完成");
}));
} catch (Exception e) {
log.error("AsyncProcessorWait 异步处理器初始化错误", e);
log.error("ProcessorWait 异步处理器初始化错误", e);
throw new ExceptionInInitializerError(e);
}
}
@ -99,8 +86,8 @@ public class AsyncProcessor {
* 执行任务，不管是否成功<br>
* 其实也就是包装以后的 {@link } 方法
*
* @param task
* @return
* @param task 任务
* @return boolean
*/
public boolean executeTask(Runnable task) {
try {
@ -117,14 +104,43 @@ public class AsyncProcessor {
* 当提交失败时，会抛出 {@link }
*
* @param task 任务
* @return Future<T>
*/
public <T> Future<T> submitTask(Callable<T> task) {
try {
return execute.submit(task);
} catch (RejectedExecutionException e) {
log.error("AsyncProcessorWait 执行任务被拒绝", e);
throw new UnsupportedOperationException("AsyncProcessorWait 无法提交任务，已被拒绝", e);
}
public <T> void executeTaskAndCallback(Callable<T> task, Function<CallbackResult<T>, Void> callback) {
ListenableFuture<T> future = execute.submit(task);
Futures.addCallback(future, new FutureCallback<T>() {
@Override
public void onSuccess(T result) {
CallbackResult<T> callbackResult = new CallbackResult<>();
callbackResult.setSuccess(true);
callbackResult.setResult(result);
// 线程池失败后 返回该 Runnable
callback.apply(callbackResult);
}
@Override
public void onFailure(Throwable t) {
log.error("线程名称：{} - 执行异常信息：{}", Thread.currentThread().getName(), t.getMessage());
CallbackResult<T> callbackResult = new CallbackResult<>();
callbackResult.setSuccess(false);
callback.apply(callbackResult);
}
}, execute);
}
// =================
/**
* 回调结果
* @param <T>
*/
@Data
public static class CallbackResult<T>{
/** 状态 */
private Boolean success;
/** 结果 */
private T result;
}
}

101
opsli-base-support/opsli-common/src/main/java/org/opsli/common/thread/ThreadPoolFactory.java
Unescape View File

@ -0,0 +1,101 @@
package org.opsli.common.thread;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* 线程池工厂
*
* @author Parker
* @date 2021/11/2 10:48
*/
public final class ThreadPoolFactory {
/**
* 默认最大并发数<br>
*/
private static final int DEFAULT_MAX_CONCURRENT = Runtime.getRuntime().availableProcessors() * 2;
/**
* 默认线程存活时间
*/
private static final long DEFAULT_KEEP_ALIVE = 60L;
/**
* 默认队列大小
*/
private static final int DEFAULT_SIZE = 1024;
/**
* 线程池名称格式
*/
private static final String DEFAULT_THREAD_POOL_NAME = "ProcessPool-{}-%d";
/**
* 创建默认的线程池
* @return ThreadPoolExecutor
*/
public static ThreadPoolExecutor createDefThreadPool(){
return createInitThreadPool(DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_CONCURRENT * 4, DEFAULT_KEEP_ALIVE,
TimeUnit.SECONDS, DEFAULT_SIZE, DEFAULT_THREAD_POOL_NAME, new ThreadPoolExecutor.CallerRunsPolicy());
}
/**
* 创建默认的线程池
*
* @param poolName 线程池名称
* @return ThreadPoolExecutor
*/
public static ThreadPoolExecutor createDefThreadPool(String poolName){
return createInitThreadPool(DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_CONCURRENT * 4, DEFAULT_KEEP_ALIVE,
TimeUnit.SECONDS, DEFAULT_SIZE, poolName, new ThreadPoolExecutor.CallerRunsPolicy());
}
/**
* 创建默认的线程池
*
* @param maxConcurrent 最大线程数
* @param poolName 线程池名称
* @return ThreadPoolExecutor
*/
public static ThreadPoolExecutor createDefThreadPool(int maxConcurrent, String poolName){
return createInitThreadPool(maxConcurrent, maxConcurrent * 4, DEFAULT_KEEP_ALIVE,
TimeUnit.SECONDS, DEFAULT_SIZE, poolName, new ThreadPoolExecutor.CallerRunsPolicy());
}
/**
* 创建线程池
* @param coreConcurrent 核心线程数
* @param maxConcurrent 最大线程数
* @param keepAlive 线程存活时效
* @param timeUnit 线程存活单位
* @param queueSize 队列大小
* @param poolName 线程池名称
* @param handler 拒绝处理策略
* @return ThreadPoolExecutor
*/
public static ThreadPoolExecutor createInitThreadPool(final int coreConcurrent,
final int maxConcurrent,
final long keepAlive,
final TimeUnit timeUnit,
final int queueSize,
final String poolName,
final RejectedExecutionHandler handler
){
return new ThreadPoolExecutor(coreConcurrent, maxConcurrent, keepAlive, timeUnit,
new LinkedBlockingDeque<>(queueSize),
new ThreadFactoryBuilder().setNameFormat(poolName).build(),
handler
);
}
private ThreadPoolFactory(){}
}

44
opsli-base-support/opsli-core/src/main/java/org/opsli/core/api/ApiRequestMappingHandlerMapping.java
Unescape View File

@ -0,0 +1,44 @@
package org.opsli.core.api;
import org.opsli.common.annotation.ApiVersion;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.util.Assert;
import org.springframework.web.servlet.mvc.condition.RequestCondition;
import org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerMapping;
import java.lang.reflect.Method;
import java.util.Objects;
/**
* 重写RequestMappingHandlerMapping，自定义匹配的处理器
*
* @author Parker
* @date 2021年10月27日12:40:45
*/
public class ApiRequestMappingHandlerMapping extends RequestMappingHandlerMapping {
@Override
protected RequestCondition<?> getCustomTypeCondition(Class<?> handlerType) {
// 扫描类上的 @ApiVersion
ApiVersion apiVersion = AnnotationUtils.findAnnotation(handlerType, ApiVersion.class);
return createRequestCondition(apiVersion);
}
@Override
protected RequestCondition<?> getCustomMethodCondition(Method method) {
// 扫描方法上的 @ApiVersion
ApiVersion apiVersion = AnnotationUtils.findAnnotation(method, ApiVersion.class);
return createRequestCondition(apiVersion);
}
private RequestCondition<ApiVersionCondition> createRequestCondition(ApiVersion apiVersion) {
if (Objects.isNull(apiVersion)) {
return null;
}
int value = apiVersion.value();
Assert.isTrue(value >= 1, "Api Version Must be greater than or equal to 1");
return new ApiVersionCondition(value);
}
}

63
opsli-base-support/opsli-core/src/main/java/org/opsli/core/api/ApiVersionCondition.java
Unescape View File

@ -0,0 +1,63 @@
package org.opsli.core.api;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.servlet.mvc.condition.RequestCondition;
import javax.servlet.http.HttpServletRequest;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
/**
* API 版本条件
* 重写
*
* @author Parker
* @date 2021年10月27日12:39:09
*/
@Data
@Slf4j
public class ApiVersionCondition implements RequestCondition<ApiVersionCondition> {
/**
* 接口路径中的版本号前缀，如: api/v[1-n]/fun
*/
private final static Pattern VERSION_PREFIX = Pattern.compile("/v(\\d+)/");
private int apiVersion;
ApiVersionCondition(int apiVersion) {
this.apiVersion = apiVersion;
}
/**
* 最近优先原则，方法定义的 @ApiVersion > 类定义的 @ApiVersion
*/
@Override
public ApiVersionCondition combine(ApiVersionCondition other) {
return new ApiVersionCondition(other.getApiVersion());
}
/**
* 获得符合匹配条件的ApiVersionCondition
*/
@Override
public ApiVersionCondition getMatchingCondition(HttpServletRequest request) {
Matcher m = VERSION_PREFIX.matcher(request.getRequestURI());
if (m.find()) {
int version = Integer.parseInt(m.group(1));
if (version >= getApiVersion()) {
return this;
}
}
return null;
}
/**
* 当出现多个符合匹配条件的ApiVersionCondition，优先匹配版本号较大的
*/
@Override
public int compareTo(ApiVersionCondition other, HttpServletRequest request) {
return other.getApiVersion() - getApiVersion();
}
}

14
opsli-base-support/opsli-core/src/main/java/org/opsli/core/autoconfigure/conf/SpringWebMvcConfig.java
Unescape View File

@ -17,7 +17,9 @@ package org.opsli.core.autoconfigure.conf;
import lombok.extern.slf4j.Slf4j;
import org.opsli.common.annotation.ApiRestController;
import org.opsli.core.api.ApiRequestMappingHandlerMapping;
import org.opsli.core.autoconfigure.properties.ApiPathProperties;
import org.springframework.boot.autoconfigure.web.servlet.WebMvcRegistrations;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.cors.CorsConfiguration;
@ -25,6 +27,7 @@ import org.springframework.web.cors.UrlBasedCorsConfigurationSource;
import org.springframework.web.filter.CorsFilter;
import org.springframework.web.servlet.config.annotation.PathMatchConfigurer;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
import org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerMapping;
import javax.annotation.Resource;
@ -36,11 +39,20 @@ import javax.annotation.Resource;
*/
@Slf4j
@Configuration
public class SpringWebMvcConfig implements WebMvcConfigurer {
public class SpringWebMvcConfig implements WebMvcConfigurer, WebMvcRegistrations {
@Resource
private ApiPathProperties apiPathProperties;
/**
* 重写RequestMappingHandlerMapping，自定义匹配的处理器
* @return RequestMappingHandlerMapping
*/
@Override
public RequestMappingHandlerMapping getRequestMappingHandlerMapping() {
return new ApiRequestMappingHandlerMapping();
}
/**
* 配置 ApiRestController 生效
* @param configurer 配置

Write Preview
Loading…
Cancel
Save