关键词搜索

源码搜索 ×
×

Java使用Map模拟事务控制-跨系统数据一致性处理

发布2016-08-15浏览1888次

详情内容

首先，在复杂的业务系统中，尤其是跨系统交互如何保证两边数据一致性是极其重要的。下面就来模拟一下使用Map与Mybatis实现的事务控制的实例：

1.业务场景描述

WMS-OMS库存同步，实现两边库存一致性。

2.WMS-OMS接口交互缺陷

业务处理逻辑代码：

  1. try{
  2.  
  3.      // ====需要事务的业务操作====
  4.  
  5.     //WMS操作1
  6.  
  7.     //WMS操作2
  8.  
  9.     //WMS/操作3
  10.  
  11.     //OMS操作1----异步线程处理
  12.  
  13.     //WMS操作4
  14.  
  15.     //OMS操作2----异步线程处理
  16.  
  17.    //WMS操作5
  18.  
  19. }catch(Exception e){
  20.  
  21.          // 事务回滚
  22.  
  23.          TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
  24.  
  25. }finally{
  26.  
  27.          // 其他处理
  28.  
  29. }

这里处理过程中，从WMS操作4开始就已经出现弊端了，万一后面操作异常WMS这边的数据都会回滚，而线程已经执行不能够回滚处理就存在两边数据不一致的情况。

3.使用Map管理两边库存差异数据

以新增库存记录到OMS为例：

  1. public Integer insert(String key,WmStock model) {
  2. 		String memberCode = EhcacheUtil.get("JZTD_OMS_MEMBERCODE").toString();
  3. 		// 九州通达OMS同步处理
  4. 		if(EhcacheUtil.get("OMS_API_ISENABLE").toString().equals("true")&&memberCode.equals(model.getWmstCustomerCode()+"")){
  5. 			int flag=wmStockMapper.insert(model);
  6. 			if(flag>0){
  7. 				Log.getLogger(getClass()).info("<<<<<<<<<<<<<<更新数据到九州通达OMS>>>>>>>>>>>>EVENT:insert(WmStock model):"+(flag>0?true:false));
  8. 				// 线程更新库存(increment)
  9. 				//this.updateStockToOMSBySkuIndentifyInfo(true, model);
  10. 				// ===放入待处理缓存对象===
  11. 				List<JZTDOmsProduct> stocks=StockMapHandler.get(key);
  12. 				if(stocks==null||stocks.size()==0){
  13. 					List<JZTDOmsProduct> stocks2=new ArrayList<JZTDOmsProduct>();
  14. 					JZTDOmsProduct product=getProductDifferenceNumberFromStock(true, model);
  15. 					stocks2.add(product);
  16. 					StockMapHandler.put(key, stocks2);
  17. 				}else{
  18. 					JZTDOmsProduct product=getProductDifferenceNumberFromStock(true, model);
  19. 					stocks.add(product);
  20. 					StockMapHandler.put(key, stocks);
  21. 				}
  22. 			}
  23. 			return flag;
  24. 		}
  25. 		// 一般处理 
  26. 		return wmStockMapper.insert(model);
  27. 	}
注：新增库存两边直接同步不需要计算差异。
 交互数据结构： 
  1. /**
  2. 	 * 封装差异库存数据
  3. 	 * 
  4. 	 * @MethodName: getProductDifferenceNumberFromStock 
  5. 	 * @Description: 
  6. 	 * @param key
  7. 	 * @param isIncrement
  8. 	 * @param wmStock
  9. 	 * @return
  10. 	 * @throws
  11. 	 */
  12. 	private JZTDOmsProduct getProductDifferenceNumberFromStock(boolean isIncrement,WmStock wmStock){
  13. 		int memberCode =  Integer.valueOf(EhcacheUtil.get("JZTD_OMS_MEMBERCODE").toString());
  14. 		CdWhItme item=CdWhItmeService.selectByPrimaryKeyOfTabel(memberCode, wmStock.getWmstSkuId()); 
  15. 		String productCode=item==null?"":item.getCdskItemCode();
  16. 		String productDate=wmStock.getWmstProductDate()==null?null:DateUtil.date2String(wmStock.getWmstProductDate(), DateUtil.PATTERN_STANDARD);
  17. 		CdCustomer customer=cdWhCustomerService.selectByCustomerName(wmStock.getWmstCustomer(), memberCode+"");
  18. 		String customerCode=customer==null?"":customer.getCdstCustomerCode();
  19. 		JZTDOmsProduct product=null;
  20. 		if(isIncrement){
  21. 			product=new JZTDOmsProduct(customerCode,productCode, wmStock.getWmstNowNumber(), wmStock.getWmstEnabledNumber(), wmStock.getWmstSkuUnit(), wmStock.getWmstSkuBatch(),productDate );
  22. 		}else{
  23. 			product=new JZTDOmsProduct(customerCode,productCode, wmStock.getWmstNowNumber()==0?0:-wmStock.getWmstNowNumber(), wmStock.getWmstEnabledNumber()==0?0:-wmStock.getWmstEnabledNumber(), wmStock.getWmstSkuUnit(), wmStock.getWmstSkuBatch(),productDate );
  24. 		}
  25. 		return product;
  26. 	}
  27. 	
  28. 	@Override
  29. 	public void executeStockToOMSBySkuIndentifyInfo(String key,String token) {
  30. 		List<JZTDOmsProduct> produts=StockMapHandler.get(key);
  31. 		if(produts==null||produts.size()==0){
  32. 			Log.getLogger(getClass()).info(">>>>>>>>开启线程处理库存同步到九州通达OMS:.................验证缓存库存同步数据失败:key="+key);
  33. 			return ;
  34. 		}
  35. 		Map<String,Object> map=new HashMap<String, Object>();
  36. 		map.put("Products", produts);
  37. 		String requestBody= Tools.toJson(map);
  38. 		// 线程处理库存同步
  39. 		Log.getLogger(getClass()).info(">>>>>>>>开启线程处理库存同步到九州通达OMS:.................线程处理中---DATA:"+requestBody);
  40. 		new UpdateProductStockThread(jztdapiService, ebInterfaceLogService,requestBody, token, StaticProperty.OMS_WMS_INTERFACE_UPDATEPRODUCTSTOCK,true).start();
  41. 	}

