高并发架构设计原则与实践

1
// 示例：使用Spring Session + Redis实现分布式会话
2
@Configuration
3
@EnableRedisHttpSession(maxInactiveIntervalInSeconds = 1800) // 30分钟会话过期
4
public class HttpSessionConfig {
5

6
    @Value("${spring.redis.host:localhost}")
7
    private String redisHost;
8

9
    @Value("${spring.redis.port:6379}")
10
    private int redisPort;
11

12
    @Value("${spring.redis.password:}")
13
    private String redisPassword;
14

15
    @Value("${spring.redis.database:0}")
16
    private int database;
17

18
    @Bean
19
    public RedisConnectionFactory connectionFactory() {
20
        RedisStandaloneConfiguration config = new RedisStandaloneConfiguration();
21
        config.setHostName(redisHost);
22
        config.setPort(redisPort);
23
        if (StringUtils.hasText(redisPassword)) {
24
            config.setPassword(redisPassword);
25
        }
26
        config.setDatabase(database);
27

28
        // 使用Lettuce客户端
29
        LettuceClientConfiguration clientConfig = LettuceClientConfiguration.builder()
30
            .commandTimeout(Duration.ofSeconds(2))
31
            .shutdownTimeout(Duration.ZERO)
32
            .build();
33

34
        return new LettuceConnectionFactory(config, clientConfig);
35
    }
36

37
    @Bean
38
    public RedisTemplate<String, Object> redisTemplate() {
39
        RedisTemplate<String, Object> template = new RedisTemplate<>();
40
        template.setConnectionFactory(connectionFactory());
41

42
        // 使用StringRedisSerializer来序列化和反序列化redis的key值
43
        template.setKeySerializer(new StringRedisSerializer());
44
        template.setHashKeySerializer(new StringRedisSerializer());
45

46
        // 使用GenericJackson2JsonRedisSerializer来序列化和反序列化redis的value值
47
        template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
48
        template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
49

50
        template.afterPropertiesSet();
51
        return template;
52
    }
53
}

1
spring:
2
  redis:
3
    host: ${REDIS_HOST:localhost}
4
    port: ${REDIS_PORT:6379}
5
    password: ${REDIS_PASSWORD:}
6
    database: 0
7
    timeout: 2000ms
8
    lettuce:
9
      pool:
10
        max-active: 8
11
        max-wait: -1ms
12
        max-idle: 8
13
        min-idle: 0

1
-- 示例：MySQL分表设计
2
CREATE TABLE `order_2023` (
3
    `id` bigint(20) NOT NULL AUTO_INCREMENT,
4
    `order_no` varchar(32) NOT NULL,
5
    `user_id` bigint(20) NOT NULL,
6
    `amount` decimal(10,2) NOT NULL,
7
    `create_time` datetime NOT NULL,
8
    PRIMARY KEY (`id`),
9
    KEY `idx_user_id` (`user_id`),
10
    UNIQUE KEY `uk_order_no` (`order_no`)
11
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

1
// 生产者配置
2
@Configuration
3
@EnableKafka
4
public class KafkaProducerConfig {
5

6
    @Value("${spring.kafka.bootstrap-servers}")
7
    private String bootstrapServers;
8

9
    @Value("${spring.kafka.producer.retries:3}")
10
    private int retries;
11

12
    @Value("${spring.kafka.producer.ack:all}")
13
    private String acks;
14

15
    @Value("${spring.kafka.producer.batch-size:16384}")
16
    private int batchSize;
17

18
    @Value("${spring.kafka.producer.buffer-memory:33554432}")
19
    private long bufferMemory;
20

21
    @Value("${spring.kafka.producer.linger-ms:100}")
22
    private long lingerMs;
23

24
    @Value("${spring.kafka.producer.compression-type:gzip}")
25
    private String compressionType;
26

27
    @Bean
28
    public ProducerFactory<String, String> producerFactory() {
29
        Map<String, Object> configProps = new HashMap<>();
30
        // 基础配置
31
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
32
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
33
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
34

35
        // 可靠性配置
36
        configProps.put(ProducerConfig.ACKS_CONFIG, acks);
37
        configProps.put(ProducerConfig.RETRIES_CONFIG, retries);
38
        configProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
39

40
        // 性能调优
41
        configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
42
        configProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
43
        configProps.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs);
44
        configProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType);
45

46
        // 安全配置（如果需要）
47
        if (bootstrapServers.contains("SASL")) {
48
            configProps.put("security.protocol", "SASL_SSL");
49
            configProps.put("sasl.mechanism", "PLAIN");
50
        }
51

52
        return new DefaultKafkaProducerFactory<>(configProps);
53
    }
54

55
    @Bean
