夢の終着駅

1
package com.yilena.seckill.config;
2

3
import lombok.extern.slf4j.Slf4j;
4
import org.springframework.context.annotation.Bean;
5
import org.springframework.context.annotation.Configuration;
6
import org.springframework.core.io.ClassPathResource;
7
import org.springframework.data.redis.connection.RedisConnectionFactory;
8
import org.springframework.data.redis.core.RedisTemplate;
9
import org.springframework.data.redis.core.script.DefaultRedisScript;
10
import org.springframework.data.redis.core.script.RedisScript;
11
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
12
import org.springframework.data.redis.serializer.StringRedisSerializer;
13

14
@Configuration
15
@Slf4j
16
public class RedisConfig {
17

18
    @Bean
19
    public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) {
20
        log.info("开始创建redis模板对象...");
21
        RedisTemplate redisTemplate = new RedisTemplate();
22
        //设置redis的连接工厂对象
23
        redisTemplate.setConnectionFactory(redisConnectionFactory);
24
        //设置redis key的序列化器
25
        redisTemplate.setKeySerializer(new StringRedisSerializer());
26
        return redisTemplate;
27
    }
28

29
    @Bean(name = "seckillScript")
30
    public RedisScript<Long> seckillScript() {
31
        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
32
        script.setLocation(new ClassPathResource("scripts/seckill.lua")); // 确保路径正确
33
        script.setResultType(Long.class);
34
        return script;
35
    }
36
}

1
package com.yilena.seckill.config;
2

3
import com.yilena.seckill.constant.RabbitMqConstant;
4
import lombok.RequiredArgsConstructor;
5
import lombok.extern.slf4j.Slf4j;
6
import org.springframework.amqp.core.Binding;
7
import org.springframework.amqp.core.BindingBuilder;
8
import org.springframework.amqp.core.DirectExchange;
9
import org.springframework.amqp.core.Queue;
10
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
11
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
12
import org.springframework.amqp.rabbit.core.RabbitTemplate;
13
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
14
import org.springframework.amqp.support.converter.MessageConverter;
15
import org.springframework.context.annotation.Bean;
16
import org.springframework.context.annotation.Configuration;
17

18
@Slf4j
19
@Configuration
20
@RequiredArgsConstructor
21
public class RabbitMqConfig {
22

23
    private final ConnectionFactory connectionFactory;
24

25
    @Bean
26
    public MessageConverter jsonMessageConverter() {
27
        return new Jackson2JsonMessageConverter();
28
    }
29

30
    @Bean
31
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
32
        // 实例化
33
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
34
        // 设置消息转换器为jsonMessageConverter，确保消息以JSON格式进行发送和接收
35
        template.setMessageConverter(jsonMessageConverter());
36
        return template;
37
    }
38

39
    // 创建监听容器工厂
40
    @Bean
41
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
42
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
43
        // 设置连接工厂，用于后续创建的监听器容器连接到RabbitMQ
44
        factory.setConnectionFactory(connectionFactory);
45
        factory.setMessageConverter(jsonMessageConverter());
46
        return factory;
47
    }
48

49
    // 创建交换机
50
    @Bean
51
    public DirectExchange directExchange() {
52
        return new DirectExchange(RabbitMqConstant.SECKILL_EXCHANGE);
53
    }
54

55
    // 创建队列
56
    @Bean
57
    public Queue seckillQueue() {
58
        return new Queue(RabbitMqConstant.SECKILL_QUEUE);
59
    }
60

61
    // 绑定队列和交换机
62
    @Bean
63
    public Binding binding(DirectExchange directExchange , Queue seckillQueue) {
64
        return BindingBuilder.bind(seckillQueue)
65
                .to(directExchange)
66
                .with(RabbitMqConstant.SECKILL_BINDING_KEY);
67
    }
68
}

1
package com.yilena.seckill.config;
2

3
import com.yilena.seckill.service.RetryService;
4
import lombok.RequiredArgsConstructor;
5
import lombok.extern.slf4j.Slf4j;
6
import org.springframework.amqp.rabbit.core.RabbitTemplate;
7
import org.springframework.beans.BeansException;
8
import org.springframework.beans.factory.config.BeanPostProcessor;
9
import org.springframework.stereotype.Component;
10

