Redis生产事件总结报告

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

private void initializeSlotsCache(Set<HostAndPort> startNodes, GenericObjectPoolConfig poolConfig, String password) {
    for (HostAndPort hostAndPort : startNodes) {
      Jedis jedis = new Jedis(hostAndPort.getHost(), hostAndPort.getPort());
      if (password != null) {
        jedis.auth(password);
      }
      try {
        cache.discoverClusterNodesAndSlots(jedis);
        break;
      } catch (JedisConnectionException e) {
        // try next nodes
      } finally {
        if (jedis != null) {
          jedis.close();
        }
      }
    }
  }

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @author huqinuo
 * @version 1.0
 * @description Redis节点上下文管理器
 * @data 2025/12/1 9:25
 */
@Component
public class RedisNodeContext {

    private final Logger log = LoggerFactory.getLogger(RedisNodeContext.class);

    private final Set<String> unhealthyNodes = new CopyOnWriteArraySet<>();

    private final Set<String> allNodes = new CopyOnWriteArraySet<>();

    /**
     * 初始化方法，从静态数据加载
     */
    public void initializeFromStaticData() {
        if (RedisHealthChecker.isHealthCheckCompleted()) {
            this.allNodes.clear();
            this.allNodes.addAll(RedisHealthChecker.getAllNodes());

            this.unhealthyNodes.clear();
            this.unhealthyNodes.addAll(RedisHealthChecker.getUnhealthyNodes());

            log.info("RedisNodeContext initialized from static data: {}/{} nodes healthy",
                    getHealthyNodeCount(), allNodes.size());
        }
    }

    /**
     * 添加不健康节点
     */
    public void addUnhealthyNode(String node) {
        unhealthyNodes.add(node);
    }

    /**
     * 移除不健康节点（节点恢复时调用）
     */
    public void removeUnhealthyNode(String node) {
        unhealthyNodes.remove(node);
    }

    /**
     * 获取所有不健康节点
     */
    public Set<String> getUnhealthyNodes() {
        return new HashSet<>(unhealthyNodes);
    }

    /**
     * 检查节点是否不健康
     */
    public boolean isNodeUnhealthy(String node) {
        return unhealthyNodes.contains(node);
    }

    /**
     * 设置所有节点
     */
    public void setAllNodes(Set<String> nodes) {
        allNodes.clear();
        allNodes.addAll(nodes);
    }

    /**
     * 获取所有节点
     */
    public Set<String> getAllNodes() {
        return new HashSet<>(allNodes);
    }

    /**
     * 添加单个节点
     */
    public void addNode(String node) {
        allNodes.add(node);
    }

    /**
     * 获取健康节点数量
     */
    public int getHealthyNodeCount() {
        return allNodes.size() - unhealthyNodes.size();
    }

