简易版秒杀系统实现Redis+RocketMQ+MySQL_redis和rocketmq集成实际使用

在应用高并发时，应尽量减少数据库操作，大批量数据操作入库会导致数据库服务挂掉。

解决思路

1.创建秒杀时，先将秒杀信息存入redis
2.秒杀前一段时间，数据信息查询量较大，从缓存查询
3.秒杀时，从redis减少库存，添加用户下单信息到MQ，此处最好多放入一些用户，大于秒杀库存，避免MQ消费失败，库存没有完全消耗。当redis库存不足时，直接返回秒杀完毕
4.消费者开始消费，当数据库秒杀表库存为0时，MQ后消息秒杀失败
5.redis下单成功后，页面等待中， 可轮询结果。先查询订单，如果不存在， 再查询秒杀库存是否为0，如果不为0，返回继续等待，否则返回失败
注意：秒杀库存和商品库存都应该使用乐观锁或分布式锁实现，避免超卖

pom.xml

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-openfeign</artifactId>
    </dependency>

    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
        <version>2.1.1.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        <version>2.1.1.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.1.1</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jdbc</artifactId>
    </dependency>
    <dependency>
        <groupId>org.mybatis.spring.boot</groupId>
        <artifactId>mybatis-spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.data</groupId>
        <artifactId>spring-data-redis</artifactId>
    </dependency>
    <dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>2.9.0</version>
    </dependency>
    <dependency>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-boot-starter</artifactId>
        <version>3.4.0</version>
    </dependency>

</dependencies>
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

bootstrap.yml

server:
  port: 8081

spring:
  application:
    name: consumer
  datasource:
    url: jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=utf8&characterSetResults=utf8&serverTimezone=GMT%2B8
    username: root
    password: root
    driver-class-name: com.mysql.jdbc.Driver

  redis:
    host: 127.0.0.1
    port: 6379

  cloud:
    nacos:
      config:
        server-addr: localhost:8848
        file-extension: yaml
      discovery:
        server-addr: localhost:8848

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: Producer-Group
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

创建秒杀

@Override
public boolean create(SecondKillDTO dto) {
    // 将库存持久化至redis 50件商品 可以创建100个redis预存
    if (dto.getGoodsId() == null || dto.getStore() < 0) {
        return false;
    }
    GoodsDTO goods = goodsService.get(dto.getGoodsId());
    if (goods == null) {
        return false;
    }
    // 这里可以不校验库存 mq下单时 如果库存不足 直接返回失败
    if (goods.getStore() < dto.getStore()) {
        return false;
    }
    if (dto.getPrice() == null) {
        dto.setPrice(goods.getPrice());
    }
    if (dto.getPrice().compareTo(BigDecimal.ZERO) <= 0) {
        return false;
    }
    if (dto.getStartTime() == null) {
        dto.setStartTime(new Date());
    }
    if (dto.getEndTime() == null) {
        dto.setEndTime(new Date(dto.getStartTime().getTime() + 60 * 60 * 1000L));
    }
    if (dto.getEndTime().before(new Date())) {
        return false;
    }
    if (secondKillMapper.insert(dto) != 1) {
        return false;
    }
    redisTemplate.opsForValue().set(REDIS_PATH_SK_STORE + ":" + dto.getId(), String.valueOf(dto.getStore() * 2)); // 增加入队数据 避免因处理失败导致不能售空
    redisTemplate.opsForHash().put(REDIS_PATH_SK_DETAIL, String.valueOf(dto.getId()), JSONObject.toJSONString(dto));
    return true;
}
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

查看秒杀详情

