基于Redis和Redisson的分布式全局锁，实现并发请求分组排队处理（组内排队，不同组并行）

ゞ浴缸里的玫瑰 2023-03-03 12:28 3阅读 0赞

@Service
    public class ReceiptsBatchServiceImpl implements ReceiptsBatchService
    {
            //private static final String REDISSON_RECEIPTS_BATCH_LOCK = "RECEIPTS_BATCH_LOCK_";	//批次处理全局锁
    	private static final String REDISSON_RECEIPTS_BATCH_NO_LOCK = "RECEIPTS_BATCH_NO_LOCK_";	//获取批次序号全局锁
    	private static final String REDIS_RECEIPTS_BATCH_NO_MAP = "RECEIPTS_BATCH_NO_MAP";//redis Map，保存各组线程的最新序号。key：分组关键字，value：组内最新分配的线程序号
    	private static final String REDIS_RECEIPTS_BATCH_STATUS_MAP = "RECEIPTS_BATCH_STATUS_MAP";//redis Map,保存线程处理状态。key：分组关键字_线程序号，value：状态值(Y/N)
    	private static final String BATCH_STATUS_Y = "Y";	//批次处理完成
    	private static final String BATCH_STATUS_N = "N";	//批次处理中
    	
    	
    	@Autowired
    	private RedissonClient redissonClient;
    	@Autowired
    	private RedisTemplate<String, Object> redisTemplate;
    	
    	//初始化全局锁，并给当前线程分配一个组内唯一序号
    	private int initializeBatchNoAndStatus(String projectId)
    	{
    		RLock globalLock = this.redissonClient.getLock(REDISSON_RECEIPTS_BATCH_NO_LOCK + projectId);
    		while(true)
    		{
    			if(!(globalLock.tryLock()))
    			{
    				waitForSomeTime(2000L);
    				continue;
    			}
    			break;
    		}
    		
    		int currentBatchNo = 0;
    		try
    		{
    			Object lastBatchNo = this.redisTemplate.opsForHash().get(REDIS_RECEIPTS_BATCH_NO_MAP, projectId);
    			if(lastBatchNo == null)
    			{
    				currentBatchNo = 1;
    			}
    			else
    			{
    				currentBatchNo = Integer.valueOf(lastBatchNo.toString()) + 1;
    			}
    			
    			this.redisTemplate.opsForHash().put(REDIS_RECEIPTS_BATCH_NO_MAP, projectId, currentBatchNo);
    			this.redisTemplate.opsForHash().put(REDIS_RECEIPTS_BATCH_STATUS_MAP, projectId + "_" + currentBatchNo, BATCH_STATUS_N);
    		}
    		finally
    		{
    			globalLock.unlock();
    		}
    		
    		return currentBatchNo;
    	}
    	
    	//检查当前线程的前一个线程的处理状态是否已完成
    	private boolean isPreviousBatchFinish(String projectId, int currentBatchNo)
    	{
    		String previousBatchStatusKey = projectId + "_" + (currentBatchNo - 1);
    		Object previousBatchStatus = this.redisTemplate.opsForHash().get(REDIS_RECEIPTS_BATCH_STATUS_MAP, previousBatchStatusKey);
    		return (previousBatchStatus == null) || (previousBatchStatus != null && BATCH_STATUS_Y.equals(previousBatchStatus.toString()));
    	}
    	
    	//当前线程休眠等待一段时间
    	private void waitForSomeTime(long millis)
    	{
    		try
    		{
    			Thread.currentThread().sleep(millis);
    		}
    		catch(InterruptedException e)
    		{
    			e.printStackTrace();
    		}
    	}
    	
    	//完成业务处理，更新线程处理状态
    	private void finishAndUpdateStatus(String projectId, int batchNo)
    	{
    		this.redisTemplate.opsForHash().put(REDIS_RECEIPTS_BATCH_STATUS_MAP, projectId + "_" + batchNo, BATCH_STATUS_Y);
    	}
    	
    	/**
    	 * 业务方法
    	 */
    	@Override
    	public BaseResponse<Object> submitReceiptsBatch(ReceiptsBatchDto receiptsBatchDto, HttpServletRequest request)
    	{
    		//保存批次
    		this.saveBatchInfo(receiptsBatchDto);
    		
    		int currentBatchNo = initializeBatchNoAndStatus(receiptsBatchDto.getProjectId());
    		//RLock globalLock = this.redissonClient.getLock(REDISSON_RECEIPTS_BATCH_LOCK + receiptsBatchDto.getProjectId());
    		while(true)
    		{
    			/*if(!(globalLock.tryLock()))//获取全局锁失败，休眠等待4秒后重试
    			{
    				waitForSomeTime(4000L);
    				continue;
    			}*/
    			
    			//排序处理的关键：如果上一个业务批次还未完成，则放弃锁，休眠等待一段时间后重试
    			if(!isPreviousBatchFinish(receiptsBatchDto.getProjectId(), currentBatchNo))
    			{
    				//globalLock.unlock();
    				waitForSomeTime(60000L);
    				continue;
    			}
    			
    			break;
    		}
    		
    		//业务处理开始。。。
    		List<TransactionDetailSaveVo> successDetailList = new ArrayList<TransactionDetailSaveVo>();	//成功明细
    		try
    		{
    			//按客户分组
    			Map<ReceiptsBatchDto, List<TransactionDetailSaveVo>> groupMap = receiptsBatchDto.getTransactionDetail().stream().collect(Collectors.groupingBy(detail -> new ReceiptsBatchDto(detail.getClientId(), detail.getClientName(), receiptsBatchDto.getBatchId())));
    			groupMap.forEach((key, value) -> {
    				key.setProjectId(receiptsBatchDto.getProjectId());
    				key.setAccountPeriod(receiptsBatchDto.getAccountPeriod());
    				key.setCustomerNames(value.get(0).getClientName());
    				key.setPaymentWay(receiptsBatchDto.getPaymentWay());
    				key.setPaymentWayName(receiptsBatchDto.getPaymentWayName());
    				key.setFillDate(receiptsBatchDto.getFillDate());
    				key.setTransactionDetail(value);
    				
    				EntityUtil.setSaveInfo(key);
    				this.oneCustomerReceipts(key, request, successDetailList);  //每个客户分别收款
    			});
    		}
    		catch(Exception e)
    		{
    			e.printStackTrace();
    			logger.error(e.getMessage());
    		}
    		finally
    		{
    			//更新当前批次业务处理状态
    			finishAndUpdateStatus(receiptsBatchDto.getProjectId(), currentBatchNo);
    			//globalLock.unlock();  //释放锁
    		}
    		//业务处理结束。。。
    	}
    }