重构AlarmRedisConnector类

skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/handler/CollectionServerDataHandler.java

@@ -2,7 +2,7 @@ package com.ai.cloud.skywalking.reciever.handler;
 
 import com.ai.cloud.skywalking.reciever.buffer.DataBufferThreadContainer;
 import com.ai.cloud.skywalking.reciever.conf.Config;
-import com.ai.cloud.skywalking.reciever.storage.AlarmRedisConnector;
+import com.ai.cloud.skywalking.reciever.util.RedisConnector;
 import com.ai.cloud.skywalking.util.ProtocolPackager;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
@@ -32,7 +32,7 @@ public class CollectionServerDataHandler extends SimpleChannelInboundHandler<byt
     private void dealFailedPackage(ChannelHandlerContext ctx) {
         InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().localAddress();
         String key = ctx.name() + "-" + socketAddress.getHostName() + ":" + socketAddress.getPort();
-        Jedis jedis = AlarmRedisConnector.getJedis();
+        Jedis jedis = RedisConnector.getJedis();
         if (jedis.setnx(key, 0 + "") == 1) {
             jedis.expire(key, Config.Server.FAILED_PACKAGE_WATCHING_TIME_WINDOW);
         }

skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/alarm/AbstractSpanChecker.java

@@ -7,7 +7,7 @@ import org.apache.logging.log4j.Logger;
 
 import redis.clients.jedis.Jedis;
 
-import com.ai.cloud.skywalking.reciever.storage.AlarmRedisConnector;
+import com.ai.cloud.skywalking.reciever.util.RedisConnector;
 
 public abstract class AbstractSpanChecker implements ISpanChecker {
 	private static Logger logger = LogManager.getLogger(AbstractSpanChecker.class);
@@ -15,11 +15,11 @@ public abstract class AbstractSpanChecker implements ISpanChecker {
 	protected void saveAlarmMessage(String key, String traceId, String alarmMsg) {
         Jedis jedis = null;
         try {
-            jedis = AlarmRedisConnector.getJedis();
+            jedis = RedisConnector.getJedis();
             jedis.hsetnx(key, traceId, alarmMsg);
             jedis.expire(key, ALARM_EXPIRE_SECONDS);
         } catch (Exception e) {
-        	AlarmRedisConnector.reportJedisFailure();
+        	RedisConnector.reportJedisFailure();
             logger.error("Failed to set data.", e);
         } finally {
             if (jedis != null) {

skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/RedisConnector.java

+package com.ai.cloud.skywalking.reciever.util;
+
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
+
+import com.ai.cloud.skywalking.reciever.conf.Config;
+import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
+import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading;
+
+/**
+ * redis连接器，用于管理redis连接
+ * 
+ * @author wusheng
+ *
+ */
+public class RedisConnector {
+	private static JedisPool jedisPool;
+
+	static {
+		new RedisInspector().connect().start();
+	}
+
+	public static Jedis getJedis() {
+		if (Config.Alarm.ALARM_OFF_FLAG) {
+			return null;
+		} else {
+			return jedisPool.getResource();
+		}
+	}
+
+	public static void reportJedisFailure() {
+		RedisInspector.needConnectInit = true;
+	}
+
+	private static class RedisInspector extends Thread {
+		private static Logger logger = LogManager
+				.getLogger(RedisInspector.class);
+
+		private static boolean needConnectInit = true;
+
+		private String[] config;
+
+		public RedisInspector() {
+			super("RedisInspectorThread");
+			String redisServerConfig = Config.Redis.REDIS_SERVER;
+			if (redisServerConfig == null || redisServerConfig.length() <= 0) {
+				logger.error("Redis server is not setting. Switch off alarm module. ");
+				Config.Alarm.ALARM_OFF_FLAG = true;
+			} else {
+				config = redisServerConfig.split(":");
+				if (config.length != 2) {
+					logger.error("Redis server address is illegal setting, need to be 'ip:port'. Switch off alarm module. ");
+					Config.Alarm.ALARM_OFF_FLAG = true;
+				}
+			}
+		}
+
+		private RedisInspector connect() {
+			if (jedisPool != null && !jedisPool.isClosed()) {
+				jedisPool.close();
+			}
+
+			GenericObjectPoolConfig genericObjectPoolConfig = buildGenericObjectPoolConfig();
+			jedisPool = new JedisPool(genericObjectPoolConfig, config[0],
+					Integer.valueOf(config[1]));
+			// Test connect redis.
+			Jedis jedis = null;
+			try {
+				jedis = jedisPool.getResource();
+				jedis.get("ok");
+				needConnectInit = false;
+			} catch (Exception e) {
+				logger.error("can't connect to redis["
+						+ Config.Redis.REDIS_SERVER + "]", e);
+			} finally {
+				if (jedis != null) {
+					jedis.close();
+				}
+			}
+			return this;
+		}
+
+		@Override
+		public void run() {
+			if (Config.Alarm.ALARM_OFF_FLAG)
+				return;
+
+			while (true) {
+				try {
+					if (needConnectInit) {
+						connect();
+					}
+
+					if (needConnectInit) {
+						ServerHealthCollector.getCurrentHeathReading(null)
+								.updateData(ServerHeathReading.ERROR,
+										"Failed to connect the redis.");
+					} else {
+						ServerHealthCollector.getCurrentHeathReading(null)
+								.updateData(ServerHeathReading.INFO,
+										"Success to connect the redis");
+					}
+				} catch (Throwable t) {
+					logger.error("redis init connect failue", t);
+				}
+
+				try {
+					Thread.sleep(Config.Alarm.ALARM_REDIS_INSPECTOR_INTERVAL);
+				} catch (InterruptedException e) {
+					logger.error("Failure sleep.", e);
+				}
+			}
+		}
+
+		private GenericObjectPoolConfig buildGenericObjectPoolConfig() {
+			GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
+			genericObjectPoolConfig.setTestOnBorrow(true);
+			genericObjectPoolConfig.setMaxIdle(Config.Redis.REDIS_MAX_IDLE);
+			genericObjectPoolConfig.setMinIdle(Config.Redis.REDIS_MIN_IDLE);
+			genericObjectPoolConfig.setMaxTotal(Config.Redis.REDIS_MAX_TOTAL);
+			return genericObjectPoolConfig;
+		}
+	}
+}