zookeeper 主备切换方案实现分布式锁

程序员文章站 2022-05-03 16:18:46

...

zookeeper 主备切换方案实现分布式锁

   点关注不迷路，欢迎再访！

精简博客内容，尽量已行业术语来分享。
努力做到对每一位认可自己的读者负责。
帮助别人的同时更是丰富自己的良机。

Zookeeper 是一个开源的分布式协调服务框架，利用其强一致性，能够很好地保证在分布式高并发情况下节点的创建一定能够保证全局唯一性，即 ZooKeeper 将会保证客户端无法创建一个已经存在的 ZNode。也就是说，如果同时有多个客户端请求创建同一个临时节点，那么最终一定只有一个客户端请求能够创建成功。利用这个特性，就能很容易地在分布式环境中进行 Master 选举了。

成功创建该节点的客户端所在的机器就成为了 Master。同时，其他没有成功创建该节点的客户端，都会在该节点上注册一个节点变更的 Watcher，用于监控当前 Master 机器是否存活，一旦发现当前的 Master 挂了，那么其他客户端将会重新进行 Master 选举。这样就实现了 Master 的动态选举。。

zookeeper 主备切换方案实现分布式锁

二.引入依赖

<dependency>
	<groupId>org.apache.zookeeper</groupId>
	<artifactId>zookeeper</artifactId>
	<version>3.4.9</version>
</dependency>

<dependency>
	<groupId>org.apache.curator</groupId>
	<artifactId>curator-recipes</artifactId>
	<version>4.2.0</version>
</dependency>

三.实现代码

3.1封装zookeeper参数

public class ZKConfig implements Serializable{
	
	private static final long serialVersionUID = -5837486392158655001L;
	private  String connectString;
    private  int sessionTimeout;
    private  String znodeName;
    private  int connectionTimeout;

	public String getConnectString() {
		return connectString;
	}

	public void setConnectString(String connectString) {
		this.connectString = connectString;
	}

	public int getSessionTimeout() {
		return sessionTimeout;
	}

	public void setSessionTimeout(int sessionTimeout) {
		this.sessionTimeout = sessionTimeout;
	}

	public String getZnodeName() {
		return znodeName;
	}

	public void setZnodeName(String znodeName) {
		this.znodeName = znodeName;
	}

	public int getConnectionTimeout() {
		return connectionTimeout;
	}

	public void setConnectionTimeout(int connectionTimeout) {
		this.connectionTimeout = connectionTimeout;
	}    
}

3.2定义Zookeeper客户端基本配置

@Configuration
public class LeaderSelectorConfig {
	private final static Logger logger = LoggerFactory.getLogger(LeaderSelectorConfig.class);
	
	@Value("${zk.connectString}")
    String connectString;
	
	@Value("${zk.sessionTimeout}")
    int sessionTimeout;
	
	@Value("${zk.connectionTimeout}")
    int connectionTimeout;
	
	@Value("${disconf.env}")
    String env;
	
	@Value("${disconf.appName}")
    String appName;
    
	@Bean
    public ZKConfig zkConfig() throws IOException {
		ZKConfig zkConfig = new ZKConfig();
		zkConfig.setConnectionTimeout(connectionTimeout);
		zkConfig.setSessionTimeout(sessionTimeout);
		zkConfig.setConnectString(connectString);
		logger.info("当前应用名："+appName);
		logger.info("当前环境："+env);
        //ZNode路径为"应用名称"+"环境"，保证不同环境下独立选主
		zkConfig.setZnodeName("/ZKLeaderSelector/zk/master/"+appName+"/"+env);
		return zkConfig;
    }
    
}

3.3 启动 LeaderSelector 选主监听

Zookeeper的节点监听机制，可以保障占有锁的方式有序而且高效。每个线程抢占锁之前，先抢号创建自己的ZNode。同样，释放锁的时候，就需要删除抢号的Znode。抢号成功后，如果不是排号最小的节点，就处于等待通知的状态。等谁的通知呢？不需要其他人，只需要等前一个Znode 的通知就可以了。当前一个Znode 删除的时候，就是轮到了自己占有锁的时候，击鼓传花似的依次向后。

其实，Zookeeper的内部机制，能保证后面的节点能够正常的监听到删除和获得锁。在创建取号节点的时候，尽量创建临时znode 节点而不是永久znode 节点，一旦这个 znode 的客户端与Zookeeper集群服务器失去联系，这个临时 znode 也将自动删除。排在它后面的那个节点，也能收到删除事件，从而获得锁。

说Zookeeper的节点监听机制，是非常完美的？

Zookeeper这种首尾相接，后面监听前面的方式，可以避免羊群效应。所谓羊群效应就是每个节点挂掉，所有节点都去监听，然后做出反映，这样会给服务器带来巨大压力，所以有了临时顺序节点，当一个节点挂掉，只有它后面的那一个节点才做出反映。

@Component("leaderSelector")
public class LeaderSelector {
	
	private final static Logger logger = LoggerFactory.getLogger(LeaderSelector.class);

	@Autowired
	private ZKConfig zkConfig;
	
    private CuratorFramework client;

	private LeaderLatch latch;
	
    //全局变量标识-当前节点是否为master
    private boolean isMaster = false;
    
    @PostConstruct
    public void startSelector() {
        this.start();
    }
    
    /**
     * 
     *<p>
     *description:启动选主监听
     *</p>
     * @author andy
     * @see
     */
    public void start() {
        client = CuratorFrameworkFactory.builder()
                .connectString(zkConfig.getConnectString())
                .connectionTimeoutMs(zkConfig.getConnectionTimeout())
                .sessionTimeoutMs(zkConfig.getSessionTimeout())
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .connectionStateErrorPolicy(new SessionConnectionStateErrorPolicy())
                .build();

        latch = new LeaderLatch(client, zkConfig.getZnodeName());

        try {
            client.start();
            //添加监听程序，通过心跳检测判断选主
            //LeaderLatchListener 是LeaderLatch客户端节点成为Leader后的回调方法，有isLeader()，notLeader()两个方法
            latch.addListener(new LeaderLatchListener() {
                public void isLeader() {
                	isMaster = true;
                	logger.info("[ZKLeaderSelector]Current node is  a master.");
                }

                public void notLeader() {
                	isMaster = false;
                	logger.info("[ZKLeaderSelector]Current node is a slaver.");
                }
            });
            latch.start();
        } catch (Exception e) {
            logger.error("启动ZK选主监听程序失败！"+e.getMessage());
        }
    }

    public void stop() {
        CloseableUtils.closeQuietly(latch);
        CloseableUtils.closeQuietly(client);
    }

	public boolean getIsMaster() {
		return isMaster;
	}

	public void setIsMaster(boolean isMaster) {
		this.isMaster = isMaster;
	}
  
}

3.4编写自动任务测试

@Component
@Configuration
@EnableScheduling
public class ZKConsumerTask {
	
	private final Log logger = LogFactory.getLog(getClass());

	@Autowired
	private LeaderSelector leaderSelector;
	
	@Scheduled(cron = "${time.task}")
	public void consumerFileTask() {
			
		if(leaderSelector.getIsMaster()) {
			logger.info("当前节点为master！ Date:" + LocalDate.now());
			//执行业务逻辑
		}else {
			logger.info("当前节点为salve！ Date:" + LocalDate.now());
			//退出当前任务
		}
	}
}

3.5 application.propertiesZK 客户端参数配置

time.task = 0 0/01 * * * ?
zk.connectString = X.X.X.X:2181
zk.sessionTimeout = 5000
zk.connectionTimeout = 3000
disconf.env = sit
disconf.appName = andy