11
/**
12
* 重试机制
13
*/
14
@Component
15
@Slf4j
16
@RequiredArgsConstructor
17
public class RabbitTemplateCallbackConfigurer implements BeanPostProcessor {
18

19
    private final RetryService retryService;
20

21
    // 在监听器初始化后才会触发
22
    @Override
23
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
24
        // 检查是否是RabbitTemplate
25
        if (bean instanceof RabbitTemplate) {
26
            RabbitTemplate rabbitTemplate = (RabbitTemplate) bean;
27
            // 设置确认回调
28
            rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
29
                // 当消息发送未被确认时，会调用此方法
30
                if (!ack) {
31
                    log.error("消息发送失败: {}", correlationData != null ? correlationData.getId() : "unknown");
32
                    retryService.asyncRetry(correlationData);
33
                }
34
            });
35
        }
36
        // 返回处理后的bean实例
37
        return bean;
38
    }
39
}

1
package com.yilena.seckill.config;
2

3
import com.yilena.seckill.filter.CachedBodyFilter;
4
import org.springframework.boot.web.servlet.FilterRegistrationBean;
5
import org.springframework.context.annotation.Bean;
6
import org.springframework.context.annotation.Configuration;
7
import org.springframework.core.Ordered;
8

9
@Configuration
10
public class FilterConfig {
11

12
    @Bean
13
    public FilterRegistrationBean<CachedBodyFilter> requestCachingFilterRegistration() {
14
        FilterRegistrationBean<CachedBodyFilter> registration = new FilterRegistrationBean<>();
15
        registration.setFilter(new CachedBodyFilter());
16
        // 设置拦截路径
17
        registration.addUrlPatterns("/seckill/buy");
18
        // 设置最高优先级，确保先于其他Filter执行
19
        registration.setOrder(Ordered.HIGHEST_PRECEDENCE);
20
        return registration;
21
    }
22
}

1
package com.yilena.seckill.config;
2

3
import com.yilena.seckill.intercepter.SeckillInterceptor;
4
import lombok.RequiredArgsConstructor;
5
import org.springframework.context.annotation.Configuration;
6
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
7
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
8

9
@Configuration
10
@RequiredArgsConstructor
11
public class WebMvcInterceptorConfig implements WebMvcConfigurer {
12

13
    private final SeckillInterceptor seckillInterceptor;
14

15
    @Override
16
    public void addInterceptors(InterceptorRegistry registry) {
17
        registry.addInterceptor(seckillInterceptor)
18
                .addPathPatterns("/**");
19
    }
20
}

1
package com.yilena.seckill.config;
2

3
import com.yilena.seckill.utils.SnowFlake;
4
import com.yilena.seckill.utils.SnowFlakeProperties;
5
import org.springframework.boot.context.properties.EnableConfigurationProperties;
6
import org.springframework.context.annotation.Bean;
7
import org.springframework.context.annotation.Configuration;
8

9
@Configuration
10
@EnableConfigurationProperties(SnowFlakeProperties.class)
11
public class SnowFlakeConfig {
12

13
    @Bean
14
    public SnowFlake snowFlake(SnowFlakeProperties properties) {
15
      return new SnowFlake(properties);
16
    }
17
}

1
package com.yilena.seckill.constant;
2

3
public class RabbitMqConstant {
4
    public static final String SECKILL_EXCHANGE = "seckill.exchange";
5
    public static final String SECKILL_QUEUE = "seckill.queue";
6
    public static final String SECKILL_BINDING_KEY = "seckill.binding.key";
7
}

1
package com.yilena.seckill.constant;
2

3
public class RedisKeyConstant {
4
    public static final String SHOP_SECKILL_MSG_KEY = "shop:seckill:msg:";
5
    public static final String SHOP_SECKILL_COUNT_KEY = "shop:seckill:count:";
6
    public static final String SHOP_SECKILL_ORDER_KEY = "shop:seckill:order:";
7
    public static final String SHOP_SECKILL_ORDER_MQ_KEY = "shop:seckill:order:mq:";
8
}

1
package com.yilena.seckill.utils;
2

3
import jakarta.servlet.ReadListener;
4
import jakarta.servlet.ServletInputStream;
5
import jakarta.servlet.http.HttpServletRequest;
6
import jakarta.servlet.http.HttpServletRequestWrapper;
7