56
    public KafkaTemplate<String, String> kafkaTemplate() {
57
        KafkaTemplate<String, String> template = new KafkaTemplate<>(producerFactory());
58

59
        // 设置默认主题（可选）
60
        template.setDefaultTopic("default-topic");
61

62
        // 设置生产者监听器
63
        template.setProducerListener(new ProducerListener<String, String>() {
64
            private final Logger log = LoggerFactory.getLogger(ProducerListener.class);
65

66
            @Override
67
            public void onSuccess(ProducerRecord<String, String> record, RecordMetadata metadata) {
68
                log.info("Message sent successfully - Topic: {}, Partition: {}, Offset: {}, Value: {}",
69
                        metadata.topic(),
70
                        metadata.partition(),
71
                        metadata.offset(),
72
                        record.value());
73
            }
74

75
            @Override
76
            public void onError(ProducerRecord<String, String> record, Exception exception) {
77
                log.error("Failed to send message - Topic: {}, Key: {}, Value: {}",
78
                        record.topic(),
79
                        record.key(),
80
                        record.value(),
81
                        exception);
82
                // 这里可以添加重试或告警逻辑
83
            }
84
        });
85

86
        // 设置事务ID前缀（如果启用事务）
87
        template.setTransactionIdPrefix("tx-");
88

89
        return template;
90
    }
91

92
    // 批量消息支持
93
    @Bean
94
    public BatchMessagingTemplate batchMessagingTemplate() {
95
        return new BatchMessagingTemplate(kafkaTemplate());
96
    }

1
@Service
2
@Slf4j
3
@RequiredArgsConstructor
4
public class OrderService {
5

6
    @Value("${kafka.topic.order:orders}")
7
    private String orderTopic;
8

9
    private final KafkaTemplate<String, String> kafkaTemplate;
10
    private final ObjectMapper objectMapper = new ObjectMapper();
11

12
    public void createOrder(Order order) {
13
        try {
14
            // 发送消息
15
            ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(
16
                orderTopic,
17
                order.getOrderId(),
18
                objectMapper.writeValueAsString(order)
19
            );
20

21
            // 异步回调
22
            future.addCallback(
23
                result -> log.info("Order created: {}", order.getOrderId()),
24
                ex -> log.error("Failed to create order: {}", order.getOrderId(), ex)
25
            );
26

27
        } catch (JsonProcessingException e) {
28
            log.error("Failed to serialize order: {}", order.getOrderId(), e);
29
            throw new RuntimeException("Failed to create order", e);
30
        }
31
    }
32

33
    // 批量发送消息
34
    public void batchCreateOrders(List<Order> orders) {
35
        List<CompletableFuture<SendResult<String, String>>> futures = orders.stream()
36
            .map(order -> {
37
                try {
38
                    return kafkaTemplate.send(
39
                        orderTopic,
40
                        order.getOrderId(),
41
                        objectMapper.writeValueAsString(order)
42
                    ).completable()
43
                     .whenComplete((result, ex) -> {
44
                         if (ex != null) {
45
                             log.error("Failed to send order: {}", order.getOrderId(), ex);
46
                         } else {
47
                             log.info("Order sent: {}", order.getOrderId());
48
                         }
49
                     });
50
                } catch (JsonProcessingException e) {
51
                    log.error("Failed to serialize order: {}", order.getOrderId(), e);
52
                    return CompletableFuture.<SendResult<String, String>>failedFuture(e);
53
                }
54
            })
55
            .collect(Collectors.toList());
56

57
        // 等待所有消息发送完成
58
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
59
    }
60
    }
61
}

1
spring:
2
  kafka:
3
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
4
    producer:
5
      retries: 3
6
      acks: all
7
      batch-size: 16384
8
      buffer-memory: 33554432
9
      linger-ms: 100
10
      compression-type: gzip
11
      properties:
12
        max.request.size: 1048576
13
        request.timeout.ms: 30000
14
        delivery.timeout.ms: 120000
15
        max.block.ms: 60000
16
    consumer:
17
      group-id: ${spring.application.name:myapp}-group
18
      auto-offset-reset: latest
19
      enable-auto-commit: false
20
      max-poll-records: 500
21
      fetch-max-wait: 500
22
      fetch-min-size: 1
23
      heartbeat-interval: 3000
24
      session-timeout: 10000
25
      max-poll-interval: 300000
26
      properties:
27
        max.partition.fetch.bytes: 1048576
28
        request.timeout.ms: 30000
29
        session.timeout.ms: 10000
30
        max.poll.records: 500

