aggregation-platform/src/com/platform/service/OracleStatusService.java

package com.platform.service;

import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.ReplicationController;

import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

import com.platform.entities.OracleConnectorParams;
import com.platform.kubernetes.SimpleKubeClient;
import com.platform.oracle.OracleConnector;
import com.platform.utils.Configs;
import com.platform.utils.Constant;

/**
 * 计时 尝试连接oracle的时间，（超过10分钟判断失败）
 * 
 * @author chen
 * 
 */
public class OracleStatusService {
	private static Map<String, Timer> alliveTask = new Hashtable<String, Timer>();
	public final static int EXEC_TIME = 9;// 连接多少次后不成功，取消链接
	public final static long INTERVAL_TIME = 60 * 1000;// 每隔多少毫秒执行一次连接任务
	public final static long DELAY_TIME = 30 * 1000; // 延迟多少秒后执行

	public void connectToOracle(String replicasName) {
		SimpleKubeClient sKubeClient = new SimpleKubeClient();
		if (alliveTask.containsKey(replicasName)) {
			killAlliveTask(replicasName);
		}
		OracleConnectorParams orp = new OracleConnectorParams();
		Timer timer = new Timer();
		alliveTask.put(replicasName, timer);
		timer.schedule(new connectTask(replicasName, orp, sKubeClient),
				DELAY_TIME, INTERVAL_TIME);

	}

	public void cancelToOracle(String replicasName, String operate) {
		if (operate.equals("stop")) {
			String cmd = "kubectl label --overwrite rc " + replicasName + " status=0";
			List<String> rList = Constant.ganymedSSH
					.execCmdWaitAcquiescent(cmd);
			StringBuffer sb = new StringBuffer();
			for (String str : rList)
				sb.append(str).append("\n");
			Configs.CONSOLE_LOGGER.info(sb.toString());
			// SimpleKubeClient sKubeClient = new SimpleKubeClient();
			// sKubeClient.updateOrAddReplicasLabelById(replicasName,
			// "status","0");
		}
		killAlliveTask(replicasName);
	}

	/**
	 * 取消并移除指定定时任务
	 * 
	 * 
	 * @param taskName
	 */
	public void killAlliveTask(String taskName) {
		if (alliveTask.containsKey(taskName)) {
			alliveTask.get(taskName).cancel();
			alliveTask.remove(taskName);
		}
	}

	public void killAlliveTasks(String... tasksName) {
		for (String taskName : tasksName)
			killAlliveTask(taskName);
	}

	/**
	 * 清空定时任务
	 */
	public void cleanUpAlliveTask() {
		Iterator<Map.Entry<String, Timer>> iterator = alliveTask.entrySet()
				.iterator();
		while (iterator.hasNext()) {
			Map.Entry<String, Timer> entry = iterator.next();
			entry.getValue().cancel();
		}
		alliveTask.clear();
	}

	/**
	 * 链接oracle任务类
	 * 
	 * @author wuming
	 * 
	 */
	class connectTask extends TimerTask {
		private String taskName;
		private int count;
		private OracleConnectorParams ocp;
		private SimpleKubeClient client;

		public connectTask(String taskName, OracleConnectorParams ocp,
				SimpleKubeClient client) {
			this.taskName = taskName;
			this.ocp = ocp;
			this.count = 0;
			this.client = client;
		}

		public connectTask(OracleConnectorParams ocp, SimpleKubeClient client) {
			this.taskName = ocp.getName();
			this.ocp = ocp;
			this.count = 0;
			this.client = client;
		}

		@Override
		public void run() {
			if (count == EXEC_TIME && alliveTask.containsKey(taskName)) { // 如果任务已经执行10次，则任务oracle启动失败，并取消oracle连接
				killAlliveTask(taskName);
				// client.updateOrAddReplicasLabelById(taskName, "status", "1");
				// //更新ReplicationController标签，将oracle状态标示未1(0:启动中，1：失败，2：成功)
				String cmd = "kubectl label --overwrite rc " + taskName + " status=1";
				List<String> rList = Constant.ganymedSSH
						.execCmdWaitAcquiescent(cmd);
				StringBuffer sb = new StringBuffer();
				for (String str : rList)
					sb.append(str).append("\n");
				Configs.CONSOLE_LOGGER.info("更新replicationController标签： "
						+ taskName + "\t[标签更新为： 失败]");
			} else { // 否则，执行连接oracle任务，判断oracle是否启动成功
				Pod pod = filterPod();
				if (pod != null) {
					String ip = client.getPodHostIp(pod);
					int port = client.getPodContainerport(pod);
					if (ip != null && port != 0) {
						String url = "jdbc:oracle:thin:@" + ip + ":" + port
								+ ":" + ocp.getDatabaseName();
						boolean flag = OracleConnector.canConnect(url, // 连接结果返回参数，true标示连接成功，false标示连接失败
								ocp.getUser(), ocp.getPassword());
						Configs.CONSOLE_LOGGER.info("url:" + url + ",user:"
								+ ocp.getUser() + ",password:"
								+ ocp.getPassword());
						String message = "失败";
						if (flag && alliveTask.containsKey(taskName)) {
							String cmd = "kubectl label --overwrite rc " + taskName
									+ " status=2";
							// client.updateOrAddReplicasLabelById(taskNSyame,
							// "status", "2");
							List<String> rList = Constant.ganymedSSH
									.execCmdWaitAcquiescent(cmd);
							StringBuffer sb = new StringBuffer();
							for (String str : rList)
								sb.append(str).append("\n");
							Configs.CONSOLE_LOGGER.info(sb.toString());
							message = "成功";
							killAlliveTask(taskName); // 连接成功，取消连接
							Configs.CONSOLE_LOGGER
									.info("更新replicationController标签： "
											+ taskName + "\t[标签更新为： 成功]");
						}
						Configs.CONSOLE_LOGGER.info("连接到数据库服务： " + taskName
								+ "\t[连接结果： " + message + "]");
					}
				}
			}
			count++;
		}

		/**
		 * 获取oracle的连接ip地址和端口号
		 * 
		 * @return
		 */
		private Pod filterPod() {
			Pod pod = null;
			ReplicationController replicationController = client
					.getReplicationController(taskName);
			if (null != replicationController) {
				List<Pod> filterPods = client
						.getPodsForApplicaList(replicationController);
				if (filterPods != null && filterPods.size() > 0) {
					pod = filterPods.get(0);
				}
			}
			return pod;
		}

		public String getTaskName() {
			return taskName;
		}

		public void setTaskName(String taskName) {
			this.taskName = taskName;
		}

		public int getCount() {
			return count;
		}
	}
}