8
import java.io.BufferedReader;
9
import java.io.ByteArrayInputStream;
10
import java.io.IOException;
11
import java.io.InputStream;
12
import java.io.InputStreamReader;
13

14
public class CachedBodyHttpServletRequest extends HttpServletRequestWrapper {
15

16
    // 用于缓存请求体序列化的字节数组
17
    private final byte[] cachedBody;
18

19
    // 构造函数，将请求体序列化到cachedBody中
20
    public CachedBodyHttpServletRequest(HttpServletRequest request) throws IOException {
21
        super(request);
22
        InputStream inputStream = request.getInputStream();
23
        this.cachedBody = inputStream.readAllBytes();
24
    }
25

26
    @Override
27
    public ServletInputStream getInputStream() {
28
        return new CachedBodyServletInputStream(cachedBody);
29
    }
30

31
    //  重写getReader方法，返回可重复读取的BufferedReader
32
    @Override
33
    public BufferedReader getReader() {
34
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(cachedBody);
35
        return new BufferedReader(new InputStreamReader(byteArrayInputStream));
36
    }
37

38
    // 提供可重复读取的ServletInputStream
39
    public static class CachedBodyServletInputStream extends ServletInputStream {
40
        private final ByteArrayInputStream inputStream;
41

42
        public CachedBodyServletInputStream(byte[] cachedBody) {
43
            this.inputStream = new ByteArrayInputStream(cachedBody);
44
        }
45

46
        @Override
47
        public int read() {
48
            return inputStream.read();
49
        }
50

51
        @Override
52
        public boolean isFinished() {
53
            return inputStream.available() == 0;
54
        }
55

56
        @Override
57
        public boolean isReady() {
58
            return true;
59
        }
60

61
        @Override
62
        public void setReadListener(ReadListener listener) {
63
            throw new UnsupportedOperationException();
64
        }
65
    }
66
}

1
package com.yilena.seckill.utils;
2

3
import java.util.ArrayList;
4
import java.util.List;
5
import java.util.concurrent.CompletableFuture;
6

7
public class ThreadPoolUtil {
8

9
    public static <T> List<List<T>> slicingData(List<T> dataList, Integer splitSize) {
10

11
        int dataSize = dataList.size();
12
        int groupSize = dataSize / splitSize;
13
        if (dataSize % splitSize != 0) {
14
            groupSize = groupSize + 1;
15
        }
16

17
        List<List<T>> chunks = new ArrayList<>(groupSize);
18
        for (int i = 0; i < groupSize; i++) {
19
            int fromIndex = i * splitSize;
20
            int toIndex = fromIndex + splitSize;
21
            List<T> suppliesReadData;
22
            if (i == groupSize - 1) {
23
                suppliesReadData = dataList.subList(fromIndex, dataList.size());
24
            } else {
25
                suppliesReadData = dataList.subList(fromIndex, toIndex);
26
            }
27
            chunks.add(suppliesReadData);
28
        }
29
        return chunks;
30
    }
31

32
    public static void allFuturesWait(List<CompletableFuture<Void>> completableFutures) {
33
        // 等待所有异步任务完成后，可以继续进行下一步操作
34
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0]));
35
        allFutures.join();
36
    }
37
}

1
package com.yilena.seckill.utils;
2

3
import org.springframework.stereotype.Component;
4

5
@Component
6
public class SnowFlake {
7

8
    private SnowFlakeProperties properties;
9

10
    public SnowFlake() {
11
    }
12

13
    public SnowFlake(SnowFlakeProperties properties) {
14
        this.properties = properties;
15
    }
16
    public synchronized  long getID(){
17
        //获得当前时间
18
        long now = System.currentTimeMillis();
19
        long lastTimeStamp = properties.getLastTimeStamp();
20
        //当前系统时间小于上一次记录时间
21
        if (now < lastTimeStamp){
22
            throw new RuntimeException("时钟回拨异常");
23
        }
24
        //相同时间 要序列号进制增量
25
        if (now == lastTimeStamp){
26
            //防止溢出
27
            long sequence = properties.getSequence();
28
            sequence = (sequence + 1) & properties.getSequenceMask();
29
            //更新sequence 的值
30
            properties.setSequence(sequence);
31
            if (sequence == 0L){
32
                //溢出处理
33
                try {
34
                    Thread.sleep(1L);
35
                } catch (InterruptedException e) {
36
                    throw new RuntimeException(e);
37
                }
38
                //获取下一毫秒时间 （有锁）
39
                now = System.currentTimeMillis();
40
            }
41
        }else {
42
            //置0
43
            properties.setSequence(0L);
44
        }
45
        //记录当前时间
46
        properties.setLastTimeStamp(now);
47
        return ((now - properties.getStartTimeStamp()) << properties.getTimeStampIndex()) |
48
                (properties.getDataCenterID() << properties.getDataCenterIndex())|
49
                (properties.getWorkID() << properties.getWorkIDIndex()) | properties.getSequence();
50
    }
51
}