1
@Slf4j
2
@Component
3
public class OrderConsumer {
4

5
    @Value("${kafka.topic.order:orders}")
6
    private String orderTopic;
7
    private final ObjectMapper objectMapper;
8
    private final OrderService orderService;
9

10
    public OrderConsumer(ObjectMapper objectMapper, OrderService orderService) {
11
        this.objectMapper = objectMapper;
12
        this.orderService = orderService;
13
    }
14

15
    @KafkaListener(
16
        topics = "${kafka.topic.order:orders}",
17
        groupId = "${spring.kafka.consumer.group-id:order-service-group}",
18
        containerFactory = "kafkaListenerContainerFactory"
19
    )
20
    public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
21
        try {
22
            // 反序列化消息
23
            Order order = objectMapper.readValue(record.value(), Order.class);
24
            log.info("Processing order: {}", order.getOrderId());
25

26
            // 业务处理
27
            orderService.processOrder(order);
28

29
            // 手动提交偏移量
30
            ack.acknowledge();
31
            log.info("Successfully processed order: {}", order.getOrderId());
32

33
        } catch (Exception e) {
34
            log.error("Error processing order: {}", record.value(), e);
35
            // 可以根据异常类型决定是否重试
36
            if (shouldRetry(e)) {
37
                throw new RuntimeException("Retryable error processing order", e);
38
            }
39
        }
40
    }
41

42
    private boolean shouldRetry(Exception e) {
43
        // 实现重试逻辑
44
        return e instanceof NetworkException || e instanceof TimeoutException;
45
    }
46

47
    // 批量消费示例
48
    @KafkaListener(
49
        topics = "${kafka.topic.batch-orders:batch-orders}",
50
        containerFactory = "batchKafkaListenerContainerFactory"
51
    )
52
    public void batchConsume(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
53
        log.info("Received batch with {} records", records.size());
54

55
        try {
56
            List<Order> orders = records.stream()
57
                .map(record -> {
58
                    try {
59
                        return objectMapper.readValue(record.value(), Order.class);
60
                    } catch (JsonProcessingException e) {
61
                        log.error("Failed to deserialize order: {}", record.value(), e);
62
                        return null;
63
                    }
64
                })
65
                .filter(Objects::nonNull)
66
                .collect(Collectors.toList());
67

68
            // 批量处理订单
69
            orderService.batchProcessOrders(orders);
70

71
            // 手动提交偏移量
72
            ack.acknowledge();
73
            log.info("Successfully processed batch of {} orders", orders.size());
74

75
        } catch (Exception e) {
76
            log.error("Error processing batch of {} records", records.size(), e);
77
            throw new RuntimeException("Failed to process batch", e);
78
        }
79
    }
80
}
81

82
// 消费者配置类
83
@Configuration
84
@EnableKafka
85
public class KafkaConsumerConfig {
86

87
    @Value("${spring.kafka.bootstrap-servers}")
88
    private String bootstrapServers;
89

90
    @Value("${spring.kafka.consumer.group-id}")
91
    private String groupId;
92

93
    @Value("${spring.kafka.consumer.auto-offset-reset:latest}")
94
    private String autoOffsetReset;
95

96
    @Value("${spring.kafka.consumer.enable-auto-commit:false}")
97
    private boolean enableAutoCommit;
98

99
    @Value("${spring.kafka.consumer.max-poll-records:500}")
100
    private int maxPollRecords;
101

102
    @Value("${spring.kafka.consumer.concurrency:3}")
103
    private int concurrency;
104

105
    @Bean
106
    public ConsumerFactory<String, String> consumerFactory() {
107
        Map<String, Object> props = new HashMap<>();
108
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
109
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
110
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
111
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
112
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
113
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
114
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
115

116
        // 安全配置（如果需要）
117
        if (bootstrapServers.contains("SASL")) {
118
            props.put("security.protocol", "SASL_SSL");
119
            props.put("sasl.mechanism", "PLAIN");
120
        }
121

122
        return new DefaultKafkaConsumerFactory<>(props);
123
    }
124

125
    @Bean
126
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
127
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
128
        factory.setConsumerFactory(consumerFactory());
129

130
        // 设置并发消费者数量
131
        factory.setConcurrency(concurrency);
132

133
        // 设置手动提交
134
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
135

136
        // 设置批量消费
137
        factory.setBatchListener(false);
138

139
        // 设置异常处理器
140
        factory.setErrorHandler(new SeekToCurrentErrorHandler(
141
            new FixedBackOff(1000L, 3) // 重试3次，每次间隔1秒
142
        ));
143

144
        return factory;
145
    }
146

147
    @Bean
