1.先上原理圖

2.說(shuō)明

兩個(gè)mysql采用主主同步的方式進(jìn)行部署。

在安裝mysql的服務(wù)器上安裝客戶端（目前是這么做，以后想在zookeeper擴(kuò)展集成），客戶端實(shí)時(shí)監(jiān)控mysql應(yīng)用的可用性，可用時(shí)想zookeepercreateNode，當(dāng)網(wǎng)絡(luò)不可用或者mysql應(yīng)用不可用時(shí)，建立的znode消失。

在客戶端，通過(guò)改造PRoxool數(shù)據(jù)庫(kù)連接池的方式，在建立連接之前，從zookeeper中去取真實(shí)的數(shù)據(jù)庫(kù)URL，如果有多個(gè)URL，即有多個(gè)服務(wù)時(shí)，采用隨機(jī)算法去拿連接（以后準(zhǔn)備擴(kuò)展權(quán)重）。當(dāng)連接不可用時(shí)，數(shù)據(jù)庫(kù)連接池將重建連接，這時(shí)候又回去zookeeper拿連接，因?yàn)閍gent建立的臨時(shí)znode消失了，就不能拿到已經(jīng)失效的url了。

這個(gè)方案只是初步的實(shí)驗(yàn)和實(shí)現(xiàn)了，還有很多后續(xù)的問(wèn)題，主要為了解決lvs+keepalived只能在同一個(gè)區(qū)域內(nèi)的問(wèn)題。

3.部分實(shí)現(xiàn)

　　1).agent

/** * 數(shù)據(jù)庫(kù)可用性檢測(cè) * @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"));// 動(dòng)態(tài)加載mysql驅(qū)動(dòng)                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 事件觸發(fā)后繼續(xù)執(zhí)行當(dāng)前線程    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 服務(wù)器    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(); // 喚醒當(dāng)前正在執(zhí)行的線程                            }                        }                    });            latch.await(); // 使當(dāng)前線程處于等待狀態(tài)        } 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); // 創(chuàng)建一個(gè)臨時(shí)性且有序的 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); // 創(chuàng)建一個(gè)臨時(shí)性且有序的 ZNode                    logger.info("create zookeeper node ({})", path);                }            } catch (Exception e) {                e.printStackTrace();            }        }        return zk;    }    // 創(chuàng)建 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); // 創(chuàng)建一個(gè)臨時(shí)性且有序的 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();    }}

/** * 數(shù)據(jù)庫(kù)檢測(cè)測(cè)試主類 * @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 事件觸發(fā)后繼續(xù)執(zhí)行當(dāng)前線程        private CountDownLatch latch = new CountDownLatch(1);             // 定義一個(gè) volatile 成員變量，用于保存最新的 RMI 地址（考慮到該變量或許會(huì)被其它線程所修改，一旦修改后，該變量的值會(huì)影響到所有線程）        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());        }                // 構(gòu)造器        public ZkClient(LBUrl lbUrl) {            this.lbUrl = lbUrl;            zk = connectServer(); // 連接 ZooKeeper 服務(wù)器并獲取 ZooKeeper 對(duì)象            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 服務(wù)        public String getUrl() {            if (dataList!=null&&dataList.size()>0) {               return this.lbUrl.getUrl(dataList);            }            return null;        }                public List<String> getUrls(){            return dataList;        }             // 連接 ZooKeeper 服務(wù)器        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(); // 喚醒當(dāng)前正在執(zhí)行的線程                        }                    }                });                latch.await(); // 使當(dāng)前線程處于等待狀態(tài)            } catch (Exception e) {                logger.error("", e);            }            return zk;        }             // 觀察 /registry 節(jié)點(diǎn)下所有子節(jié)點(diǎn)是否有變化        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(); // 若子節(jié)點(diǎn)有變化，則重新調(diào)用該方法（為了獲取最新子節(jié)點(diǎn)中的數(shù)據(jù)）                        }                    }                });                List<String> dataList = new ArrayList<String>(); // 用于存放 /registry 所有子節(jié)點(diǎn)中的數(shù)據(jù)                for (String node : nodeList) {                    byte[] data = zk.getData(ZkInfoDefinition.zkPath + "/" + node, false, null); // 獲取 /registry 的子節(jié)點(diǎn)中的數(shù)據(jù)                    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;                    }            }            }