创建Handler：

  1. package com.wlyd.fmcgwms.util;
  2.  
  3. import java.util.List;
  4. import java.util.Map;
  5. import java.util.concurrent.ConcurrentHashMap;
  6.  
  7. import com.wlyd.fmcgwms.persistence.beans.api.JZTDOmsProduct;
  8. import com.wlyd.fmcgwms.service.basic.WmStockService;
  9.  
  10. /**
  11.  * 库存缓存对象处理类
  12.  * 
  13.  * @packge com.wlyd.fmcgwms.util.StockMapHandler
  14.  * @date 2016年8月15日 上午10:30:10
  15.  * @author pengjunlin
  16.  * @comment
  17.  * @update
  18.  */
  19. public class StockMapHandler{
  20.  
  21. 	/**
  22. 	 * WmStock库存差异对象存储
  23. 	 */
  24. 	public static Map<String, List<JZTDOmsProduct>> map = new ConcurrentHashMap<String, List<JZTDOmsProduct>>(
  25. 			5000);
  26.  
  27. 	/**
  28. 	 * 获取List对象
  29. 	 * 
  30. 	 * @MethodName: get
  31. 	 * @Description:
  32. 	 * @param key
  33. 	 * @return
  34. 	 * @throws
  35. 	 */
  36. 	public static List<JZTDOmsProduct> get(String key) {
  37. 		if (map.containsKey(key)) {
  38. 			return map.get(key);
  39. 		}
  40. 		return null;
  41. 	}
  42.  
  43. 	/**
  44. 	 * 存储List对象
  45. 	 * 
  46. 	 * @MethodName: put
  47. 	 * @Description:
  48. 	 * @param key
  49. 	 * @param produts
  50. 	 * @throws
  51. 	 */
  52. 	public static void put(String key, List<JZTDOmsProduct> produts) {
  53. 		map.put(key, produts);
  54. 	}
  55.  
  56. 	/**
  57. 	 * 移除List对象
  58. 	 * 
  59. 	 * @MethodName: put
  60. 	 * @Description:
  61. 	 * @param key
  62. 	 * @throws
  63. 	 */
  64. 	public static void remove(String key) {
  65. 		if (map.containsKey(key)) {
  66. 			map.remove(key);
  67. 		}
  68. 	}
  69. 	
  70. 	/**
  71. 	 * 清除Map中存储的所有对象
  72. 	 * 
  73. 	 * @MethodName: clear 
  74. 	 * @Description: 
  75. 	 * @throws
  76. 	 */
  77. 	public static void clear(){
  78. 		map.clear();
  79. 	}
  80. 	
  81.  
  82. 	public class Handler implements Runnable{
  83. 		
  84. 		private WmStockService wmStockService;
  85. 		
  86. 		private String key;
  87. 		
  88. 		public Handler( WmStockService wmStockService, String key){
  89. 			this.wmStockService=wmStockService;
  90. 			this.key=key;
  91. 		}
  92.  
  93. 		@Override
  94. 		public void run() {
  95. 			wmStockService.executeStockToOMSBySkuIndentifyInfo(key, SAASTokenManager.getToken());
  96. 		}
  97. 		
  98. 	}
  99.  
  100. }