148
    public ConcurrentKafkaListenerContainerFactory<String, String> batchKafkaListenerContainerFactory() {
149
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
150
        factory.setConsumerFactory(consumerFactory());
151

152
        // 启用批量消费
153
        factory.setBatchListener(true);
154

155
        // 设置并发消费者数量
156
        factory.setConcurrency(concurrency);
157

158
        // 设置手动提交
159
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
160

161
        // 设置批量消费属性
162
        factory.getContainerProperties().setIdleEventInterval(60000L);
163
        factory.getContainerProperties().setPollTimeout(3000);
164

165
        return factory;
166
    }
167

168
    // 死信队列配置
169
    @Bean
170
    public DeadLetterPublishingRecoverer deadLetterPublishingRecoverer(KafkaTemplate<?, ?> template) {
171
        return new DeadLetterPublishingRecoverer(template,
172
            (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()));
173
    }
174

175
    @Bean
176
    public SeekToCurrentErrorHandler errorHandler(DeadLetterPublishingRecoverer recoverer) {
177
        return new SeekToCurrentErrorHandler(recoverer, new FixedBackOff(1000L, 3));
178
    }
179
}

1
<dependency>
2
    <groupId>io.github.resilience4j</groupId>
3
    <artifactId>resilience4j-spring-boot2</artifactId>
4
    <version>1.7.1</version>
5
</dependency>
6
<dependency>
7
    <groupId>org.springframework.boot</groupId>
8
    <artifactId>spring-boot-starter-aop</artifactId>
9
</dependency>

1
resilience4j:
2
  circuitbreaker:
3
    configs:
4
      default:  # 默认配置
5
        slidingWindowType: COUNT_BASED  # 滑动窗口类型: COUNT_BASED(基于调用次数)或TIME_BASED(基于时间)
6
        slidingWindowSize: 10           # 滑动窗口大小(次数或秒数)
7
        minimumNumberOfCalls: 5         # 计算失败率前的最小调用次数
8
        permittedNumberOfCallsInHalfOpenState: 3  # 半开状态下允许的调用次数
9
        automaticTransitionFromOpenToHalfOpenEnabled: true  # 自动从开启状态转为半开状态
10
        waitDurationInOpenState: 5s     # 开启状态转为半开状态的等待时间
11
        failureRateThreshold: 50         # 失败率阈值百分比，超过则触发熔断
12
        slowCallRateThreshold: 100       # 慢调用率阈值百分比
13
        slowCallDurationThreshold: 2s    # 慢调用时间阈值
14
        recordExceptions:                # 记录为失败的异常类型
15
          - org.springframework.web.client.HttpServerErrorException
16
          - java.io.IOException
17
          - java.util.concurrent.TimeoutException
18
          - org.springframework.web.client.ResourceAccessException
19
        ignoreExceptions:                # 忽略的异常类型(不计入失败统计)
20
          - com.example.BusinessException
21
    instances:
22
      paymentService:  # 支付服务专用配置
23
        baseConfig: default
24
        failureRateThreshold: 30  # 支付服务更敏感，使用更低的失败率阈值
25
      externalApiService:  # 外部API服务配置
26
        baseConfig: default
27
        slidingWindowSize: 20
28
        minimumNumberOfCalls: 10
29
        waitDurationInOpenState: 10s
30

31
# 健康检查端点配置
32
management:
33
  endpoint:
34
    health:
35
      show-details: always
36
      circuitbreakers:
37
        enabled: true
38
  endpoints:
39
    web:
40
      exposure:
41
        include: health,metrics,circuitbreakers

1
@Service
2
@RequiredArgsConstructor
3
public class PaymentService {
4

5
    private final RestTemplate restTemplate;
6
    private final CircuitBreakerRegistry circuitBreakerRegistry;
7

8
    @Value("${payment.service.url}")
9
    private String paymentServiceUrl;
10

11
    @CircuitBreaker(name = "paymentService", fallbackMethod = "processPaymentFallback")
12
    public PaymentResult processPayment(PaymentRequest request) {
13
        log.info("Processing payment: {}", request);
14
        return restTemplate.postForObject(
15
            paymentServiceUrl + "/api/payments",
16
            request,
17
            PaymentResult.class
18
        );
19
    }
20

21
    // 降级方法
22
    private PaymentResult processPaymentFallback(PaymentRequest request, Exception e) {
23
        log.warn("Payment service unavailable, using fallback", e);
24
        return PaymentResult.builder()
25
            .status("FALLBACK")
26
            .message("Payment service is temporarily unavailable")
27
            .build();
28
    }
29

30
    // 获取熔断器状态
31
    public CircuitBreaker.State getCircuitBreakerState() {
32
        return circuitBreakerRegistry.circuitBreaker("paymentService").getState();
33
    }
34
}