1
package com.yilena.seckill.utils;
2

3
import lombok.AllArgsConstructor;
4
import lombok.Data;
5
import lombok.Setter;
6
import org.springframework.boot.context.properties.ConfigurationProperties;
7

8
@Data
9
@AllArgsConstructor
10
@ConfigurationProperties(prefix = "sf")
11
public class SnowFlakeProperties {
12
    // 起始时间戳
13
    private long startTimeStamp = 1577808000L;
14
    // 机器ID
15
    private long workID;
16
    // 数据中心ID
17
    private long dataCenterID = 0;
18
    // 序列号
19
    private long sequence = 0L;
20
    // 数据中心位移位数
21
    private long dataCenterIndex;
22
    // 机器ID位移位数
23
    private long workIDIndex;
24
    // 时间戳位移位数
25
    private long timeStampIndex;
26
    // 记录上一次时间戳
27
    private long lastTimeStamp = -1L;
28
    // 序列号长度
29
    private long sequenceLength;
30

31
    // 序列号掩码（动态计算）
32
    private long sequenceMask;
33

34
    public SnowFlakeProperties() {
35
        this.sequence = 0L;
36
    }
37

38
    // 当sequenceLength被设置时自动计算sequenceMask
39
    public void setSequenceLength(long sequenceLength) {
40
        this.sequenceLength = sequenceLength;
41
        this.sequenceMask = ~(-1L << sequenceLength);
42
    }
43
}

1
package com.yilena.seckill.utils;
2

3
public class SnowFlakeValidator {
4

5
    public static boolean isSnowFlakeID(long id, SnowFlakeProperties props) {
6
        // 验证序列号：必须在 0 ~ sequenceMask 之间
7
        long sequenceMask = props.getSequenceMask();
8
        long sequence = id & sequenceMask;
9
        if (sequence < 0 || sequence > sequenceMask) {
10
            return false;
11
        }
12

13
        // 验证数据中心ID
14
        long dataCenterBits = props.getWorkIDIndex() - props.getDataCenterIndex();
15
        long dataCenterMask = (1L << dataCenterBits) - 1;
16
        long dataCenterID = (id >> props.getDataCenterIndex()) & dataCenterMask;
17
        if (dataCenterID != props.getDataCenterID()) {
18
            return false;
19
        }
20

21
        // 验证机器ID
22
        long workIDBits = props.getTimeStampIndex() - props.getWorkIDIndex();
23
        long workIDMask = (1L << workIDBits) - 1;
24
        long workID = (id >> props.getWorkIDIndex()) & workIDMask;
25
        if (workID != props.getWorkID()) {
26
            return false;
27
        }
28

29
        // 验证时间戳
30
        long timeStamp = (id >> props.getTimeStampIndex()) + props.getStartTimeStamp();
31
        if (timeStamp < props.getStartTimeStamp()) {
32
            return false;
33
        }
34

35
        return true;
36
    }
37
}

1
package com.yilena.seckill.controller;
2

3
import com.yilena.seckill.entity.Result;
4
import com.yilena.seckill.entity.po.Order;
5
import com.yilena.seckill.service.SeckillService;
6
import com.yilena.seckill.utils.SnowFlake;
7
import lombok.RequiredArgsConstructor;
8
import lombok.extern.slf4j.Slf4j;
9
import org.springframework.web.bind.annotation.*;
10

