基于SpringBoot的注解式请求合并
某一天突然想到之前写过一个请求合并的代码,但是代码嵌入太强,就想自己写一个请求合并的注解,但是好像网上的相关代码较少,就去找到了Hystrix的请求合并相关资料,但是这代码可不是我所能看懂的。真是太难懂了,抓绕。
以下注解写的不是很全面,不是一个万能的,只能针对某一些类型的请求合并,并且是在spring boot和mybatis-plus基础上写的,要改为其它的也比较简单。
之后会慢慢学习完善的。如果各位看官有啥好滴思路,尽情到评论区发表看法啦,小弟再此谢谢啦。
下一篇“完善篇“通过反射完善好了大部分的局限性。
请求合并注解
/**
* 请求合并请求的注解
* @author liuyanbin
*/
@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();
}
对应的实体类,采用的是mybatis-plus
package cn.ljobin.bibi.domain;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import lombok.ToString;
@TableName("tb_environment")
@Data
@ToString
public class Environl {
@TableId(value = "id",type = IdType.AUTO)
private Long id;
private String imei;
private Float ambient_temp;
private Float ambient_hum;
private Float ambient_light;
private Float soil_temp;
private Float soil_hum;
private Float atmo_pressure;
private Float wind_speed;
private String wind_direction;
private String snow_rain;
private Float pm25;
private Float co2;
private Float co;
private Float so2;
private String colltime;
@TableField(exist = false)
private String endtime;
@TableField(exist = false)
private String status;
@TableField(exist = false)
private String uid;
@TableField(exist = false)
private String username;
private String deptid;
public String getEndtime() {
return endtime;
}
public void setEndtime(String endtime) {
this.endtime = endtime;
}
public Environl() {
super();
}
}
自定义每个request封装的参数对象
@Data
@AllArgsConstructor
@ToString
public class MyRequest {
/**
* 该对象id
*/
private Object token;
/**
* 参数
*/
private Object param;
/**
* 批量处理的方法名
*/
private String batchMethod;
private ProceedingJoinPoint jpt;
private CompletableFuture<Object> future;
}
aop环绕增强
package cn.ljobin.bibi.aop;
import cn.ljobin.bibi.annotion.PullRequests;
import cn.ljobin.bibi.domain.Environl;
import cn.ljobin.bibi.utils.FactoryBuilder;
import cn.ljobin.bibi.utils.PullRequestsMethodUtils;
import cn.ljobin.bibi.utils.ReflectionUtils;
import cn.stylefeng.roses.core.reqres.response.SuccessResponseData;
import cn.stylefeng.roses.core.util.ToolUtil;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.util.*;
import java.util.concurrent.*;
/**
* @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 TODO 这个是使用该注解的方法上的参数,目前只能针对单个参数,且是long的
* @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();
//Class[] parmes = pullRequests.parmes();
CompletableFuture<Object> future = new CompletableFuture<>();
MyRequest myRequest = new MyRequest(pid,id,did,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()));
}
//Type returnType = method.getReturnType();
/**获取当前被切的那个类**/
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<>();
Set<Long> longs = new HashSet<>();
for (int i = 0; i < BATCH_SIZE; i++) {
MyRequest os = deque.pollFirst();
if(os==null){
break;
}
longs.add((Long)os.getParam());
requestList.add(os);
}
log.info("当前批量处理了"+requestList.size()+"个线程请求");
/**执行批量请求的方法**/
ArrayList<Object> list = (ArrayList<Object>) mm.invoke(target,new ArrayList<>(longs));
requestList.forEach(request->{
for (Object o:list) {
/**TODO 这里是写死了返回的类型,不知道怎么动态来改**/
System.err.println(((Environl) o).toString());
/**这里根据条件判断应该返回的数据,这里只是返回一个,也可以返回集合,就改一点代码即可**/
if(((Long) request.getParam()).equals(((Environl) o).getId())){
request.getFuture().complete(o);
break;
}
}
request.getFuture().complete(new Environl());
});
} catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
e.printStackTrace();
}catch (Throwable throwable){
throwable.printStackTrace();
}
/**下面的延时执行我设置为time,但是这里为了演示效果,就将这个时间调大了**/
},0,time*500, TimeUnit.MILLISECONDS);
}
try {
/**下面的延时执行我设置为time*100,但是这里为了演示效果,就将这个时间调大了**/
/**阻塞获取**/
Object map = myRequest.getFuture().get(time * 1000,TimeUnit.MILLISECONDS);
return SuccessResponseData.success(map);
}catch (TimeoutException e){
return SuccessResponseData.error("请求处理超时");
}catch (Throwable throwable) {
throwable.printStackTrace();
return SuccessResponseData.error("服务繁忙");
}
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.context=applicationContext;
}
}
Service类
package cn.ljobin.bibi.service.Impl;
import cn.ljobin.bibi.annotion.PullRequests;
import cn.ljobin.bibi.domain.Environl;
import cn.ljobin.bibi.mapper.EnvironlMapper;
import cn.ljobin.bibi.service.IEnvironlService;
import cn.stylefeng.roses.core.reqres.response.ResponseData;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.io.Serializable;
import java.util.List;
@Service
public class EnvironlServiceImpl extends ServiceImpl<EnvironlMapper, Environl> implements IEnvironlService {
/**
* 主要是下面这个注解和对应的批处理请求的方法
*/
@Override
@PullRequests(id="idids",batchMethod = "gs")
public ResponseData getByIds(Long id) {
return ResponseData.success();
}
@Override
public List<Environl> gs(List<Long> l){
return this.baseMapper.selectBatchIds(l);
}
}
控制层
/**
* @program: IoT-Plat
* @description:
* @author: Mr.Liu
* @create: 2020-05-19 23:38
**/
@RestController
@RequestMapping("/test")
public class Test {
@Autowired
private IEnvironlService environlService;
@GetMapping("/{id}")
public ResponseData getids(@PathVariable("id") Long id){
return environlService.getByIds(id);
}
}
效果演示
直接发起两个不同的请求GET http://localhost:5051/test/15,在每一次对应的定时任务内发起请求即可。
可见是通过批量获取的方法获取的数据,并且对结果就行对应的分类返回。