java将请求写入队列，再用异步消费该队列，高并发实战。_异步线程消费队列

一、背景

   在开发项目过程中，遇到一个问题就是，一个系统与上百个系统对接，有大量的请求和有可能数据是重复发送，但是在数据必须保证数据不能重复消费。

二、设计思路

  刚才开始设计思路，按照往常写代码，编写controller，service，dao层，一气呵成。但是在使用过程出现，大量的请求导致服务器不可以，直接宕机。出现这个问题，那没有办法得优化。

设计思路：

   1、将所有的请求，放到队列里面。

   2、异步消费队列里面的数据。

三、代码实现

1、controller层代码  

@RequestMapping("/api")
@RestController
@Api(value = "账号监控平台系统相关接口", tags = {"账号监控平台系统相关接口"})
@Slf4j
public class UserDepartmentController {
 
    private static LinkedBlockingDeque<List<UserLogin>> deque = new LinkedBlockingDeque<>();
    @Autowired
    private UserLoginManagementService userLoginManagementService;
 
    @Autowired
    private UserLoginService userLoginService;
    @Autowired
    private Sid sid;
 
 
    @ApiOperation(value = "推送用户登录日志接口",notes = "推送用户登录日志接口", httpMethod = "POST")
    @PostMapping("/userLog/add")
    public R addUserLogQueue(@Validated @RequestBody List< @Valid UserLogin> list)  {
        if (list == null || list.isEmpty()) {
            return R.errorMsg("传数据不能为空");
        }
        try {
            List<DataMsg>   resultList =  new ArrayList<>();  //返回给调用接口集合
            List<UserLogin>   saveList =  new ArrayList<>();  // 保存数据的list
            // 判断参数不能为空，
            for(UserLogin userLogin : list){
                R   result = ParameterUtil.userParameter(userLogin);
                if(result.getCode().equals(RStatus.fail.getCode())){
                    DataMsg  dataMsg = new DataMsg();
                    BeanUtils.copyProperties(userLogin,dataMsg);
                    dataMsg.setMessage(result.getMsg());
                    log.error("【UserDepartmentController -- addUserLog】传入的登录信息数据有空，入参的参数数据：{},消息的提醒：{}",userLogin.toString(),result.getMsg());
                    resultList.add(dataMsg);
                }else {
                    userLogin.setUserId(userLogin.getSystemName()+"-"+userLogin.getUserId());
                    saveList.add(userLogin);
                }
            }
            if(saveList != null && saveList.size() > 0 ){
              //  String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                //log.info("【UserDepartmentController -- addUserLog】传入的登录信息数据，入参的时间：{}，入参的数据:{} ",time, saveList.toString());
                if(deque.remainingCapacity() > 0){
                    deque.put(saveList);
                }else {
                    return R.errorMsg("队列已经满了,请稍后重试");
                }
                SaveUserLogQueueRunnable saveUserLogQueueRunnable = new SaveUserLogQueueRunnable(deque,userLoginService,userLoginManagementService,sid);
             //刚开始使用多线程，但是如果相同的数据发送多次请求，会导致数据会重复。后面修改上面的方式，单线程去处理。
                ThreadPoolExecutor cupInstance = ThreadPoolUtils.getCUPInstance();
                cupInstance.execute(saveUserLogQueueRunnable);
 
            }
            if(resultList != null && resultList.size() >0){
                return R.build(500,"失败",resultList);
            }else{
                return R.ok();
            }
        }catch (Exception e){
            e.printStackTrace();
            return R.errorMsg("请求接口异常");
        }
 
    }
}

@Slf4j
public class SaveUserLogQueueRunnable implements Runnable {
 
    private UserLoginService userLoginService;
    private UserLoginManagementService userLoginManagementService;
    private Sid sid;
    private  LinkedBlockingDeque<List<UserLogin>> deque;
 
    public SaveUserLogQueueRunnable(LinkedBlockingDeque<List<UserLogin>> deque, UserLoginService userLoginService, UserLoginManagementService userLoginManagementService, Sid sid){
        this.deque = deque;
        this.userLoginService = userLoginService;
        this.userLoginManagementService = userLoginManagementService;
        this.sid = sid;
    }
 
 
    @Override
    public   void run() {
        try {
            //开始处理请求队列中的请求,按照队列的FIFO的规则,先处理先放入到队列中的请求
            while (deque != null && deque.size() > 0){
                List<UserLogin> take = deque.take(); //取出队列中的请求
                addLog(take); //处理请求
            }
        }catch (Exception e){
            log.error("【SaveUserLogQueueRunnable   】处理队列出现错误如下：");
            e.printStackTrace();
        }
 
    }
 
    public synchronized  void   addLog( List<UserLogin> list)  {
        String time1 = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        System.out.println("list的大小： "+list.size()+"----时间为： "+ time1);
        for (UserLogin userLogin :list){
            // 这是自己写业务处理的代码。
        }
    }
}

public class Constants {
    // 消费者,单一线程， 进行处理业务逻辑
    public static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor();


}

四、测试

代码写完了用jmeter 工具测试，，20个并发。

请求参数：

测试结果：预期结果，数据库里面的数据和入参的数据一致。但是出现有重复的数据。有出现问题，还得解决。

解决办法：使用 newSingleThreadScheduledExecutor 单线程去执行，从队列里面拿一条数据消费一条数据

再次测试：还是 jmeter 工具测试，，20个并发。发现不会出现重复的数据。