11
@Slf4j
12
@RestController
13
@RequestMapping("/seckill")
14
@RequiredArgsConstructor
15
public class SeckillController {
16

17
    private final SeckillService seckillService;
18

19
    @PostMapping( "/buy")
20
    public Result buy(@RequestBody Order order){
21
        log.info("用户id为{}，下单了秒杀商品：{}", order.getUserId(), order.getShopId());
22
        return seckillService.buy(order);
23
    }
24
}

1
package com.yilena.seckill.service.serviceImpl;
2

3
import cn.hutool.json.JSONUtil;
4
import com.yilena.seckill.constant.RabbitMqConstant;
5
import com.yilena.seckill.constant.RedisKeyConstant;
6
import com.yilena.seckill.dao.SeckillDao;
7
import com.yilena.seckill.entity.Result;
8
import com.yilena.seckill.entity.po.Order;
9
import com.yilena.seckill.entity.po.ShopSeckill;
10
import com.yilena.seckill.service.SeckillService;
11
import com.yilena.seckill.utils.SnowFlake;
12
import jakarta.annotation.Resource;
13
import lombok.RequiredArgsConstructor;
14
import lombok.extern.slf4j.Slf4j;
15
import org.springframework.amqp.core.MessageDeliveryMode;
16
import org.springframework.amqp.rabbit.connection.CorrelationData;
17
import org.springframework.amqp.rabbit.core.RabbitTemplate;
18
import org.springframework.data.redis.core.StringRedisTemplate;
19
import org.springframework.data.redis.core.script.RedisScript;
20
import org.springframework.stereotype.Service;
21

22
import java.time.LocalDateTime;
23
import java.util.Collections;
24

25
@Slf4j
26
@Service
27
@RequiredArgsConstructor
28
public class SeckillServiceImpl implements SeckillService {
29

30
    private final StringRedisTemplate stringRedisTemplate;
31
    private final SnowFlake snowFlake;
32
    private final RabbitTemplate rabbitTemplate;
33

34
    @Resource(name = "seckillScript")
35
    private RedisScript<Long> seckillScript;
36

37

38
    @Override
39
    public Result buy(Order order) {
40

41
        // 获取订单信息
42
        Long shopId = order.getShopId();
43
        Long userId = order.getUserId();
44

45
         //执行lua脚本，执行下单逻辑
46
        Long result = stringRedisTemplate.execute(
47
                seckillScript,
48
                Collections.emptyList(),
49
                userId.toString(), shopId.toString(),
50
                String.valueOf(LocalDateTime.now()
51
                        .atZone(java.time.ZoneId.systemDefault())
52
                        .toInstant()
53
                        .toEpochMilli())
54
        );
55

56
        // 处理可能的null返回值
57
        if (result == null) {
58
            log.error("Lua脚本返回null, shopId: {}", shopId);
59
            return Result.error("系统异常，请重试");
60
        }
61

62
        int resultCode = result.intValue();
63

64
        // 扩展错误处理
65
        switch (resultCode) {
66
            case 500:
67
                // 下单完成，生成订单ID
68
                Long orderId = snowFlake.getID();
69
                order.setId(orderId);
70
                order.setCreateTime(LocalDateTime.now());
71
                order.setStatus(0);
72

73
                // 将订单信息发送到mq，异步处理
74
                rabbitTemplate.convertAndSend(
75
                        RabbitMqConstant.SECKILL_EXCHANGE,
76
                        RabbitMqConstant.SECKILL_BINDING_KEY,
77
                        order,
78
                        message -> {
79
                            // 消息持久化, 防止消息丢失
80
                            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
81
                            return message;
82
                        },
83
                        // 生成关联id，便于重试
84
                        new CorrelationData(order.getId().toString())
85
                );
86

87
                // 先返回订单信息交给前端进行响应，优化用户体验
88
                return Result.success(order);
89

90
            case 400:
91
                return Result.error("当前商品已售空！");
92
            case 401:
93
                return Result.error("商品不存在");
94
            case 402:
95
                return Result.error("秒杀活动尚未开始！");
96
            case 403:
97
                return Result.error("相同商品仅限一单！");
98
            case 404:
99
                return Result.error("商品库存信息异常！");
100
            default:
101
                log.warn("未知返回码: {}", resultCode);
102
                return Result.error("秒杀失败，请重试");
103
        }
104
    }
105
}