基於zookeeper的MySQL主主負載均衡的簡單實現

1.先上原理圖

2.說明

兩個mysql采用主主同步的方式進行部署。

在安裝mysql的服務器上安裝客戶端（目前是這么做，以后想在zookeeper擴展集成），客戶端實時監控mysql應用的可用性，可用時想zookeepercreateNode，當網絡不可用或者mysql應用不可用時，建立的znode消失。

在客戶端，通過改造proxool數據庫連接池的方式，在建立連接之前，從zookeeper中去取真實的數據庫URL，如果有多個URL，即有多個服務時，采用隨機算法去拿連接（以后准備擴展權重）。當連接不可用時，數據庫連接池將重建連接，這時候又回去zookeeper拿連接，因為agent建立的臨時znode消失了，就不能拿到已經失效的url了。

這個方案只是初步的實驗和實現了，還有很多后續的問題，主要為了解決lvs+keepalived只能在同一個區域內的問題。

3.部分實現

　　1).agent

　　

/**
 * 數據庫可用性檢測
 * @author tomsnail
 * @date 2015年4月3日 上午10:11:51
 */
public class TestMySQL {

    public static boolean test(String url){
        
         Connection conn = null;
         Statement stmt = null;
         ResultSet rs  = null;
         String sql = ConfigHelp.getLocalConifg("jdbc_inventory.house-keeping-test-sql", "select 0");
            try {
                Class.forName(ConfigHelp.getLocalConifg("jdbc_inventory.driver-class", "com.mysql.jdbc.Driver"));// 動態加載mysql驅動
                conn = DriverManager.getConnection(url);
                stmt = conn.createStatement();
                rs = stmt.executeQuery(sql);
                while (rs.next()) {
                }
                return true;
            } catch (SQLException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    if(rs!=null){
                        rs.close();
                    }
                    if(stmt!=null){
                        stmt.close();
                    }
                    if(conn!=null)
                        conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }

            }
        return false;
    }
}

/**
 * zookeeper客戶端
 * @author tomsnail
 * @date 2015年4月3日 上午10:11:51
 */
public class TestServer {

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

    private static ZooKeeper zk;
    
    private String path;

    //同步鎖
    private Lock _lock = new ReentrantLock();
    
    // 用於等待 SyncConnected 事件觸發后繼續執行當前線程
    private CountDownLatch latch = new CountDownLatch(1);
    

