分布式选主--利用MysqlACID和Lease协议实现选主和高可用

 一些配置信息和时间相关、休眠周期相关的时间变量

 final long interval = lease / intervalDivisor;
 long waitForLeaseChallenging = 0L; 
 lease = lease / 1000L;
 long challengeFailTimes = 0L; 
 long takeRest = 0L; 
 long dbExceptionTimes = 0L; 
 long offlineTime = 0L; 
 Random rand = new Random();
 Status stateMechine = Status.START;
 long activeNodeLease = 0L; 
 long activeNodeTimeStamp = 0L; 

 KeepAlive keepaliveNode = null;
 try {
 /* first of all get it from mysql */
 keepaliveNode = dbService.accquireAliveNode();
 if (stateMechine != Status.START && keepaliveNode==null)
 throw new Exception();
 // recount , avoid network shake
 dbExceptionTimes = 0L;
 } catch (Exception e) {
 log.fatal("[Scanner] Database Exception with times : " + dbExceptionTimes++);
 if (stateMechine == Status.OFFLINE) {
 log.warn("[Scanner] Database Exception , OFFLINE ");
 } else if (dbExceptionTimes >= 3) {
 log.fatal("[Scanner] Database Exception , Node Offline Mode Active , uniqueid : " + uniqueID);
 stateMechine = Status.OFFLINE;
 dbExceptionTimes = 0L;
 offlineTime = System.currentTimeMillis();
 online = false;
 } else
 continue;
 }

 while (true) {
 SqlSession session = dbConnecction.openSession();
 ActionScanMapper dbService = session.getMapper(ActionScanMapper.class);
 KeepAlive keepaliveNode = null;
 try {
 /* first of all get it from mysql */
 keepaliveNode = dbService.accquireAliveNode();
 if (stateMechine != Status.START && keepaliveNode==null)
 throw new Exception();
 // recount , avoid network shake
 dbExceptionTimes = 0L;
 } catch (Exception e) {
 log.fatal("[Scanner] Database Exception with times : " + dbExceptionTimes++);
 if (stateMechine == Status.OFFLINE) {
 log.warn("[Scanner] Database Exception , OFFLINE ");
 } else if (dbExceptionTimes >= 3) {
 log.fatal("[Scanner] Database Exception , Node Offline Mode Active , uniqueid : " + uniqueID);
 stateMechine = Status.OFFLINE;
 dbExceptionTimes = 0L;
 offlineTime = System.currentTimeMillis();
 online = false;
 } else
 continue;
 }
 try {
 activeNodeLease = keepaliveNode!=null ? keepaliveNode.getLease() : activeNodeLease;
 activeNodeTimeStamp = keepaliveNode!=null ? keepaliveNode.getTimestamp() : activeNodeTimeStamp;
 takeRest = interval;
 switch (stateMechine) {
 case START:
 if (keepaliveNode == null) {
 log.fatal("[START] Accquire node is null , ignore ");
 // if no node register here , we challenge it
 stateMechine = Status.CHALLENGE_REGISTER;
 takeRest = 0;
 } else {
 // check the lease , wether myself or others 
 if (activeNodeLease < timestampGap(activeNodeTimeStamp)) {
 log.warn("[START] Lease Timeout scanner for uniqueid : " + uniqueID + ", timeout : "
 + timestampGap(activeNodeTimeStamp));
 if (keepaliveNode.getStatus().equals(STAT_CHALLENGE))
 stateMechine = Status.HEARTBEAT;
 else {
 stateMechine = Status.CHALLENGE_MASTER;
 takeRest = 0;
 }
 } else if (keepaliveNode.getUniqueID().equals(uniqueID)) {
 // I'am restart
 log.info("[START] Restart Scanner for uniqueid : " + uniqueID
 + ", timeout : " + timestampGap(activeNodeTimeStamp));
 stateMechine = Status.HEARTBEAT;
 } else {
 log.info("[START] Already Exist Keepalive Node with uniqueid : " + uniqueID);
 stateMechine = Status.HEARTBEAT;
 }
 }
 break;
 case HEARTBEAT:
 /* uniqueID == keepaliveNode.uniqueID */
 if (keepaliveNode.getUniqueID().equals(uniqueID)) {
 if (activeNodeLease < timestampGap(activeNodeTimeStamp)) {
 // we should challenge now , without nessesary to checkout Status[CHALLENGE]
 log.warn("[HEARTBEAT] HEART BEAT Lease is timeout for uniqueid : " + uniqueID
 + ", time : " + timestampGap(activeNodeTimeStamp));
 stateMechine = Status.CHALLENGE_MASTER;
 takeRest = 0;
 break;
 } else {
 // lease ok , just update mysql keepalive status
 dbService.updateAliveNode(keepaliveNode.setLease(lease));
 online = true;
 log.info("[HEARTBEAT] update equaled keepalive node , uniqueid : " + uniqueID
 + ", lease : " + lease + "s, remain_usable : " +
 ((activeNodeTimeStamp * 1000L + lease * 1000L) - System.currentTimeMillis()) + " ms");
 }
 } else {
 /* It's others , let's check lease */
 if (activeNodeLease < timestampGap(activeNodeTimeStamp)) {
 if (keepaliveNode.getStatus().equals(STAT_CHALLENGE)) {
 waitForLeaseChallenging = (long) (activeNodeLease * awaitFactor);
 if ((waitForLeaseChallenging) < timestampGap(activeNodeTimeStamp)) {
 log.info("[HEARTBEAT] Lease Expired , Diff[" + timestampGap(activeNodeTimeStamp) + "] , Lease[" + activeNodeLease + "]");
 stateMechine = Status.CHALLENGE_MASTER;
 takeRest = 0;
 } else {
 log.info("[HEARTBEAT] Other Node Challenging , We wait for a moment ...");
 }
 } else {
 log.info("[HEARTBEAT] Lease Expired , Diff[" + timestampGap(activeNodeTimeStamp) + "] , lease[" + activeNodeLease + "]");
 stateMechine = Status.CHALLENGE_MASTER;
 takeRest = 0;
 }
 } else {
 online = false;
 log.info("[HEARTBEAT] Exist Active Node On The Way with uniqueid : "
 + keepaliveNode.getUniqueID() + ", lease : " + keepaliveNode.getLease());
 }
 }
 break;
 case CHALLENGE_MASTER:
 dbService.challengeAliveNode(new KeepAlive().setUniqueID(uniqueID).setLease(lease));
 online = false;
 // wait for the expired node offline automatic
 // and others also have changce to challenge
takeRest = activeNodeLease;
 stateMechine = Status.CHALLENGE_COMPLETE;
 log.info("[CHALLENGE_MASTER] Other Node is timeout["
 + timestampGap(activeNodeTimeStamp) + "s] , I challenge with uniqueid : " + uniqueID
 + ", lease : " + lease + ", wait : " + lease);
 break;
 case CHALLENGE_REGISTER:
 dbService.registerNewNode(new KeepAlive().setUniqueID(uniqueID).setLease(lease));
 online = false;
 // wait for the expired node offline automatic 
 // and others also have changce to challenge
 takeRest = activeNodeLease;
 stateMechine = Status.CHALLENGE_COMPLETE;
 log.info("[CHALLENGE_REGISTER] Regiter Keepalive uniqueid : " + uniqueID + ", lease : " + lease);
 break;
 case CHALLENGE_COMPLETE :
 if (keepaliveNode.getUniqueID().equals(uniqueID)) {
 dbService.updateAliveNode(keepaliveNode.setLease(lease));
 online = true;
 log.info("[CHALLENGE_COMPLETE] I Will be the Master uniqueid : " + uniqueID);
 // make the uptime correct
 stateMechine = Status.HEARTBEAT;
 } else {
 online = false;
 log.warn("[CHALLENGE_COMPLETE] So unlucky , Challenge Failed By Other Node with uniqueid : " + keepaliveNode.getUniqueID());
 if (challengeFailTimes++ >= (rand.nextLong() % maxChallenge) + minChallenge) {
 // need't challenge anymore in a long time
 takeRest=maxChallengeAwaitInterval;
 stateMechine = Status.HEARTBEAT;
 challengeFailTimes = 0L;
 log.info("[CHALLENGE_COMPLETE] Challenge Try Times Used Up , let's take a long rest !");
 } else {
stateMechine = Status.HEARTBEAT;
 log.info("[CHALLENGE_COMPLETE] Challenge Times : " + challengeFailTimes + ", Never Give Up , to[" + stateMechine + "]");
 }
 }
 break;
 case OFFLINE :
 log.fatal("[Scanner] Offline Mode Node with uniqueid : " + uniqueID);
 if (System.currentTimeMillis() - offlineTime >= maxOfflineFrozen) {
 // I am relive forcely
 log.info("[Scanner] I am relive to activie node , uniqueid : " + uniqueID);
 stateMechine = Status.HEARTBEAT;
 offlineTime = 0L;
 } else if (keepaliveNode != null) {
 // db is reconnected
 stateMechine = Status.HEARTBEAT;
 offlineTime = 0L;
 log.info("[Scanner] I am relive to activie node , uniqueid : " + uniqueID);
 }
 break;
 default :
 System.exit(0);
 }
 session.commit();
 session.close();
 if (takeRest != 0)
 Thread.sleep(takeRest);
 log.info("[Scanner] State Stage [" + stateMechine + "]");
 } catch (InterruptedException e) {
 log.fatal("[System] Thread InterruptedException : " + e.getMessage());
 } finally {
 log.info("[Scanner] UniqueID : " + uniqueID + ", Mode : " + (online?"online":"offline"));
 }
 }
 }
 enum Status {
 START, HEARTBEAT, CHALLENGE_MASTER, CHALLENGE_REGISTER, CHALLENGE_COMPLETE, OFFLINE
 }