1
@FeignClient(
2
    name = "user-service",
3
    url = "${feign.client.config.user-service.url}",
4
    fallback = UserServiceFallback.class
5
)
6
public interface UserServiceClient {
7

8
    @GetMapping("/users/{userId}")
9
    ResponseEntity<User> getUserById(@PathVariable("userId") Long userId);
10

11
    @PostMapping("/users/batch")
12
    ResponseEntity<List<User>> batchGetUsers(@RequestBody List<Long> userIds);
13
}
14

15
// 降级实现
16
@Component
17
public class UserServiceFallback implements UserServiceClient {
18

19
    @Override
20
    public ResponseEntity<User> getUserById(Long userId) {
21
        log.warn("User service fallback triggered for user: {}", userId);
22
        return ResponseEntity.ok(User.builder()
23
            .id(userId)
24
            .name("Default User")
25
            .status("INACTIVE")
26
            .build());
27
    }
28

29
    @Override
30
    public ResponseEntity<List<User>> batchGetUsers(List<Long> userIds) {
31
        log.warn("Batch get users fallback triggered for: {}", userIds);
32
        return ResponseEntity.ok(Collections.emptyList());
33
    }
34
}

1
feign:
2
  client:
3
    config:
4
      default:
5
        connectTimeout: 5000
6
        readTimeout: 5000
7
        loggerLevel: basic
8
      user-service:
9
        url: http://user-service:8080
10
  circuitbreaker:
11
    enabled: true
12
  compression:
13
    request:
14
      enabled: true
15
    response:
16
      enabled: true
17

18
# Hystrix 配置 (如果使用)
19
hystrix:
20
  command:
21
    default:
22
      execution:
23
        isolation:
24
          thread:
25
            timeoutInMilliseconds: 10000
26
      circuitBreaker:
27
        requestVolumeThreshold: 20
28
        errorThresholdPercentage: 50
29
        sleepWindowInMilliseconds: 5000

1
-- KEYS[1]: 限流key
2
-- ARGV[1]: 时间窗口内最大请求数
3
-- ARGV[2]: 时间窗口大小(秒)
4
local key = KEYS[1]
5
local max_requests = tonumber(ARGV[1])
6
local window = tonumber(ARGV[2])
7

8
-- 获取当前时间戳(秒)
9
local current_time = redis.call('TIME')[1]
10
local window_start = current_time - window
11

12
-- 移除时间窗口外的请求记录
13
redis.call('ZREMRANGEBYSCORE', key, 0, window_start)
14

15
-- 获取当前窗口内的请求数
16
local current_requests = redis.call('ZCARD', key)
17

18
if current_requests < max_requests then
19
    -- 添加当前请求
20
    redis.call('ZADD', key, current_time, current_time .. '-' .. math.random(1000, 9999))
21
    -- 设置key的过期时间
22
    redis.call('EXPIRE', key, window)
23
    return 1  -- 允许访问
24
else
25
    return 0  -- 拒绝访问
26
end

1
@Service
2
@Slf4j
3
public class RateLimitService {
4

5
    private final RedisTemplate<String, String> redisTemplate;
6
    private final DefaultRedisScript<Long> rateLimitScript;
7

8
    public RateLimitService(RedisTemplate<String, String> redisTemplate) {
9
        this.redisTemplate = redisTemplate;
10
        this.rateLimitScript = new DefaultRedisScript<>();
11
        this.rateLimitScript.setScriptSource(new ResourceScriptSource(
12
            new ClassPathResource("lua/rate_limiter.lua")));
13
        this.rateLimitScript.setResultType(Long.class);
14
    }
15

16
    /**
17
     * 检查是否允许访问
18
     * @param key 限流key，如: rate_limit:user:123
19
     * @param maxRequests 时间窗口内最大请求数
20
     * @param windowInSeconds 时间窗口大小(秒)
21
     * @return 是否允许访问
22
     */
23
    public boolean allowRequest(String key, int maxRequests, int windowInSeconds) {
24
        try {
25
            Long result = redisTemplate.execute(
26
                rateLimitScript,
27
                Collections.singletonList(key),
28
                String.valueOf(maxRequests),
29
                String.valueOf(windowInSeconds)
30
            );
31
            return result != null && result == 1L;
32
        } catch (Exception e) {
33
            log.error("Rate limit check failed for key: " + key, e);
34
            // 限流组件异常时，根据业务需求决定是否放行
35
            return true;
36
        }
37
    }
38
}