    public TestServer() {
        zk = connectServer();
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        Thread.currentThread().sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //logger.info("check zk...");
                    _lock.lock();
                    if (zk != null) {
                        if (zk.getState().isAlive()
                                && zk.getState().isConnected()) {
                            //logger.info("zk is ok");
                            _lock.unlock();
                            continue;
                        }
                    }
                    close();
                    logger.info("reConnectServer ...");
                    zk = connectServer();
                    logger.info("reConnectServer ok");
                    _lock.unlock();
                }

            }

            private void close() {
                if(zk!=null){
                    try {
                        zk.close();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    zk = null;
                }
            }
        }).start();
    }



    // 連接 ZooKeeper 服務器
    private ZooKeeper connectServer() {
        ZooKeeper zk = null;
        try {

            zk = new ZooKeeper(ConfigHelp.ZK_CONNECTION_STRING,
                    ConfigHelp.ZK_SESSION_TIMEOUT, new Watcher() {
                        @Override
                        public void process(WatchedEvent event) {
                            if (event.getState() == Event.KeeperState.SyncConnected) {
                                latch.countDown(); // 喚醒當前正在執行的線程
                            }
                        }
                    });
            latch.await(); // 使當前線程處於等待狀態
        } catch (Exception e) {
            logger.error("", e);
        }
        if (zk != null) {
            try {
                Stat stat = zk.exists(ConfigHelp.ZK_ROOT_PATH, false);
                if (stat == null) {
                    String path = zk.create(ConfigHelp.ZK_ROOT_PATH,
                            "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                            CreateMode.PERSISTENT); // 創建一個臨時性且有序的 ZNode
                    logger.info("create zookeeper node ({})", path);
                }
                stat = zk.exists(ConfigHelp.ZK_RMI_PATH, false);
                if (stat == null) {

                    String path = zk.create(ConfigHelp.ZK_RMI_PATH,
                            "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                            CreateMode.PERSISTENT); // 創建一個臨時性且有序的 ZNode
                    logger.info("create zookeeper node ({})", path);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return zk;
    }

    // 創建 ZNode
    public void createNode(String url) {
        _lock.lock();
        try {
            byte[] data = url.getBytes();
            path = zk.create(ConfigHelp.ZK_RMI_PATH + "/", data,
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL); // 創建一個臨時性且有序的 ZNode
            logger.info("create zookeeper node ({} => {})", path, url);
        } catch (Exception e) {
            logger.error("", e);
            e.printStackTrace();
        }
        _lock.unlock();
    }
    
    public void deleteNode(String url){
        _lock.lock();
        try {
            Stat stat = zk.exists(path, false);
            if(stat!=null){
                zk.delete(url, stat.getVersion());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        _lock.unlock();
    }
}

/**
 * 數據庫檢測測試主類
 * @author tomsnail
 * @date 2015年4月3日 上午10:11:51
 */
public class TestMain {
    
    private static TestServer testServer = new TestServer();

    public static void main(String[] args) {
        String url = ConfigHelp.getLocalConifg("jdbc_inventory.driver-url", "select 0");
        boolean isOK = false;
        while(true){
            if(TestMySQL.test(url)){
                if(isOK){
                    
                }else{
                    testServer.createNode(url);//建立znode
                }
                isOK = true;
            }else{
                isOK = false;
                testServer.deleteNode(url);//刪除znode
            }
            
            try {
                Thread.currentThread().sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

 

　　2).proxool

/**
 * zookeeper信息定義類
 * @author tomsnail
 * @date 2015年4月2日 下午6:49:13
 */
public class ZkInfoDefinition {
    
    public static final String PREFIX_ZK = "zookeeper";
    
    public static final String ZK_URL = "zkUrl";
    
    public static final String ZK_SESSION_TIMEOUT = "sessionTimeout";
    
    public static final String ZK_PATH = "zkPath";
    
    public static final String ZK_ENABLE = "zkEnable";

    public static String zkUrl="192.168.102.1:31315";
    
    public static int sessionTimeout = 5000;
    
    public static boolean isEnable = false;
    
    public static String zkPath = "/root/db";

    public String getZkUrl() {
        return zkUrl;
    }

    public void setZkUrl(String zkUrl) {
        this.zkUrl = zkUrl;
    }

    public int getSessionTimeout() {
        return sessionTimeout;
    }

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

    public String getZkPath() {
        return zkPath;
    }

    public void setZkPath(String zkPath) {
        this.zkPath = zkPath;
    }

    public ZkInfoDefinition(String zkUrl, int sessionTimeout, String zkPath) {
        super();
        this.zkUrl = zkUrl;
        this.sessionTimeout = sessionTimeout;
        this.zkPath = zkPath;
    }
    public ZkInfoDefinition(){
        
    }
}

/**
 * zookeeper客戶端
 * @author tomsnail
 * @date 2015年4月3日 上午10:15:11
 */
public class ZkClient {

       private static final Logger logger = LoggerFactory.getLogger(ZkClient.class);
       
        // 用於等待 SyncConnected 事件觸發后繼續執行當前線程
        private CountDownLatch latch = new CountDownLatch(1);
     
        // 定義一個 volatile 成員變量，用於保存最新的 RMI 地址（考慮到該變量或許會被其它線程所修改，一旦修改后，該變量的值會影響到所有線程）
        private volatile List<String> dataList = new ArrayList<String>();
     
        private Lock _lock = new ReentrantLock();
        
        private static  ZooKeeper zk;
        
        private LBUrl lbUrl;
        
        
        public ZkClient(){
            this(new BasicLBUrl());
        }
        
        // 構造器
        public ZkClient(LBUrl lbUrl) {
            this.lbUrl = lbUrl;
            zk = connectServer(); // 連接 ZooKeeper 服務器並獲取 ZooKeeper 對象
            watchNode();
            new Thread(new Runnable() {
                
                @Override
                public void run() {
                    while (true) {
                        try {
                            Thread.currentThread().sleep(3000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        _lock.lock();
                        if (zk != null) {
                            if (zk.getState().isAlive()
                                    && zk.getState().isConnected()) {
                                _lock.unlock();
                                continue;
                            }
                        }
                        if(zk!=null){
                            try {
                                zk.close();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            zk = null;
                        }
                        zk = connectServer();
                        _lock.unlock();
                    }
                }
            }).start();
        }
     
        // 查找 URL 服務
        public String getUrl() {
            if (dataList!=null&&dataList.size()>0) {
               return this.lbUrl.getUrl(dataList);
            }
            return null;
        }
        
        public List<String> getUrls(){
            return dataList;
        }
     
        // 連接 ZooKeeper 服務器
        private ZooKeeper connectServer() {
            ZooKeeper zk = null;
            try {
                zk = new ZooKeeper(ZkInfoDefinition.zkUrl, ZkInfoDefinition.sessionTimeout, new Watcher() {
                    @Override
                    public void process(WatchedEvent event) {
                        if (event.getState() == Event.KeeperState.SyncConnected) {
                            latch.countDown(); // 喚醒當前正在執行的線程
                        }
                    }
                });
                latch.await(); // 使當前線程處於等待狀態
            } catch (Exception e) {
                logger.error("", e);
            }
            return zk;
        }
     
        // 觀察 /registry 節點下所有子節點是否有變化
        private void watchNode() {
            _lock.lock();
            if(zk!=null&&zk.getState().isAlive()&&zk.getState().isConnected()){
                
            }else{
                if(zk!=null){
                    try {
                        zk.close();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    zk = null;
                }
                zk = connectServer();
            }
            try {
                List<String> nodeList = zk.getChildren(ZkInfoDefinition.zkPath, new Watcher() {
                    @Override
                    public void process(WatchedEvent event) {
                        if (event.getType() == Event.EventType.NodeChildrenChanged) {
                            watchNode(); // 若子節點有變化，則重新調用該方法（為了獲取最新子節點中的數據）
                        }
                    }
                });
                List<String> dataList = new ArrayList<String>(); // 用於存放 /registry 所有子節點中的數據
                for (String node : nodeList) {
                    byte[] data = zk.getData(ZkInfoDefinition.zkPath + "/" + node, false, null); // 獲取 /registry 的子節點中的數據
                    dataList.add(new String(data));
                   
                }
                logger.debug("node data: {}", dataList);
                this.dataList = dataList;
            } catch (Exception e) {
                logger.error("", e);
            }
            _lock.unlock();
        }
     
        public static void main(String[] args) {
            ZkClient client = new ZkClient();
            System.out.println(client.getUrl());
        }
}

/**
 * 從zookeeper獲得URL連接操作類
 * @author tomsnail
 * @date 2015年4月2日 下午6:56:06
 */
public class ZkUrlOperation {
    
    private static final ZkUrlOperation instance = new ZkUrlOperation();

    private static ZkInfoDefinition zkInfoDefinition;
    
    private static ZkClient zkClient;
    
    private static final byte[] _lock = new byte[0];
    
    private  ZkUrlOperation(){
        
    }
    
    public static ZkUrlOperation getInstance(){
        return instance;
    }
    
    public  void addZkInfoDefinition(ZkInfoDefinition zkInfoDefinition){
        ZkUrlOperation.zkInfoDefinition = zkInfoDefinition;
    }
    
    public  void addZkInfoDefinition(String key,String value){
        if(ZkUrlOperation.zkInfoDefinition==null){
            ZkUrlOperation.zkInfoDefinition = new ZkInfoDefinition();
        }
        if(key.contains(ZkInfoDefinition.ZK_PATH)){
            ZkUrlOperation.zkInfoDefinition.setZkPath(value);
        }
        if(key.contains(ZkInfoDefinition.ZK_SESSION_TIMEOUT)){
            ZkUrlOperation.zkInfoDefinition.setSessionTimeout(Integer.valueOf(value));;
        }
        if(key.contains(ZkInfoDefinition.ZK_URL)){
            ZkUrlOperation.zkInfoDefinition.setZkUrl(value);;
        }
        if(key.contains(ZkInfoDefinition.ZK_ENABLE)){
            ZkUrlOperation.zkInfoDefinition.isEnable = Boolean.valueOf(value);
        }
    }
    
    
    public String getUrl(){
        synchronized (_lock) {
            if(zkInfoDefinition.isEnable){
                if(zkClient==null){
                    zkClient = new ZkClient();
                }
                
                String url = zkClient.getUrl();
                return url;
            }else{
                return "";
            }
            
        }
        
        
    }
    
    public boolean isAvailUrl(String url){
        synchronized (_lock) {
            if(zkInfoDefinition.isEnable){
                if(zkClient==null){
                    zkClient = new ZkClient();
                }
                List<String> urls = zkClient.getUrls();
                for(int i=0;i<urls.size();i++){
                    if(url.equals(urls.get(i))){
                        return true;
                    }
                }
                return false;
            }
            return false;
            
        }
        
    }
    
    
    
}