@Override
public SecondKillDTO detail(long id) {
    // 从redis中获取商品详情 大并发100w请求
    String str = (String) redisTemplate.opsForHash().get(REDIS_PATH_SK_DETAIL, String.valueOf(id));
    if (StringUtils.isEmpty(str)) {
        return null;
    }
    SecondKillDTO dto = JSONObject.parseObject(str, SecondKillDTO.class);
    if (dto == null) {
        dto = secondKillMapper.selectById(id);
        if (dto == null) {
            return null;
        }
    }
    if (dto.getEndTime().before(new Date())) {
        return null;
    }
    redisTemplate.opsForHash().put(REDIS_PATH_SK_DETAIL, String.valueOf(dto.getId()), JSONObject.toJSONString(dto));
    return dto;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

下单

向队列中发送消息，可不使用事务消息，因为redis预库存大于实际秒杀库存，部分失败对系统不影响

@Override
public boolean sendOrderMsg(OrderDTO dto) {
    // 做一些检查、重复下单、接口攻击
    if (dto.getSecondKillId() == null || dto.getGoodsCount() <= 0) {
        return false;
    }
    // redis cas更新库存
    Long currentStore = redisTemplate.opsForValue().increment(REDIS_PATH_SK_STORE + ":" + dto.getSecondKillId(), -dto.getGoodsCount());
    if (currentStore == null || currentStore < 0) {
        return false;
    }
    // 进入队列 使用队列消费
    boolean result = sendSyncMsg("create_order", JSONObject.toJSONString(dto));
    if (!result) {
        // TODO 库存回退
    }
    return true;
}

boolean sendSyncMsg(String tag, String msg) {
    String destination = StringUtils.isEmpty(tag) ? TOPIC : TOPIC + ":" + tag;
    org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(msg).build();
    SendResult sendResult = rocketMQTemplate.syncSend(destination, message);
    System.out.println("Send syn msg result: " + sendResult);
    return sendResult.getSendStatus() == SendStatus.SEND_OK;
}
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

查询下单结果

@Override
publiclong queryResult(long userId, long secondKillId) {
    // 根据当前用户信息和当前秒杀id查询结果 100个设备循环查询
    // TODO 此处查询最好添加缓存 避免高并发查询
    OrderDTO order = orderService.getBySecondKillId(secondKillId, userId);
    if (order != null) {
        return order.getId();
    }
    // 校验秒杀是否已经完成
    SecondKillDTO secondKill = secondKillMapper.selectById(secondKillId);
    if (secondKill == null || secondKill.getStore() <= secondKill.getOrderCount()) {
        return -1; // 没有表示秒杀失败
    }
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

消费者

因为使用的是乐观锁更新数据，所以应设置最大并发消费线程个数consumeThreadMax不能太多，避免因并发量太大，导致大部分乐观锁更新失败

@Component
@RocketMQMessageListener(topic = "TOPIC_A", consumerGroup = "Group_creat_order", selectorExpression = "create_order", consumeThreadMax = 2)
public class CreateOrderListener implements RocketMQListener<String> {

    @Autowired
    private OrderService orderService;

    public void onMessage(String s) {
        OrderDTO order = JSONObject.parseObject(s, OrderDTO.class);
        if (order == null) {
            System.out.println("Order msg parse failed: " + s);
            return;
        }
        order = orderService.create(order);
        if (order == null) {
            System.out.println("Order msg consumer failed: " + s);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

执行下单逻辑

@Override
@Transactional
public OrderDTO create(OrderDTO dto) {
    // TODO check
    if (dto.getSecondKillId() != null) {
        SecondKillDTO secondKill = secondKillService.get(dto.getSecondKillId());
        if (secondKill == null) {
            return null;
        }
        if (!secondKillService.reduceStore(dto.getSecondKillId(), dto.getGoodsCount())) {
            throw new RuntimeException("Reduce store failed");
        }
        dto.setGoodsId(secondKill.getGoodsId());
        dto.setGoodsPrice(secondKill.getPrice());
    } else {
        if (dto.getGoodsId() == null) {
            throw new RuntimeException("Goods id cannot null");
        }
        GoodsDTO goods = goodsService.get(dto.getGoodsId());
        if (goods == null) {
            throw new RuntimeException("Goods not found");
        }
        dto.setGoodsPrice(goods.getPrice());
    }
    dto.setAmount(dto.getGoodsPrice().multiply(BigDecimal.valueOf(dto.getGoodsCount())));
    if (!goodsService.reduceStore(dto.getGoodsId(), dto.getGoodsCount())) {
        throw new RuntimeException("Reduce store failed");
    }
    dto.setCreatedTime(new Date());
    if (orderMapper.insert(dto) != 1) {
        throw new RuntimeException("Create order failed");
    }
    return dto;
}
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

以上只是简单的实现秒杀逻辑，还有许多待优化，如有问题，欢迎指正！