1
@Aspect
2
@Component
3
@RequiredArgsConstructor
4
public class RateLimitAspect {
5

6
    private final RateLimitService rateLimitService;
7

8
    @Around("@annotation(rateLimit)")
9
    public Object around(ProceedingJoinPoint joinPoint, RateLimit rateLimit) throws Throwable {
10
        String key = rateLimit.key();
11
        // 支持SpEL表达式，如 #userId
12
        if (key.startsWith("#")) {
13
            key = parseSpel(key, joinPoint);
14
        }
15

16
        if (!rateLimitService.allowRequest(
17
            "rate_limit:" + key,
18
            rateLimit.maxRequests(),
19
            rateLimit.windowInSeconds())) {
20

21
            throw new RateLimitException("Too many requests, please try again later");
22
        }
23

24
        return joinPoint.proceed();
25
    }
26

27
    private String parseSpel(String expression, ProceedingJoinPoint joinPoint) {
28
        // 实现SpEL解析逻辑
29
        // ...
30
        return "parsed_key";
31
    }
32
}
33

34
// 限流注解
35
@Target(ElementType.METHOD)
36
@Retention(RetentionPolicy.RUNTIME)
37
public @interface RateLimit {
38
    String key();  // 限流key，支持SpEL
39
    int maxRequests() default 100;  // 窗口期内最大请求数
40
    int windowInSeconds() default 60;  // 时间窗口大小(秒)
41
}

1
@Configuration
2
@EnableRetry
3
public class RetryConfig {
4

5
    @Bean
6
    public RetryTemplate retryTemplate() {
7
        RetryTemplate retryTemplate = new RetryTemplate();
8

9
        // 指数退避策略
10
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
11
        backOffPolicy.setInitialInterval(1000);  // 初始间隔1秒
12
        backOffPolicy.setMultiplier(2.0);  // 倍数
13
        backOffPolicy.setMaxInterval(10000);  // 最大间隔10秒
14

15
        // 重试策略
16
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
17
        retryPolicy.setMaxAttempts(3);  // 最多重试3次
18

19
        retryTemplate.setBackOffPolicy(backOffPolicy);
20
        retryTemplate.setRetryPolicy(retryPolicy);
21

22
        return retryTemplate;
23
    }
24
}

1
@Service
2
@RequiredArgsConstructor
3
public class OrderService {
4

5
    private final PaymentGateway paymentGateway;
6
    private final RetryTemplate retryTemplate;
7

8
    public void processOrder(Order order) {
9
        retryTemplate.execute(context -> {
10
            try {
11
                // 尝试支付
12
                PaymentResult result = paymentGateway.charge(order);
13
                if (!result.isSuccess()) {
14
                    throw new PaymentException("Payment failed: " + result.getMessage());
15
                }
16
                return result;
17
            } catch (Exception e) {
18
                log.warn("Payment attempt {} failed: {}",
19
                    context.getRetryCount() + 1, e.getMessage());
20

21
                // 最后一次重试仍然失败
22
                if (context.getRetryCount() >= 2) {
23
                    log.error("All payment attempts failed", e);
24
                    // 触发补偿逻辑
25
                    compensateOrder(order, e);
26
                }
27
                throw e;  // 继续重试
28
            }
29
        });
30
    }
31

32
    private void compensateOrder(Order order, Exception e) {
33
        // 订单补偿逻辑
34
        log.error("Initiating order compensation for order: {}", order.getId(), e);
35
        // ...
36
    }
37
}

1
@RefreshScope
2
@Configuration
3
public class ResilienceConfig {
4

5
    @Value("${resilience.circuit-breaker.enabled:true}")
6
    private boolean circuitBreakerEnabled;
7

8
    @Value("${resilience.rate-limit.enabled:true}")
9
    private boolean rateLimitEnabled;
10

11
    @Value("${resilience.retry.max-attempts:3}")
12
    private int maxRetryAttempts;
13

14
    // Getters and setters
15
}

1
# 弹性配置
2
resilience:
3
  circuit-breaker:
4
    enabled: true
5
    failure-rate-threshold: 50
6
    wait-duration-in-open-state: 5s
7
    ring-buffer-size-in-half-open-state: 10
8
    ring-buffer-size-in-closed-state: 100
9

10
  rate-limit:
11
    enabled: true
12
    limit-for-period: 100
13
    limit-refresh-period: 1s
14
    timeout-duration: 0s
15

16
  retry:
17
    max-attempts: 3
18
    wait-duration: 1s
19
    enable-exponential-backoff: true
20
    exponential-backoff-multiplier: 2.0
21
    max-wait-duration: 10s

1
management:
2
  endpoints:
3
    web:
4
      exposure:
5
        include: health,info,metrics,prometheus
6
  metrics:
7
    export:
8
      prometheus:
9
        enabled: true
10
    distribution:
11
      percentiles-histogram:
12
        http.server.requests: true
13
    web:
14
      server:
15
        request:
16
          autotime:
17
            enabled: true

