zookeeper分布式锁解决集群定时任务重复执行问题_zk分布式锁 防止任务重复执行

package cn.biz.impl;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.stereotype.Component;

/**
 * @Description 可动态更改时间的定时任务
 * @Author
 * @Date 2021/3/2
 * @Version 1.0.0
 */
@Component
public class SchedulingTask implements SchedulingConfigurer {

    private final Logger logger = LoggerFactory.getLogger(SchedulingTask .class);


    // cron表达式，我们动态更改此属性的值即可更改定时任务的执行时间
    private String expression = "0 0/2 * * * ?";

    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {

        // 定时任务要执行的方法 
        Runnable task = () -> {
	       	//1、重试策略：初试时间为1s 重试3次
	        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
	        //2、通过工厂创建连接
	        CuratorFramework client = CuratorFrameworkFactory.newClient(zkAddress, retryPolicy);
	        //3、开启连接
	        client.start();
	        //4、获取zk分布式锁
	        InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock");
	 		boolean flag = false;
	        try {
	        	//尝试获取锁，最多等待0秒
	            flag = mutex.acquire(0, TimeUnit.SECONDS);
	            Thread currentThread = Thread.currentThread();
	            if(flag){//锁抢占成功 执行业务逻辑
	                logger.info(">>> configureTasks-----------【"+expression+"】");
	                 //开始业务逻辑
	            }
	            //
	            Thread.sleep(1000);
	        } catch (Exception e) {
	            e.printStackTrace();
	        } finally{
	            if(flag){
	                //1.先释放锁
                    if(flag){
                        try {
                            mutex.release();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    //2.后关闭zk客户端连接
                    if (client != null) {
                        client.close();
                    }
	            }
	        }
        };
        // 调度实现的时间控制
        Trigger trigger = triggerContext -> {
            CronTrigger cronTrigger = new CronTrigger(expression);
            return cronTrigger.nextExecutionTime(triggerContext);
        };
        taskRegistrar.addTriggerTask(task, trigger);
    }

    public String getExpression() {
        return expression;
    }

    public void setExpression(String expression) {
        this.expression = expression;
    }
}
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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86