搜狐视频(sohu tv)Redis私有云平台 ：支持Redis多种架构Standalone、Sentinel、Cluster

package com.sohu.cache.redis.impl; import com.google.common.collect.HashBasedTable; import com.google.common.collect.Table; import com.sohu.cache.async.AsyncService; import com.sohu.cache.async.AsyncThreadPoolFactory; import com.sohu.cache.async.KeyCallable; import com.sohu.cache.constant.*; import com.sohu.cache.dao.*; import com.sohu.cache.entity.*; import com.sohu.cache.exception.SSHException; import com.sohu.cache.machine.MachineCenter; import com.sohu.cache.protocol.MachineProtocol; import com.sohu.cache.protocol.RedisProtocol; import com.sohu.cache.redis.RedisCenter; import com.sohu.cache.redis.enums.RedisInfoEnum; import com.sohu.cache.redis.enums.RedisReadOnlyCommandEnum; import com.sohu.cache.redis.util.*; import com.sohu.cache.report.ReportDataComponent; import com.sohu.cache.ssh.SSHService; import com.sohu.cache.ssh.SSHUtil; import com.sohu.cache.stats.instance.InstanceStatsCenter; import com.sohu.cache.task.BaseTask; import com.sohu.cache.task.constant.InstanceInfoEnum.InstanceTypeEnum; import com.sohu.cache.util.*; import com.sohu.cache.web.enums.BooleanEnum; import com.sohu.cache.web.enums.ClientTypeEnum; import com.sohu.cache.web.enums.SuccessEnum; import com.sohu.cache.web.service.AppService; import com.sohu.cache.web.service.ModuleService; import com.sohu.cache.web.service.WebClientComponent; import com.sohu.cache.web.util.DateUtil; import com.sohu.cache.web.vo.RedisSlowLog; import org.apache.commons.collections.CollectionUtils; import org.apache.commons.collections.MapUtils; import org.apache.commons.lang.StringUtils; import org.apache.commons.pool2.impl.GenericObjectPoolConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Lazy; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.AsyncResult; import org.springframework.stereotype.Service; import org.springframework.util.Assert; import redis.clients.jedis.DebugParams; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; import redis.clients.jedis.JedisSentinelPool; import redis.clients.jedis.Pipeline; import redis.clients.jedis.Protocol; import redis.clients.jedis.ScanParams; import redis.clients.jedis.ScanResult; import redis.clients.jedis.HostAndPort; import redis.clients.jedis.Module; import redis.clients.jedis.commands.ProtocolCommand; import redis.clients.jedis.exceptions.JedisAskDataException; import redis.clients.jedis.exceptions.JedisDataException; import redis.clients.jedis.exceptions.JedisMovedDataException; import redis.clients.jedis.util.SafeEncoder; import redis.clients.jedis.util.Slowlog; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.sql.Timestamp; import java.text.DecimalFormat; import java.util.*; import java.util.Map.Entry; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; /** * Created by yijunzhang on 14-6-10. */ @Service("redisCenter") public class RedisCenterImpl implements RedisCenter { public static final int REDIS_DEFAULT_TIME = 1000; public static final String REDIS_SLOWLOG_POOL = "redis-slowlog-pool"; private static final int COUNT = 1000; private static List<RedisInfoEnum> otherNeedCalDifRedisInfoEnumList = new ArrayList<>(); private final Logger logger = LoggerFactory.getLogger(this.getClass()); private final Lock lock = new ReentrantLock(); @Autowired private AppStatsDao appStatsDao; @Autowired private AsyncService asyncService; @Autowired private InstanceDao instanceDao; @Autowired private InstanceStatsDao instanceStatsDao; @Autowired private InstanceStatsCenter instanceStatsCenter; @Autowired @Lazy private MachineCenter machineCenter; private volatile Map<String, JedisPool> jedisPoolMap = new HashMap<String, JedisPool>(); @Autowired private AppDao appDao; @Autowired private RedisModuleConfigDao redisModuleConfigDao; @Autowired private AppAuditLogDao appAuditLogDao; @Autowired private AppService appService; @Autowired private InstanceSlowLogDao instanceSlowLogDao; @Autowired private InstanceLatencyHistoryDao instanceLatencyHistoryDao; @Autowired private WebClientComponent webClientComponent; @Autowired SSHService sshService; @Autowired private ModuleService moduleService; @Autowired private ReportDataComponent reportDataComponent; @PostConstruct public void init() { asyncService.assemblePool(getThreadPoolKey(), AsyncThreadPoolFactory.REDIS_SLOWLOG_THREAD_POOL); otherNeedCalDifRedisInfoEnumList.add(RedisInfoEnum.mem_fragmentation_ratio); } private JedisPool maintainJedisPool(String host, int port, String password) { String hostAndPort = ObjectConvert.linkIpAndPort(host, port); JedisPool jedisPool = jedisPoolMap.get(hostAndPort); if (jedisPool == null) { lock.lock(); try { //double check jedisPool = jedisPoolMap.get(hostAndPort); if (jedisPool == null) { try { if (StringUtils.isNotBlank(password)) { jedisPool = new JedisPool(new GenericObjectPoolConfig(), host, port, Protocol.DEFAULT_TIMEOUT, password); } else { jedisPool = new JedisPool(new GenericObjectPoolConfig(), host, port, Protocol.DEFAULT_TIMEOUT); } jedisPoolMap.put(hostAndPort, jedisPool); } catch (Exception e) { logger.error(e.getMessage(), e); } finally { } } } finally { lock.unlock(); } } return jedisPool; } private String buildFutureKey(long appId, long collectTime, String host, int port) { StringBuilder keyBuffer = new StringBuilder("redis-"); keyBuffer.append(collectTime); keyBuffer.append("-"); keyBuffer.append(appId); keyBuffer.append("-"); keyBuffer.append(host + ":" + port); return keyBuffer.toString(); } @Override public List<InstanceSlowLog> collectRedisSlowLog(long appId, long collectTime, String host, int port) { Assert.isTrue(appId > 0); Assert.hasText(host); Assert.isTrue(port > 0); InstanceInfo instanceInfo = instanceDao.getInstByIpAndPort(host, port); //不存在实例/实例异常/下线 if (instanceInfo == null) { return null; } if (TypeUtil.isRedisSentinel(instanceInfo.getType())) { //忽略sentinel redis实例 return null; } // 从redis中获取慢查询日志 List<RedisSlowLog> redisLowLogList = getRedisSlowLogs(appId, host, port, 100); if (CollectionUtils.isEmpty(redisLowLogList)) { return Collections.emptyList(); } // transfer final List<InstanceSlowLog> instanceSlowLogList = new ArrayList<InstanceSlowLog>(); for (RedisSlowLog redisSlowLog : redisLowLogList) { InstanceSlowLog instanceSlowLog = transferRedisSlowLogToInstance(redisSlowLog, instanceInfo); if (instanceSlowLog == null) { continue; } instanceSlowLogList.add(instanceSlowLog); } if (CollectionUtils.isEmpty(instanceSlowLogList)) { return Collections.emptyList(); } //处理 String key = getThreadPoolKey() + "_" + host + "_" + port; boolean isOk = asyncService.submitFuture(getThrea