我们如何在代码中实现主从的选择-Java

Posted 2022-05-20 小肥宅Lee

问题描述：我们在分布式任务系统之中，会经常遇见一种情况，比如：定时的对任务进行刷新，然后进行一些操作。通常我们会将这个定时任我Job单独部署成一个服务，这个服务如果我们不进行主选择，便会出现多个服务同时进行刷新，造成业务的紊乱。

解决方案：使用Zookeeper来进行主的选择

1、添加对应的依赖

<!-- zookeeper 客户端 -->
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>2.13.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>2.13.0</version>
</dependency>

code

package com.lee.schedule;


import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author Lee
 * @date 2022-05-18 17:02
 */
public class SelectMasterTest 

	//自己的zookeeper地址
    private String selectMasterZookeeper="localhost:2181";

    //可以放很多节点
    Map<String, Boolean> masterMap = new HashMap<String, Boolean>();


    /**
     * 选主
     * @param leaderPath  zookeeper目录节点
     */
    public void selectMaster (String leaderPath) 
        CuratorFramework client = CuratorFrameworkFactory.builder().
                connectString(selectMasterZookeeper)
                .sessionTimeoutMs(5000)  //超时时间
                .retryPolicy(new ExponentialBackoffRetry(1000, 3)) //连接不上重试三次
                .build();
        client.start();

        //争抢注册节点
        @SuppressWarnings("resource")
        LeaderSelector selector = new LeaderSelector(client, leaderPath,
                new LeaderSelectorListenerAdapter() 

                    @Override
                    public void takeLeadership(CuratorFramework client) throws Exception 
                        //如果争抢到当前注册节点
                        masterMap.put(leaderPath, true);
                        while (true) 
                            //抢占当前节点
                            TimeUnit.SECONDS.sleep(3);
                        
                    
                );
        masterMap.put(leaderPath, false);
        selector.autoRequeue();
        selector.start();
    

    public boolean checkMaster (String leaderPath) 
        Boolean isMaster = masterMap.get(leaderPath);
        return isMaster == null ? false : isMaster;
    

    public static void main(String[] args) throws InterruptedException 

		//leaderPath 可以根据我们的业务情况，设置对应的业务path
        String leaderPath = "/master";
        SelectMasterTest selectMaster = new SelectMasterTest();
        selectMaster.selectMaster(leaderPath);

        TimeUnit.SECONDS.sleep(1);

        while(true) 
            if(selectMaster.checkMaster(leaderPath))
                System.out.println(" 节点1 主节点 ");
            else 
                System.out.println(" 节点1 从节点 ");
            
            TimeUnit.SECONDS.sleep(6);