测试类方法： 
  1. package fmcgwms;
  2.  
  3. import org.junit.Test;
  4. import org.junit.runner.RunWith;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.test.context.ContextConfiguration;
  7. import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  8. import com.wlyd.fmcgwms.service.basic.WmStockService;
  9. import com.wlyd.fmcgwms.util.StockMapHandler;
  10.  
  11. @RunWith(SpringJUnit4ClassRunner.class)
  12. @ContextConfiguration(locations = "classpath:applicationContext.xml")
  13. public class StockMapHandlerTest {
  14. 	
  15. 	@Autowired
  16. 	WmStockService wmStockService;
  17. 	
  18. 	@Test
  19. 	public void testHandler(){
  20. 		
  21. 	      String key="10000"+System.currentTimeMillis();
  22.         
  23.         StockMapHandler stockMapHandler=new StockMapHandler();
  24.         
  25.         new Thread(stockMapHandler.new Handler(jztdapiService,ebInterfaceLogService, key,false)).start();
  26.         
  27.         try {
  28.             Thread.sleep(8000);// 线程等待执行，避免未执行完现场spring容器已销毁
  29.         } catch (InterruptedException e) {
  30.             e.printStackTrace();
  31.         }
  32. 	}
  33.  
  34. }

4.Map控制的缓存数据调用

  1. //对应的多个业务事务方法处理
  2. public void doBusiness(String key){ 
  3.  //key="uniqueString";
  4.  try{
  5.  
  6.      // ====需要事务的业务操作====
  7.  
  8.     //WMS操作1
  9.  
  10.     //WMS操作2
  11.  
  12.     //WMS/操作3
  13.  
  14.     //OMS操作1----异步线程处理（不执行线程）
  15.  
  16.     //WMS操作4（For instance:insertStock----map set different stock）
  17.  
  18.     //OMS操作2----异步线程处理（不执行线程）
  19.  
  20.     //WMS操作5(For instance:updateStock----map set different stock)
  21.  
  22.  }catch(Exception e){
  23.       // 事务回滚
  24.       TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
  25.  
  26.  }finally{
  27.         // 统一执行异步线程同步库存
  28.        //handler 处理
  29.        StockMapHandler stockMapHandler=new StockMapHandler();
  30. 	stockMapHandler.new Handler(wmStockService, key);
  31.        // 移除该业务操作的缓存对象     
  32.        StockMapHandler.remove(key); 
  33.  }
  34. }

5.关于key是否会重复的问题

在StockMapHandler.java 中新增一个生成：企业编码+UUID的唯一标识

  1. /**
  2. 	 * 生成企业的UUID编码
  3. 	 * 
  4. 	 * @MethodName: generateUuidKey 
  5. 	 * @Description: 
  6. 	 * @param esCorCode
  7. 	 * @return
  8. 	 * @throws
  9. 	 */
  10. 	public static String generateUuidKey(String esCorCode){
  11. 		String uuid=UUID.randomUUID().toString();
  12. 		uuid=uuid.replaceAll("-", "");
  13. 		uuid=esCorCode+uuid;
  14. 		return uuid;
  15. 	}

 另外需要注意，以下操作应该放在线程内执行，避免出错： 

  1. // 移除该业务操作的缓存对象     
  2. StockMapHandler.remove(key);

6.测试UUID的唯一性

原理：入库key不重复，那么map输出的长度就是循环的长度。

  1. package ebwms;
  2.  
  3. import java.util.Map;
  4. import java.util.UUID;
  5. import java.util.concurrent.ConcurrentHashMap;
  6.  
  7. public class Test {
  8. 	
  9. 	public static Map<String, Object> map = new ConcurrentHashMap<String,Object>(
  10. 			5000);
  11. 	
  12. 	public String generateUuidKey(String esCorCode){
  13. 		String uuid=UUID.randomUUID().toString();
  14. 		uuid=uuid.replaceAll("-", "");
  15. 		uuid=esCorCode+uuid;
  16. 		return uuid;
  17. 	}
  18. 	
  19. 	public static void main(String[] args) {
  20. 		
  21. 		
  22. 		Test t=new Test();
  23. 		
  24. 		// 快速验证1000000个UUID
  25. 		for (int i = 1; i <= 1000000; i++) {
  26. 			String key=t.generateUuidKey("P0000007");
  27. 			map.put(key, key);
  28. 		}
  29. 		
  30. 		System.out.println(map.size());
  31. 		
  32. 		try {
  33. 			Thread.sleep(20000);
  34. 		} catch (InterruptedException e) {
  35. 			// TODO Auto-generated catch block
  36. 			e.printStackTrace();
  37. 		}
  38. 		
  39. 	}
  40.  
  41. }
运行测试输出：1000000.

 至此，我们的事务控制配合就完美解决了线程引起的数据不一致问题。executeStockToOMSBySkuIndentifyInfo实际上就是Handler线程需要处理的方法，可以将UpdateProductStockThread线程内部的处理放到Handler稍加修改，避免线程中又开线程。



相关技术文章

最新源码

下载排行榜
点击QQ咨询
开通会员
返回顶部
×
微信扫码支付 元
微信扫码支付
确定支付下载
请使用微信描二维码支付
×

提示信息

×

选择支付方式

  • 微信支付
  • 支付宝付款
确定支付下载