    /**
     * 获取不健康节点数量
     */
    public int getUnhealthyNodeCount() {
        return unhealthyNodes.size();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

package open.ref.redis;

import 自研包.RedisFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.scheduling.annotation.EnableScheduling;

/**
 * @author huqinuo
 * @version 1.0
 * @description Redis配置类(健康检查使用)
 * @data 2025/12/1 15:31
 */
@Configuration
@EnableScheduling
public class RedisClusterConfig {

    /**
     * redis节点上下文
     */
    @Bean
    @DependsOn("redisHealthChecker")  // 确保在健康检查后执行
    public RedisNodeContext redisNodeContext() {
        RedisNodeContext context = new RedisNodeContext();
        context.initializeFromStaticData(); // 从静态数据初始化
        return context;
    }

    /**
     * 拓扑刷新器
     */
    @Bean
    @DependsOn("redisNodeContext")
    public RedisClusterTopologyRefresher redisClusterTopologyRefresher(@Autowired RedisNodeContext redisNodeContext, RedisFactory redisFactory) {
        return new RedisClusterTopologyRefresher(redisNodeContext,redisFactory);
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209

package open.ref.redis;

import open.svc.dict.Const;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.Environment;
import org.springframework.core.env.MapPropertySource;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;

import java.util.*;

/**
 * @author huqinuo
 * @version 1.0
 * @description Redis集群连接健康检查
 * @data 2025/12/1 10:02
 */
@Component
public class RedisHealthChecker implements BeanFactoryPostProcessor, EnvironmentAware {

    private ConfigurableEnvironment environment;

    private final Logger log = LoggerFactory.getLogger(RedisHealthChecker.class);

    private static final Set<String> unhealthyNodes = new HashSet<>();

    private static final Set<String> allNodes = new HashSet<>();

    private static final int CONNECTION_TIMEOUT = 5000;

    private static boolean healthCheckCompleted = false;

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = (ConfigurableEnvironment) environment;
    }

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        if (healthCheckCompleted) {
            return; // 避免重复执行
        }

        if (!isRedisClusterEnabled()) {
            log.info("Redis cluster is not enabled, skip health check");
            healthCheckCompleted = true;
            return;
        }

        String oriRedisNodeUrl = getOriRedisNodeUrl();
        if (Objects.isNull(oriRedisNodeUrl)) {
            log.warn("Application does not have Redis configuration");
            healthCheckCompleted = true;
            return;
        }

        // 执行健康检查并更新配置
        performInitialHealthCheck(oriRedisNodeUrl);
        healthCheckCompleted = true;

        // 注册静态上下文到BeanFactory
        registerNodeContextBean(beanFactory);
    }

    /**
     * 注册节点上下文到BeanFactory
     */
    private void registerNodeContextBean(ConfigurableListableBeanFactory beanFactory) {
        RedisNodeContext nodeContext = new RedisNodeContext();
        nodeContext.setAllNodes(new HashSet<>(allNodes));
        for (String node : unhealthyNodes) {
            nodeContext.addUnhealthyNode(node);
        }
        beanFactory.registerSingleton("redisNodeContext", nodeContext);
        log.info("RedisNodeContext registered as singleton bean");
    }

    /**
     * 执行初始健康检查
     */
    private void performInitialHealthCheck(String oriRedisNodeUrl) {
        // 保存所有原始节点
        allNodes.clear();
        allNodes.addAll(Arrays.asList(oriRedisNodeUrl.split(Const.Symbol.Comma)));

        log.info("Starting initial Redis cluster health check for {} nodes", allNodes.size());

        // 过滤健康节点
        List<String> healthyNodes = filterHealthyClusterNodes(oriRedisNodeUrl);

        if (healthyNodes.isEmpty()) {
            log.info("No healthy Redis cluster nodes available");
            return;
        }

        String healthyRedisNodeUrl = String.join(Const.Symbol.Comma, healthyNodes);

        // 如果有不健康节点，更新配置
        if (!oriRedisNodeUrl.equals(healthyRedisNodeUrl)) {
            updateRedisConfiguration(healthyRedisNodeUrl);
            log.warn("Redis configuration updated due to unhealthy nodes. Healthy nodes: {}", healthyRedisNodeUrl);
        }

        // 记录初始健康状态
        log.info("Initial Redis cluster health status: {}/{} nodes healthy",
                (allNodes.size() - unhealthyNodes.size()), allNodes.size());
        log.info("Unhealthy nodes: {}", unhealthyNodes);
    }

    private Boolean isRedisClusterEnabled() {
        String isEnabled = environment.getProperty("自研包.cluster.enabled");
        if (isEnabled != null) {
            return Boolean.parseBoolean(isEnabled);
        }
        return Boolean.FALSE;
    }

    private String getOriRedisNodeUrl() {
        // 提取集群配置
        String clusterNodes = environment.getProperty("spring.redis.cluster.nodes");
        log.info("Redis cluster nodes configuration: {}", clusterNodes);
        if (clusterNodes != null) {
            return clusterNodes;
        }
        return null;
    }

    private String getAuthPassword() {
        // 提取认证配置
        String auth = environment.getProperty("spring.redis.password");
        if (auth != null) {
            return auth;
        }
        return null;
    }

    private List<String> filterHealthyClusterNodes(String oriRedisNodeUrl) {
        List<String> healthyNodes = new ArrayList<>();
        String[] clusterNodes = oriRedisNodeUrl.split(Const.Symbol.Comma);

        for (String node : clusterNodes) {
            if (checkRedisNode(node)) {
                healthyNodes.add(node);
                log.info("Redis cluster node {} is healthy", node);
            } else {
                // 节点不健康，记录到静态集合
                unhealthyNodes.add(node);
                log.warn("Redis cluster node {} is unhealthy, added to monitoring list", node);
            }
        }

        return healthyNodes;
    }

    private boolean checkRedisNode(String node) {
        String[] parts = node.split(":");
        String host = parts[0];
        int port = parts.length > 1 ? Integer.parseInt(parts[1]) : 6379;
        try (Jedis jedis = new Jedis(host, port, CONNECTION_TIMEOUT)) {
            String auth = getAuthPassword();
            if (!Objects.isNull(auth)) {
                jedis.auth(auth);
            }
            String result = jedis.ping();
            return "PONG".equals(result);
        } catch (Exception e) {
            log.warn("Redis node {}:{} check failed: {}", host, port, e.getMessage());
            return false;
        }
    }

    private void updateRedisConfiguration(String healthyRedisNodeUrl) {
        Map<String, Object> updatedProperties = new HashMap<>();

        // 更新集群配置
        if (healthyRedisNodeUrl != null) {
            updatedProperties.put("athena.redis.url", healthyRedisNodeUrl);
            updatedProperties.put("spring.redis.cluster.nodes", healthyRedisNodeUrl);
        }

        // 将更新后的配置添加到Environment的最高优先级
        environment.getPropertySources().addFirst(
                new MapPropertySource("redisHealthProperties", updatedProperties)
        );

        log.info("Redis configuration has been updated with healthy nodes: {}", healthyRedisNodeUrl);
    }

    /**
     * 静态方法供其他组件访问
     */
    public static Set<String> getUnhealthyNodes() {
        return new HashSet<>(unhealthyNodes);
    }

    public static Set<String> getAllNodes() {
        return new HashSet<>(allNodes);
    }

    public static boolean isHealthCheckCompleted() {
        return healthCheckCompleted;
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282

package open.ref.redis;

import 自研包.JedisFactoryClusterImpl;
import 自研包.RedisFactory;
import open.svc.dict.Const;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import redis.clients.jedis.BinaryJedisCluster;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisSlotBasedConnectionHandler;

import java.lang.reflect.Field;
import java.util.*;

/**
 * @author huqinuo
 * @version 1.0
 * @description Redis集群拓扑刷新器
 * @data 2025/12/1 13:41
 */
@Component
public class RedisClusterTopologyRefresher {

    private final Logger log = LoggerFactory.getLogger(RedisClusterTopologyRefresher.class);

    private final RedisNodeContext redisNodeContext;

    private final RedisFactory redisFactory;

    @Value("${自研包密码}")
    private String auth;

    /**
     * 2分钟刷新一次
     */
    private static final long REFRESH_INTERVAL = 2 * 60 * 1000;

    /**
     * 连接超时时间
     */
    private static final int CONNECTION_TIMEOUT = 5000;

    /**
     * 是否刷新中
     */
    private volatile boolean isRefreshing = false;


    @Autowired
    public RedisClusterTopologyRefresher(RedisNodeContext redisNodeContext, RedisFactory redisFactory) {
        this.redisNodeContext = redisNodeContext;
        this.redisFactory = redisFactory;
        log.info("RedisClusterTopologyRefresher initialized with {} unhealthy nodes",
                redisNodeContext.getUnhealthyNodeCount());
    }

    /**
     * 定时刷新不健康节点的拓扑信息
     */
    @Scheduled(fixedRate = REFRESH_INTERVAL)
    public void refreshTopologyForUnhealthyNodes() {
        if (isRefreshing) {
            log.info("Topology refresh is already in progress, skip this execution");
            return;
        }

        //从单个节点获取集群信息，存在disconnected连接状态节点则为不健康
        checkAllNodeIfExistUnhealthyNodes();

        Set<String> unhealthyNodes = redisNodeContext.getUnhealthyNodes();

        if (unhealthyNodes.isEmpty()) {
            log.info("No unhealthy nodes to refresh");
            return;
        }

        log.info("Starting topology refresh for {} unhealthy nodes: {}",
                unhealthyNodes.size(), String.join(",", unhealthyNodes));

        isRefreshing = true;
        try {
            Set<String> recoveredNodes = new HashSet<>();

            for (String node : unhealthyNodes) {
                if (checkAndRefreshNode(node)) {
                    recoveredNodes.add(node);
                }
            }

            // 处理恢复的节点
            if (!recoveredNodes.isEmpty()) {
                handleRecoveredNodes(recoveredNodes);
            }
            log.info("Topology refresh completed. Recovered {} nodes, Redis cluster refresh status:{}", recoveredNodes.size(), getRefreshStatus());
        } catch (Exception e) {
            log.error("Error during topology refresh", e);
        } finally {
            isRefreshing = false;
        }
    }

    private void checkAllNodeIfExistUnhealthyNodes() {
        Set<String> allNodes = redisNodeContext.getAllNodes();
        String oneNode = allNodes.iterator().next();
        String[] parts = oneNode.split(":");
        String host = parts[0];
        int port = parts.length > 1 ? Integer.parseInt(parts[1]) : 6379;
        try (Jedis jedis = new Jedis(host, port, CONNECTION_TIMEOUT)) {
            jedis.auth(auth);
            String clusterNodes = jedis.clusterNodes();
            log.info("Redis cluster nodes : {}",clusterNodes);
            String[] lines = clusterNodes.split(Const.Symbol.BlankLine);
            for(String line:lines){
                if(line.trim().isEmpty()){
                    continue;
                }

                String[] lineParts = line.split("\\s+");
                if(lineParts.length<8){
                    continue;
                }
                String linkState = lineParts[7];
                if("disconnected".equals(linkState)){
                    String ipPortPart = lineParts[1];
                    String node = ipPortPart.split("@")[0];
                    log.info("Find unheanthy Redis node : {}",node);
                    redisNodeContext.addUnhealthyNode(node);
                }
            }
        } catch (Exception e) {
            log.error("Exist unhealthy Redis node", e);
            redisNodeContext.addUnhealthyNode(oneNode);
        }
    }

    /**
     * 检查并刷新单个节点
     *
     * @param node
     * @return
     */
    private boolean checkAndRefreshNode(String node) {
        String[] parts = node.split(":");
        String host = parts[0];
        int port = parts.length > 1 ? Integer.parseInt(parts[1]) : 6379;

        log.info("Checking node {}:{} for recovery", host, port);

        // 首先检查节点是否恢复
        if (!checkNodeAvailability(host, port)) {
            log.info("Node {}:{} is still unavailable", host, port);
            return false;
        }

        log.info("Node {}:{} is available again, refreshing topology", host, port);

        // 尝试刷新拓扑
        boolean refreshSuccess = refreshNodeTopology(host, port);

        if (refreshSuccess) {
            log.info("Successfully refreshed topology for recovered node: {}:{}", host, port);
            return true;
        } else {
            log.warn("Node {}:{} is available but topology refresh failed", host, port);
            return false;
        }
    }

    /**
     * 检查节点是否可用
     *
     * @param host
     * @param port
     * @return
     */
    private boolean checkNodeAvailability(String host, int port) {
        try (Jedis jedis = new Jedis(host, port, CONNECTION_TIMEOUT)) {
            jedis.auth(auth);
            String result = jedis.ping();
            return "PONG".equals(result);
        } catch (Exception e) {
            log.warn("Node {}:{} availability check failed: {}", host, port, e.getMessage());
            return false;
        }
    }

    /**
     * 刷新节点拓扑
     *
     * @param host
     * @param port
     * @return
     */
    private boolean refreshNodeTopology(String host, int port) {
        try (Jedis jedis = new Jedis(host, port, CONNECTION_TIMEOUT)) {
            jedis.auth(auth);
            // 验证节点确实是集群节点
            String clusterInfo = jedis.clusterInfo();
            if (!clusterInfo.contains("cluster_state:ok")) {
                log.warn("Node {}:{} cluster state is not ok", host, port);
                return false;
            }

            // 获取集群槽位信息
            List<Object> slotInfo = jedis.clusterSlots();
            if (slotInfo == null || slotInfo.isEmpty()) {
                log.warn("Node {}:{} returned empty cluster slots", host, port);
                return false;
            }

            log.info("Node {}:{} cluster slots info retrieved successfully", host, port);
            return topologyRefresh(jedis);

        } catch (Exception e) {
            log.error("Failed to refresh topology for node {}:{}", host, port, e);
            return false;
        }
    }

    /**
     * 拓扑刷新
     *
     * @param jedis
     * @return
     */
    private boolean topologyRefresh(Jedis jedis) {
        try {
            if (redisFactory instanceof JedisFactoryClusterImpl) {
                Object obj = redisFactory.getObject();
                JedisCluster jedisCluster = (JedisCluster) obj;
                Field connectionHandlerField = BinaryJedisCluster.class.getDeclaredField("connectionHandler");
                connectionHandlerField.setAccessible(true);
                Object connectionHandler = connectionHandlerField.get(jedisCluster);
                if (connectionHandler instanceof JedisSlotBasedConnectionHandler) {
                    JedisSlotBasedConnectionHandler handler = (JedisSlotBasedConnectionHandler) connectionHandler;
                    handler.renewSlotCache(jedis);
                }
            }
            log.info("Topology refresh triggered for recovered nodes");
            return true;
        } catch (Exception e) {
            log.error("Failed to trigger topology refresh", e);
            return false;
        }
    }

    /**
     * 处理恢复的节点
     *
     * @param recoveredNodes
     */
    private void handleRecoveredNodes(Set<String> recoveredNodes) {
        // 从上下文中移除恢复的节点
        for (String node : recoveredNodes) {
            redisNodeContext.removeUnhealthyNode(node);
        }

        log.info("{} nodes recovered and removed from unhealthy list: {}",
                recoveredNodes.size(), String.join(",", recoveredNodes));
    }

    /**
     * 获取刷新状态报告
     *
     * @return
     */
    public Map<String, Object> getRefreshStatus() {
        Map<String, Object> status = new HashMap<>();
        status.put("unhealthyNodes", redisNodeContext.getUnhealthyNodes());
        status.put("unhealthyNodeCount", redisNodeContext.getUnhealthyNodeCount());
        status.put("isRefreshing", isRefreshing);
        status.put("lastRefreshTime", new Date());
        status.put("refreshInterval", REFRESH_INTERVAL);

        return status;
    }
}