Java synchronized实现线程分组排队执行（组内同步排队，不同组并发）

偏执的太偏执、 2023-03-02 10:51 27阅读 0赞

/**
    *按关键字分组线程，同一组内的线程顺序同步执行
    */
    @Service
    public class ReceiptsBatchServiceImpl implements ReceiptsBatchService
    {
    	//对象锁Map，key：分组关键字，value：锁对象
    	private static Map<String, String> PROJECT_ID_MAP = new HashMap<String, String>();
    	//各组线程的最新序号Map，key：分组关键字，value：组内最新分配的线程序号
    	private static Map<String, Integer> PROJECT_NO_MAP = new HashMap<String, Integer>();
    	//线程处理状态Map，key：分组关键字_线程序号，value：状态值(Y/N)
    	private static Map<String, String> PROJECT_STATUS_MAP = new HashMap<String, String>();
    
    	//初始化全局锁等Map，并给当前线程分配一个组内线程序号
    	private synchronized static int initializeLockAndGetThreadNo(String projectId)
    	{
    		int currentThreadNo = 0;
    		if(PROJECT_NO_MAP.get(projectId) == null)
    		{
    			currentThreadNo = 1;
    		}
    		else
    		{
    			currentThreadNo = PROJECT_NO_MAP.get(projectId) + 1;
    		}
    		
    		PROJECT_NO_MAP.put(projectId, currentThreadNo);
    		
    		if(PROJECT_ID_MAP.get(projectId) == null)
    		{
    			PROJECT_ID_MAP.put(projectId, projectId);
    		}
    		
    		PROJECT_STATUS_MAP.put(projectId + "_" + currentThreadNo, "N");
    		return currentThreadNo;
    	}
    	
    	//根据分组关键字，获取组内锁对象
    	private static String getLockObject(String projectId)
    	{
    		return PROJECT_ID_MAP.get(projectId);
    	}
    	
    	//完成业务处理，更新线程处理状态，并notify组内其它wait中的对象
    	private static void finishAndNotify(String projectId, int threadNo)
    	{
    		PROJECT_STATUS_MAP.put(projectId + "_" + threadNo, "Y");
    		PROJECT_ID_MAP.get(projectId).notifyAll();
    	}
    	
    	//检查当前线程的前一个线程的处理状态是否已完成
    	private static boolean isPreviousThreadFinish(String projectId, int currentThreadNo)
    	{
    		String previousThreadNoKey = projectId + "_" + (currentThreadNo - 1);
    		return (PROJECT_STATUS_MAP.get(previousThreadNoKey) == null) || (PROJECT_STATUS_MAP.get(previousThreadNoKey) != null && "Y".equals(PROJECT_STATUS_MAP.get(previousThreadNoKey)));
    	}
    	
    	//等待组内对象锁
    	private static void waitForPreviousThreadFinish(String projectId)
    	{
    		try
    		{
    			PROJECT_ID_MAP.get(projectId).wait();
    		}
    		catch(InterruptedException e)
    		{
    			e.printStackTrace();
    		}
    	}
    
    	/**
    	 * 业务方法
    	 */
    	@Override
    	public BaseResponse<Object> submitReceiptsBatch(ReceiptsBatchDto receiptsBatchDto, HttpServletRequest request)
    	{
    		//获取当前线程在组内的序号
    		int currentThreadNo = initializeLockAndGetThreadNo(receiptsBatchDto.getProjectId());
    		
    		//保存批次
    		this.saveBatchInfo(receiptsBatchDto);
    		List<TransactionDetailSaveVo> successDetailList = new ArrayList<TransactionDetailSaveVo>();	//成功明细
    		synchronized(getLockObject(receiptsBatchDto.getProjectId()))	//获取组内锁对象
    		{
    			//循环检查组内前一个线程处理状态是否已完成
    			while(!isPreviousThreadFinish(receiptsBatchDto.getProjectId(), currentThreadNo))
    			{
    				//等待组内对象锁
    				waitForPreviousThreadFinish(receiptsBatchDto.getProjectId());
    			}
    			
    			//业务处理。。。。
    			//按客户分组
    			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);  //每个客户分别收款
    			});
    			
    			//完成业务处理，更新线程处理状态，并notify组内其它wait中的对象
    			finishAndNotify(receiptsBatchDto.getProjectId(), currentThreadNo);
    		}
    	}
    }