基于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();}}

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("服务繁忙");}}@Overridepublic 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",parmes = {Long.class})public ResponseData getByIds(Long id) {return ResponseData.success();}@Overridepublic 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 {@Autowiredprivate IEnvironlService environlService;@GetMapping("/{id}")public ResponseData getids(@PathVariable("id") Long id){return environlService.getByIds(id);}
}

效果演示

直接发起两个不同的请求GET http://localhost:5051/test/15，在每一次对应的定时任务内发起请求即可。
可见是通过批量获取的方法获取的数据，并且对结果就行对应的分类返回。