1
@Component
2
public class ResilienceMetrics {
3

4
    private final MeterRegistry meterRegistry;
5
    private final Map<String, Counter> circuitBreakerCounters = new ConcurrentHashMap<>();
6

7
    public ResilienceMetrics(MeterRegistry meterRegistry) {
8
        this.meterRegistry = meterRegistry;
9

10
        // 初始化熔断器状态变更监听
11
        CircuitBreakerRegistry
12
            .ofDefaults()
13
            .getEventPublisher()
14
            .onStateTransition(event -> {
15
                String circuitBreakerName = event.getCircuitBreakerName();
16
                Counter counter = circuitBreakerCounters.computeIfAbsent(
17
                    circuitBreakerName + ".state." + event.getStateTransition().getToState().name().toLowerCase(),
18
                    name -> Counter.builder("circuit_breaker_state_changes")
19
                        .tag("name", circuitBreakerName)
20
                        .tag("state", event.getStateTransition().getToState().name().toLowerCase())
21
                        .register(meterRegistry)
22
                );
23
                counter.increment();
24
            });
25
    }
26

27
    // 记录限流事件
28
    public void recordRateLimit(String key, boolean allowed) {
29
        Counter.builder("rate_limit_events")
30
            .tag("key", key)
31
            .tag("allowed", String.valueOf(allowed))
32
            .register(meterRegistry)
33
            .increment();
34
    }
35
}

