多节点,多线程下发订单,使用zookeeper分布式锁机制保证订单正确接入oms系统

 假设订单下发, 采用单机每分钟从订单OrderEntry接口表中抓100单, 接入订单oms系统中. 由于双十一期间, 订单量激增, 导致订单单机每分钟100单造成, 订单积压. 所以采用多节点多线程处理三方商家的订单接入系统.

流程图:

如图所示:

采用多节点, 多线程, 使用curator操作zk的方式实现, 订单的下发.

以前的方式每分钟下发100单, 之后采用现在的方式可以每分钟2000单, 大大增加系统的订单下发的处理速度, 系统的吞吐量, 提高了效率.

1 操作zookeeper的工具类:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
1import org.apache.curator.RetryPolicy;

2import org.apache.curator.framework.CuratorFramework;

3import org.apache.curator.framework.CuratorFrameworkFactory;

4import org.apache.curator.framework.recipes.locks.InterProcessMutex;

5import org.apache.curator.retry.ExponentialBackoffRetry;

6 

7public class ZkUtils {

8 

9    private static final String zkStr = ConfUtil.getProperty("zkStr");

10    private static final String zkPath = ConfUtil.getProperty("zkPath");

11    

12    public static InterProcessMutex createMutexLock(){

13        

14        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10); //1 重试策略：初试时间为1s 重试10次

15        

16        CuratorFramework client = CuratorFrameworkFactory.builder()//2 通过工厂创建连接

17                    .connectString(zkStr)

18                    .sessionTimeoutMs(5000)

19                    .retryPolicy(retryPolicy)

20                    .build();

21        client.start(); // 开启连接

22        InterProcessMutex mutexLock = new InterProcessMutex(client, zkPath); // 创建分布式锁

23        return mutexLock;

24        

25    }

26}

27

2 使用该工具类处理订单下发业务逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
1public class Job {

2    private final static org.slf4j.Logger logger=LoggerFactory.getLogger(HnJob.class);

3 

4    @Autowired

5    private IHNOrderService iHNOrderService;

6    

7    public void execute(){

8        

9        InterProcessMutex mutexLock = ZkUtils.createMutexLock();

10        try{

11            mutexLock.acquire();

12            bizHandler(); //执行订单下发业务处理逻辑

13            

14        }catch (Exception e) {

15            e.printStackTrace();

16        }finally{

17            try {

18                mutexLock.release();

19            } catch (Exception e) {

20                e.printStackTrace();

21            } 

22        }

23    }

24    

25    void bizHandler(){

26        long start=System.currentTimeMillis();

27        logger.info("====================订单下发开始");

28        iHNOrderService.hnOrderQuery();

29        long end=System.currentTimeMillis();

30        logger.info("=================订单下发结束,耗时={}秒",(end-start)/(1000));

31    }

32    

33    

34    

35}

36