基于SpringBoot的注解式请求合并–完善篇1
基于上一篇基于SpringBoot的注解式请求合并的局限性太大。虽然说是暂时要达到的效果,但是不可能满足于此。
修改后的请求合并注解
就添加了一个参数
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target(value = ElementType.METHOD)
public @interface PullRequests {
/**
* 批量请求时间
* @return
*/
long timeout() default 50L; //50毫秒
/**
* 该注解的自定义id,用于标识该id请求的
* @return
*/
String id();
/**
* 批量方法的名称,此类下的方法
* @return
*/
String batchMethod();
/**
* 比较不同数据正确返回给请求的调用的方法
* @return
*/
String compareMethod();
}
对应的实体类
不变,还是上篇的
自定义每个request封装的参数对象
@Data
@AllArgsConstructor
@ToString
public class MyRequest {
/**
* 该对象id
*/
private Object token;
/**
* 参数
*/
private Object param;
/**
* 批量处理的方法名
*/
private String batchMethod;
/**
* 比较不同数据正确返回给请求的调用的方法
*/
private String compareMethod;
private ProceedingJoinPoint jpt;
private CompletableFuture<Object> future;
}
aop环绕增强
/**
* @program: IoT-Plat
* @description: 合并请求的自定义注解 学习中
* @author: Mr.Liu
* @create: 2020-05-14 22:08
**/
@Aspect
@Slf4j
@NoArgsConstructor
@Component
public class PullRequestsAop implements ApplicationContextAware {
/**这个用来获取那个实体类,**/
private ApplicationContext context;
/**这个我原本想通过获取注解使用的地方的数目,但是好像行不通,就先写个死的**/
public static int size = 10;
/**分批次处理的数量**/
public static final int BATCH_SIZE = 100;
/**用来保存某个方法上对应来的请求队列**/
public static final ConcurrentHashMap<String,ConcurrentLinkedDeque<MyRequest>> allRequest = new ConcurrentHashMap<>();
/**初始化定时任务**/
public static final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(size);
/**用来做某一个方法是否是第一次请求的标识**/
public static final ConcurrentHashMap<String, Boolean> tag = new ConcurrentHashMap<>();
/**
* @param joinPoint
* @param pullRequests
* @param id
* @return
*/
@Around(value = "@annotation(pullRequests) && args(id)")
public Object aroundMethod(ProceedingJoinPoint joinPoint, PullRequests pullRequests, Object id){
/**合并请求 同类方法的id**/
String did = pullRequests.id();
/**当前请求id , 随机生成**/
String pid = UUID.randomUUID().toString();
/**批量请求的方法名称**/
String batchMethod = pullRequests.batchMethod();
/**批量请求等待的时间**/
long time = pullRequests.timeout();
String compareMethod = pullRequests.compareMethod();
CompletableFuture<Object> future = new CompletableFuture<>();
MyRequest myRequest = new MyRequest(pid,id,did,compareMethod,joinPoint,future);
ConcurrentLinkedDeque<MyRequest> requests = allRequest.get(did);
/**这里判断是否是第一次添加 did 对应的请求队列**/
if(requests == null){
synchronized (LinkedBlockingDeque.class){
requests = new ConcurrentLinkedDeque<>();
requests.add(myRequest);
allRequest.put(did,requests);
}
}else {
requests.add(myRequest);
}
/**这里判断是否是第一次加入定时任务**/
if(Objects.isNull(tag.get(did)) ||!tag.get(did)){
tag.put(did,true);
scheduledExecutorService.scheduleWithFixedDelay(()->{
if(ToolUtil.isEmpty(allRequest.get(did))||allRequest.get(did).size()<=0){
return;
}
/**获取被切的那个方法**/
Method method = PullRequestsMethodUtils.getCompensableMethod(joinPoint);
if (method == null) {
throw new RuntimeException(String.format("join point not found method, point is : %s", joinPoint.getSignature().getName()));
}
Class[] clazz = method.getParameterTypes();
Class returnType = method.getReturnType();
if(clazz.length!=1){
throw new RuntimeException(String.format("param only one and must one: %s", joinPoint.getSignature().getName()));
}
String s = clazz[0].getName();
/**TODO 下面的这个好像还可以优化的**/
Set longs;
if("java.lang.Long".equals(s)){
longs = new HashSet<Long>();
}else if("java.lang.String".equals(s)){
longs = new HashSet<String>();
}else if("java.lang.Integer".equals(s)){
longs = new HashSet<Integer>();
}else{
longs = new HashSet();
//throw new RuntimeException(String.format("Unsuport this param type: %s", joinPoint.getSignature().getName()));
}
/**获取当前被切的那个类**/
Class targetClass = ReflectionUtils.getDeclaringType(joinPoint.getTarget().getClass(), method.getName(), method.getParameterTypes());
try {
/**获取批量执行的那个方法**/
Method mm = targetClass.getMethod(batchMethod, List.class);
/**这个是因为采用了mybatis,就选择从spring容器中获取**/
Object target = context.getBean(targetClass);
ConcurrentLinkedDeque<MyRequest> deque = allRequest.get(did);
List<MyRequest> requestList = new ArrayList<>();
for (int i = 0; i < BATCH_SIZE; i++) {
MyRequest os = deque.pollFirst();
if(os==null){
break;
}
Object o = os.getParam();
longs.add(clazz[0].cast(o));
requestList.add(os);
}
log.info("当前批量处理了"+requestList.size()+"个线程请求");
/**执行批量请求的方法**/
ArrayList<Object> list = (ArrayList<Object>) mm.invoke(target,new ArrayList<>(longs));
AtomicBoolean tag = new AtomicBoolean(false);
requestList.forEach(request->{
for (Object o:list) {
try {
Method m1 = returnType.getDeclaredMethod(request.getCompareMethod());
/**这里根据条件判断应该返回的数据**/
if((clazz[0].cast(request.getParam())).equals(m1.invoke(returnType.cast(o)))){
request.getFuture().complete(o);
tag.set(true);
break;
}
} catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
e.printStackTrace();
request.getFuture().complete(null);
tag.set(true);
break;
}
}
if(!tag.get()){
try {
request.getFuture().complete(returnType.newInstance());
} catch (InstantiationException | IllegalAccessException e) {
e.printStackTrace();
request.getFuture().complete(null);
}
}
});
} catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
e.printStackTrace();
}catch (Throwable throwable){
throwable.printStackTrace();
}
/**下面的延时执行我设置为time,但是这里为了演示效果,就将这个时间调大了**/
},0,time*2, TimeUnit.MILLISECONDS);
}
try {
/**下面的延时执行我设置为time*100,但是这里为了演示效果,就将这个时间调大了**/
/**阻塞获取**/
return myRequest.getFuture().get(time * 100,TimeUnit.MILLISECONDS);
}catch (TimeoutException e){
return null;
}catch (Throwable throwable) {
throwable.printStackTrace();
return null;
}
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.context=applicationContext;
}
}
Service实现类
@Service
public class EnvironlServiceImpl extends ServiceImpl<EnvironlMapper, Environl> implements IEnvironlService {
@Override
public boolean insert(Environl environl) {
return this.save(environl);
}
/**
* 主要是下面这个注解和对应的批处理请求的方法
* getId是对应实体类中获取那个id的get方法,主要是用来对获取到的数据与传入的参数id比对,然后进行对应的请求返回对应的数据
*/
@Override
@PullRequests(id="idids",batchMethod = "gs",compareMethod = "getId")
public Environl getByIds(Long id) {
return new Environl();
}
@Override
public List<Environl> gs(List<Long> l){
return this.baseMapper.selectBatchIds(l);
}
}
控制层
@RestController
@RequestMapping("/test")
public class Test {
@Autowired
private IEnvironlService environlService;
@GetMapping("/{id}")
public ResponseData getids(@PathVariable("id") Long id){
Object o = environlService.getByIds(id);
if(o==null){
return SuccessResponseData.error("请求失败");
}
return SuccessResponseData.success(o);
}
}
效果演示
利用Jmeter进行测试,设置了四个不同参数的get请求
发起十个线程同时请求,得到效果如图:
可见,得到了该有的结果,虽然还是有比较大的局限性,但是相比上一篇,利用反射,不必局限于参数与返回值写死。也可以对从数据库获取数据进行优化,改为从redis这种获取或者其它,还有很大的进步优化空间的,慢慢学习互相进步啦。