1
// 秒杀接口实现
2
@Slf4j
3
@Service
4
public class SeckillServiceImpl implements SeckillService {
5

6
    private static final String SECKILL_STOCK_PREFIX = "seckill:stock:";
7
    private static final String SECKILL_USER_PREFIX = "seckill:user:";
8
    private static final String SECKILL_SCRIPT =
9
        "local stockKey = KEYS[1]\n" +
10
        "local userKey = KEYS[2]\n" +
11
        "local userId = ARGV[1]\n" +
12
        "local activityId = ARGV[2]\n" +
13
        "local quantity = tonumber(ARGV[3])\n" +
14
        "\n" +
15
        "-- 检查库存\n" +
16
        "local stock = tonumber(redis.call('get', stockKey) or '0')\n" +
17
        "if stock <= 0 then\n" +
18
        "    return 0\n" +
19
        "end\n" +
20
        "\n" +
21
        "-- 检查是否重复购买\n" +
22
        "local bought = redis.call('sismember', userKey, userId)\n" +
23
        "if bought == 1 then\n" +
24
        "    return 1\n" +
25
        "end\n" +
26
        "\n" +
27
        "-- 扣减库存\n" +
28
        "redis.call('decrby', stockKey, quantity)\n" +
29
        "redis.call('sadd', userKey, userId)\n" +
30
        "\n" +
31
        "-- 发送消息到队列\n" +
32
        "local orderInfo = string.format('{\"userId\":%s,\"activityId\":%s,\"quantity\":%s}', userId, activityId, quantity)\n" +
33
        "redis.call('publish', 'seckill.order', orderInfo)\n" +
34
        "\n" +
35
        "return 2";
36

37
    @Value("${seckill.limit.perUser:1}")
38
    private int seckillLimitPerUser;
39

40
    @Value("${seckill.token.expire:3600}")
41
    private int tokenExpireSeconds;
42

43
    private final StringRedisTemplate redisTemplate;
44
    private final RedissonClient redissonClient;
45
    private final OrderService orderService;
46
    private final SeckillTokenService tokenService;
47
    private final SeckillActivityMapper activityMapper;
48

49
    // 加载Lua脚本
50
    private final DefaultRedisScript<Long> seckillScript;
51

52
    public SeckillServiceImpl(StringRedisTemplate redisTemplate,
53
                            RedissonClient redissonClient,
54
                            OrderService orderService,
55
                            SeckillTokenService tokenService,
56
                            SeckillActivityMapper activityMapper) {
57
        this.redisTemplate = redisTemplate;
58
        this.redissonClient = redissonClient;
59
        this.orderService = orderService;
60
        this.tokenService = tokenService;
61
        this.activityMapper = activityMapper;
62

63
        this.seckillScript = new DefaultRedisScript<>();
64
        this.seckillScript.setScriptText(SECKILL_SCRIPT);
65
        this.seckillScript.setResultType(Long.class);
66
    }
67

68
    @Override
69
    @Transactional(rollbackFor = Exception.class)
70
    public SeckillResponse seckill(Long userId, Long seckillId, String token) {
71
        // 1. 验证令牌
72
        if (!tokenService.validateToken(userId, seckillId, token)) {
73
            return SeckillResponse.fail("非法请求");
74
        }
75

76
        // 2. 验证用户和活动
77
        User user = userService.getUser(userId);
78
        if (user == null) {
79
            return SeckillResponse.fail("用户不存在");
80
        }
81

82
        SeckillActivity activity = activityMapper.selectById(seckillId);
83
        if (activity == null || activity.getStatus() != 1) {
84
            return SeckillResponse.fail("活动不存在或已结束");
85
        }
86

87
        // 3. 使用分布式锁防止重复提交
88
        String lockKey = "seckill:lock:" + seckillId + ":" + userId;
89
        RLock lock = redissonClient.getLock(lockKey);
90

91
        try {
92
            // 尝试获取锁，最多等待100ms，锁持有时间30秒
93
            if (!lock.tryLock(100, 30000, TimeUnit.MILLISECONDS)) {
94
                return SeckillResponse.fail("操作太频繁，请稍后再试");
95
            }
96

97
            // 4. 执行Lua脚本保证原子性
98
            Long result = redisTemplate.execute(
99
                seckillScript,
100
                Arrays.asList(
101
                    SECKILL_STOCK_PREFIX + seckillId,
102
                    SECKILL_USER_PREFIX + seckillId
103
                ),
104
                String.valueOf(userId),
105
                String.valueOf(seckillId),
106
                String.valueOf(1)
107
            );
108

109
            if (result == null) {
110
                return SeckillResponse.fail("系统繁忙，请稍后再试");
111
            }
112

113
            // 5. 处理结果
114
            if (result == 0) {
115
                return SeckillResponse.fail("商品已售罄");
116
            } else if (result == 1) {
117
                return SeckillResponse.fail("您已经参加过该活动");
118
            } else if (result == 2) {
119
                // 6. 发送消息创建订单
120
                String orderNo = orderService.createOrderAsync(user, activity);
121

122
                // 7. 更新数据库库存（异步）
123
                updateStockInBackground(seckillId);
124

125
                return SeckillResponse.success(orderNo);
126
            } else {
127
                return SeckillResponse.fail("系统繁忙，请稍后再试");
128
            }
129

130
        } catch (Exception e) {
131
            log.error("秒杀失败: userId={}, seckillId={}", userId, seckillId, e);
132
            return SeckillResponse.fail("系统繁忙，请稍后再试");
133
        } finally {
134
            // 释放锁
135
            if (lock.isHeldByCurrentThread()) {
136
                lock.unlock();
137
            }
138
        }
139
    }
140

141
    // 异步更新数据库库存
142
    @Async
143
    protected void updateStockInBackground(Long seckillId) {
144
        String lockKey = "seckill:stock:update:lock:" + seckillId;
145
        RLock lock = redissonClient.getLock(lockKey);
146

147
        try {
148
            if (lock.tryLock(5, 30, TimeUnit.SECONDS)) {
149
                // 获取当前库存
150
                String stockKey = SECKILL_STOCK_PREFIX + seckillId;
151
                Integer stock = Integer.parseInt(redisTemplate.opsForValue().get(stockKey) ?? "0");
152

153
                // 批量更新数据库
154
                if (stock % 10 == 0) { // 每减少10个库存更新一次数据库
155
                    activityMapper.updateStock(seckillId, stock);
156
                }
157
            }
158
        } catch (Exception e) {
159
            log.error("更新库存失败: seckillId={}", seckillId, e);
160
        } finally {
161
            if (lock.isHeldByCurrentThread()) {
162
                lock.unlock();
163
            }
164
        }
165
    }
166

167
    // 获取秒杀令牌
168
    @Override
169
    public String getSeckillToken(Long userId, Long seckillId) {
170
        // 1. 验证活动状态
171
        SeckillActivity activity = activityMapper.selectById(seckillId);
172
        if (activity == null || activity.getStatus() != 1) {
173
            throw new BusinessException("活动不存在或已结束");
174
        }
175

176
        // 2. 验证用户资格
177
        if (hasReachedLimit(userId, seckillId)) {
178
            throw new BusinessException("您已经参加过该活动");
179
        }
180

181
        // 3. 生成令牌
182
        String token = UUID.randomUUID().toString().replace("-", "");
183
        String tokenKey = "seckill:token:" + seckillId + ":" + userId;
184

185
        // 4. 存储令牌，设置过期时间
186
        redisTemplate.opsForValue().set(
187
            tokenKey,
188
            token,
189
            tokenExpireSeconds,
190
            TimeUnit.SECONDS
191
        );
192

193
        return token;
194
    }
195

196
    // 检查用户是否达到购买限制
197
    private boolean hasReachedLimit(Long userId, Long seckillId) {
198
        String userKey = SECKILL_USER_PREFIX + seckillId;
199
        return Boolean.TRUE.equals(redisTemplate.opsForSet().isMember(userKey, userId.toString()));
200
    }
201
}