场馆预定平台高并发时间段预定实现V2

🕗 发布于 2025-01-22 10:51 MySQL SpringBoot 高并发场馆预定消息队列

🎯 本文档介绍了场馆预订系统接口V2的设计与实现，旨在解决V1版本中库存数据不一致及性能瓶颈的问题。通过引入令牌机制确保缓存和数据库库存的最终一致性，避免因服务器故障导致的库存错误占用问题。同时，采用消息队列异步处理库存扣减和订单创建，显著提升了接口的吞吐量和响应速度。测试结果显示，新版接口在高并发场景下表现优异，平均响应时间为1801毫秒，吞吐量达到了每秒1045.8次请求，异常率仅为0.22%，极大改善了用户体验。

文章目录

说明
避免空场无法预订
接口性能提升
测试结果
说明

说明

在阅读此文之前，建议先阅读预订接口V1实现：https://hellodam.blog.csdn.net/article/details/144950335

接口 V2 主要是解决 V1 存在的一些问题：

问题一：接口 V1 中存在如下问题：假如说 lua 脚本执行完成，缓存中的库存已经扣减，结果突然服务器宕机了，没有执行后续的数据库库存扣减和创建订单流程，就会出现库存被错误占用，导致缓存中库存小于实际库存。对应于现实，就是有的场空着，用户预定不到
问题二：接口 V1 中，因为库存扣减和订单创建是同步的，预订接口吞吐量较低。为了进一步提升接口性能，可以使用消息队列来异步执行库存扣减和订单创建逻辑

避免空场无法预订

缓存扣减完成之后，由于发生故障，导致没有更新数据库。这个问题本身是无法避免的，只能通过一些机制来兜底。本文通过使用令牌机制来解决空场无法预订问题。

在接口 V1 中，用户请求预定接口，先查看 Redis 缓存中的库存是否大于 0 ，大于 0 才进行后面的操作。令牌是什么，其实也是这个缓存，但是我们并不完全相信它，我们知道它可能和数据库的数据不一致。当用户获取不到令牌的时候，我们不是直接返回时间段售罄错误，而是先查询一下数据库，看看是不是真的售罄了，如果数据库中还有库存，就删除令牌缓存。这样下一个用户再发起预订时，就会重新刷新令牌缓存，这样令牌的数据就和数据库保持一致，就不会出现空场无法预订的问题。

为了实现这个思路，我们还需要考虑一个问题，难道每个用户看到没有令牌都去查数据库吗，那肯定不行，这样并发高的话，数据库很容易被打崩。可以通过分布式锁让同一时刻只有一个用户查询数据库，但是光是添加分布式锁还是不行，用户请求多时，可能出现不同时间点连续查询数据库刷新token的情况，其实不必如此频繁查询。还有一个问题，高并发时大量任务等着数据库响应，数据库更新不会那么快。如果是立刻刷新token，可能出现数据库没来得及扣减库存，就被刷新到token中了，这样会导致时间段超卖，因为令牌数量大于库存。为了解决上述问题，可以先延时10秒再刷新token，在这10秒内，其他用户访问预定接口，因为拿不到分布式锁，也不会重复执行token刷新。

/**
 * 查询数据库是否还有库存，如果还有的话，删除令牌，让下一个用户重新加载令牌缓存
 *
 * @param timePeriodId
 */
private void refreshTokenByCheckDatabase(Long timePeriodId) {
    RLock lock = redissonClient.getLock(String.format(RedisCacheConstant.VENUE_LOCK_TIME_PERIOD_REFRESH_TOKEN_KEY, timePeriodId));
    // 尝试获取分布式锁，获取不成功直接返回
    if (!lock.tryLock()) {
        return;
    }
    // 延迟 10 秒之后去检查数据库和令牌是否一致
    // 为啥要延迟？如果不延迟的话，可能高峰期时，大量请求过来，数据库还没来得及更新，就触发令牌刷新，导致超卖
    tokenRefreshExecutor.schedule(() -> {
        try {
            TimePeriodDO timePeriodDO = this.getById(timePeriodId);
            if (timePeriodDO.getStock() > 0) {
                // --if-- 数据库中还有库存，说明数据库中的库存和令牌中的库存不一致，删除缓存，让下一个用户重新获取
                stringRedisTemplate.delete(RedisCacheConstant.VENUE_TIME_PERIOD_STOCK_TOKEN_KEY);
                stringRedisTemplate.delete(RedisCacheConstant.VENUE_TIME_PERIOD_FREE_INDEX_BIT_MAP_TOKEN_KEY);
            }
        } finally {
            lock.unlock();
        }
    }, 10, TimeUnit.SECONDS);
}

接口性能提升

Controller

/**
 * 预定时间段
 */
@GetMapping("/v2/reserve")
@Idempotent(
        uniqueKeyPrefix = "vrs-venue:lock_reserve:",
        // 让用户同时最多只能预定一个时间段，根据用户名来加锁
        // key = "T(com.vrs.common.context.UserContext).getUsername()",
        // 让用户同时最多只能预定该时间段一次，但是可以同时预定其他时间段，根据用户名+时间段ID来加锁
        key = "T(com.vrs.common.context.UserContext).getUsername()+'_'+#timePeriodId",
        message = "正在执行场馆预定流程，请勿重复预定...",
        scene = IdempotentSceneEnum.RESTAPI
)
@Operation(summary = "预定时间段V2")
public Result reserve2(@RequestParam("timePeriodId") Long timePeriodId) {
    timePeriodService.reserve2(timePeriodId);
    return Results.success();
}

Service

【预订流程】

参数检验
获取令牌
- 能获取到，执行下一步
- 获取不到，查询数据库，刷新令牌缓存
发送消息，异步更新库存并生成订单

/**
 * 尝试获取令牌，令牌获取成功之后，发送消息，异步执行库存扣减和订单生成
 * 注意：令牌在极端情况下，如扣减令牌之后，服务宕机了，此时令牌的库存是小于真实库存的
 * 如果查询令牌发现库存为0，尝试去数据库中加载数据，加载之后库存还是0，说明时间段确实售罄了
 * 使用消息队列异步 扣减库存，更新缓存，生成订单
 *
 * @param timePeriodId
 */
@Override
public void reserve2(Long timePeriodId) {
     参数校验：使用责任链模式校验数据是否正确
    TimePeriodReserveReqDTO timePeriodReserveReqDTO = new TimePeriodReserveReqDTO(timePeriodId);
    chainContext.handler(ChainConstant.RESERVE_CHAIN_NAME, timePeriodReserveReqDTO);
    TimePeriodDO timePeriodDO = timePeriodReserveReqDTO.getTimePeriodDO();
    Long venueId = timePeriodReserveReqDTO.getVenueId();
    VenueDO venueDO = timePeriodReserveReqDTO.getVenueDO();
    PartitionDO partitionDO = partitionService.getPartitionDOById(timePeriodDO.getPartitionId());

     使用lua脚本获取一个空场地对应的索引，并扣除相应的库存，同时在里面进行用户的查重
    // 首先检测空闲场号缓存有没有加载好，没有的话进行加载
    this.checkBitMapCache(
            String.format(RedisCacheConstant.VENUE_TIME_PERIOD_FREE_INDEX_BIT_MAP_TOKEN_KEY, timePeriodReserveReqDTO.getTimePeriodId()),
            timePeriodId,
            partitionDO.getNum());
    // 其次检测时间段库存有没有加载好，没有的话进行加载
    this.getStockByTimePeriodId(RedisCacheConstant.VENUE_TIME_PERIOD_STOCK_TOKEN_KEY, timePeriodReserveReqDTO.getTimePeriodId());
    // 执行lua脚本
    Long freeCourtIndex = executeStockReduceByLua(
            timePeriodReserveReqDTO,
            venueDO,
            RedisCacheConstant.VENUE_TIME_PERIOD_STOCK_TOKEN_KEY,
            RedisCacheConstant.VENUE_TIME_PERIOD_FREE_INDEX_BIT_MAP_TOKEN_KEY);
    if (freeCourtIndex == -2L) {
        // --if-- 用户已经购买过该时间段
        throw new ClientException(BaseErrorCode.TIME_PERIOD_HAVE_BOUGHT_ERROR);
    } else if (freeCourtIndex == -1L) {
        // --if-- 没有空闲的场号，查询数据库，如果数据库中有库存，删除缓存，下一个用户预定时重新加载令牌
        this.refreshTokenByCheckDatabase(timePeriodId);
        throw new ServiceException(BaseErrorCode.TIME_PERIOD_SELL_OUT_ERROR);
    }

     发送消息，异步更新库存并生成订单
    SendResult sendResult = executeReserveProducer.sendMessage(ExecuteReserveMqDTO.builder()
            .timePeriodId(timePeriodId)
            .freeCourtIndex(freeCourtIndex)
            .venueId(venueId)
            .userId(UserContext.getUserId())
            .userName(UserContext.getUsername())
            .build());
    if (!sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
        log.error("消息发送失败: " + sendResult.getSendStatus());
        // 恢复令牌缓存
        this.restoreStockAndBookedSlotsCache(
                timePeriodId,
                UserContext.getUserId(),
                freeCourtIndex,
                RedisCacheConstant.VENUE_TIME_PERIOD_STOCK_TOKEN_KEY,
                RedisCacheConstant.VENUE_TIME_PERIOD_FREE_INDEX_BIT_MAP_TOKEN_KEY);
        throw new ServiceException(BaseErrorCode.MQ_SEND_ERROR);
    }
}

【获取令牌】

获取令牌的过程其实就是检验用户是否重新预订、库存数量检查、场号分配、库存扣减、场号占用，这里和接口V1的实现是一样的

/**
 * 使用lua脚本，进行缓存中的库存扣减，并分配空闲场号
 *
 * @param timePeriodReserveReqDTO
 * @param venueDO
 * @param stockKey
 * @param freeIndexBitMapKey
 * @return
 */
private Long executeStockReduceByLua(TimePeriodReserveReqDTO timePeriodReserveReqDTO, VenueDO venueDO,
                                     String stockKey, String freeIndexBitMapKey) {
    // 使用 Hutool 的单例管理容器 管理lua脚本的加载，保证其只被加载一次
    String luaScriptPath = "lua/free_court_index_allocate_by_bitmap.lua";
    DefaultRedisScript<Long> luaScript = Singleton.get(luaScriptPath, () -> {
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource(luaScriptPath)));
        redisScript.setResultType(Long.class);
        return redisScript;
    });
    // 执行用户重复预定校验、库存扣减、场号分配
    Long freeCourtIndex = stringRedisTemplate.execute(
            luaScript,
            Lists.newArrayList(
                    String.format(stockKey, timePeriodReserveReqDTO.getTimePeriodId()),
                    String.format(freeIndexBitMapKey, timePeriodReserveReqDTO.getTimePeriodId()),
                    String.format(RedisCacheConstant.VENUE_IS_USER_BOUGHT_TIME_PERIOD_KEY, timePeriodReserveReqDTO.getTimePeriodId())
            ),
            UserContext.getUserId().toString(),
            String.valueOf(venueDO.getAdvanceBookingDay() * 86400)
    );
    return freeCourtIndex;
}

lua

-- 定义脚本参数
local stock_key = KEYS[1]
local free_index_bitmap_key = KEYS[2]
-- 用来存储已购买用户的set
local set_name = KEYS[3]

-- 用户ID
local user_id = ARGV[1]
-- 过期时间 (秒)
local expire_time = tonumber(ARGV[2])

-- 检查用户是否已经购买过
if redis.call("SISMEMBER", set_name, user_id) == 1 then
    -- 用户已经购买过，返回 -2 表示失败
    return -2
end

-- 获取库存
local current_inventory = tonumber(redis.call('GET', stock_key) or 0)

-- 尝试消耗库存
if current_inventory < 1 then
    -- 库存不够了，返回-1，代表分配空场号失败
    return -1 -- 失败
end

-- 查找第一个空闲的场地（位图中第一个为 0 的位）
local free_court_bit = redis.call("BITPOS", free_index_bitmap_key, 0)

if not free_court_bit or free_court_bit == -1 then
    -- 没有空闲的场号
    return -1 -- 失败
end

-- 占用该场地（将对应位设置为 1）
redis.call("SETBIT", free_index_bitmap_key, free_court_bit, 1)
-- 更新库存
redis.call('DECRBY', stock_key, 1)
-- 添加用户到已购买集合
redis.call("SADD", set_name, user_id)
-- 设置过期时间
if expire_time > 0 then
    redis.call("EXPIRE", set_name, expire_time)
end

-- 返回分配的场地索引（注意：位图的位索引从0开始，如果你需要从1开始，这里加1）
return tonumber(free_court_bit)

【更新缓存中库存】

大家可能会疑问，为啥有了令牌，还要更新缓存中的库存和空闲场号。因为我们在前端展示的信息需要是真实的库存信息，为了加速查询，需要将库存缓存起来，这里的缓存数据需要和数据库一致。为了保证缓存和数据库的最终一致性，可以开启 binlog ，然后使用 Canal 进行监听。如果数据库中的数据更新了，就发送消息到消息队列中，消费消息时再更新缓存中的库存。

-- 定义脚本参数
local stock_key = KEYS[1]
local free_index_bitmap_key = KEYS[2]

-- 预订场号
local free_court_bit = ARGV[1]

-- 占用该场地（将对应位设置为 1）
redis.call("SETBIT", free_index_bitmap_key, free_court_bit, 1)
-- 更新库存
redis.call('DECRBY', stock_key, 1)

return 0

【检测和加载位图缓存】

/**
 * 检测位图缓存是否加载好，没有的话，执行加载操作
 *
 * @param freeIndexBitmapKey
 * @param timePeriodId
 * @param initStock
 */
@Override
public void checkBitMapCache(String freeIndexBitmapKey, Long timePeriodId, int initStock) {
    String cache = stringRedisTemplate.opsForValue().get(freeIndexBitmapKey);
    if (StringUtils.isBlank(cache)) {
        // --if-- 如果缓存中的位图为空
        RLock lock = redissonClient.getLock(String.format(RedisCacheConstant.VENUE_LOCK_TIME_PERIOD_FREE_INDEX_BIT_MAP_KEY, timePeriodId));
        lock.lock();
        try {
            // 双重判定一下，避免其他线程已经加载数据到缓存中了
            cache = stringRedisTemplate.opsForValue().get(freeIndexBitmapKey);
            if (StringUtils.isBlank(cache)) {
                // --if-- 如果缓存中的位图还是空，到数据库中加载位图
                TimePeriodDO timePeriodDO = this.getById(timePeriodId);
                if (timePeriodDO == null) {
                    throw new ServiceException(timePeriodId + "对应的时间段为null", BaseErrorCode.SERVICE_ERROR);
                }
                // 将位图信息设置到缓存中
                this.initializeFreeIndexBitmap(freeIndexBitmapKey, initStock, timePeriodDO.getBookedSlots(), 24 * 3600);
            }
        } finally {
            // 解锁
            lock.unlock();
        }
    }
}

/**
 * 初始化Redis中的位图，并设置key的过期时间
 *
 * @param freeIndexBitmapKey 位图的键名
 * @param longValue          用于初始化位图的 long 类型数据
 * @param expireSecond       key的过期时间（秒）
 */
public void initializeFreeIndexBitmap(String freeIndexBitmapKey, int initStock, long longValue, long expireSecond) {
    // 将 long 转换为64位的二进制字符串
    String binaryString = Long.toBinaryString(longValue);
    // 确保字符串长度为64位，不足的部分用0补齐
    binaryString = String.format("%64s", binaryString).replace(' ', '0');

    // 从低位到高位遍历二进制字符串，设置位图中的对应位
    for (int i = 0; i < 64 && initStock-- >= 0; i++) {
        // 注意：long的最低位对应位图的第0位
        if (binaryString.charAt(63 - i) == '1') {
            stringRedisTemplate.opsForValue().setBit(freeIndexBitmapKey, i, true).booleanValue();
        } else {
            stringRedisTemplate.opsForValue().setBit(freeIndexBitmapKey, i, false).booleanValue();
        }
    }

    // 设置过期时间，仅当expireTime大于0时进行设置
    if (expireSecond > 0) {
        stringRedisTemplate.expire(freeIndexBitmapKey, expireSecond, TimeUnit.SECONDS);
    }
}

【检验和加载库存缓存】

这里使用了封装的缓存组件，需要去仓库查看详细代码

/**
 * 获取指定时间段的库存
 *
 * @param timePeriodId
 * @return
 */
@Override
public Integer getStockByTimePeriodId(Long timePeriodId) {
    return (Integer) distributedCache.safeGet(
            String.format(RedisCacheConstant.VENUE_TIME_PERIOD_STOCK_KEY, timePeriodId),
            new TypeReference<Integer>() {
            },
            () -> {
                TimePeriodDO timePeriodDO = this.getById(timePeriodId);
                return timePeriodDO.getStock();
            },
            1,
            TimeUnit.DAYS);
}

【消费消息，执行预订流程】

和接口 V1 不同的是，V1 时同步创建订单，创建完成之后，直接访问给用户订单数据。但是在 V2 中，将任务交给消息队列之后，就要返回成功了。用户需要在前端等待订单创建结果。那前端如何感知订单是否创建成功呢？

方式一：前端轮询查询后端，如每隔一秒问一下后端，订单创建好没有，创建好了就返回给前端，这样前端就可以进行支付了
方式二：使用前后端双向通讯技术，如WebSocket。前后端一开始先建立好连接，等后端消费消息，创建订单成功之后，直接将订单信息推送给前端

/**
 * 通过消息队列执行 时间段预定 逻辑
 * @param executeReserveMqDTO
 */
@Override
public void mqExecutePreserve(ExecuteReserveMqDTO executeReserveMqDTO) {
    TimePeriodDO timePeriodDO = this.getTimePeriodDOById(executeReserveMqDTO.getTimePeriodId());
    // 编程式开启事务，减少事务粒度，避免长事务的发生
    transactionTemplate.executeWithoutResult(status -> {
        try {
            // 扣减当前时间段的库存，修改空闲场信息
            baseMapper.updateStockAndBookedSlots(timePeriodDO.getId(), timePeriodDO.getPartitionId(), executeReserveMqDTO.getTimePeriodId());
            // 更新缓存中的库存、位图
            if (!isUseBinlog) {
                // --if-- 如果不使用binlog，需要手动更新缓存

                // 首先检测空闲场号缓存有没有加载好，没有的话进行加载
                this.checkBitMapCache(
                        String.format(RedisCacheConstant.VENUE_TIME_PERIOD_FREE_INDEX_BIT_MAP_KEY, executeReserveMqDTO.getTimePeriodId()),
                        executeReserveMqDTO.getTimePeriodId(),
                        partitionService.getPartitionDOById(timePeriodDO.getPartitionId()).getNum());
                // 其次检测时间段库存有没有加载好，没有的话进行加载
                this.getStockByTimePeriodId(executeReserveMqDTO.getTimePeriodId());
                // 使用 Hutool 的单例管理容器 管理lua脚本的加载，保证其只被加载一次
                String luaScriptPath = "lua/inventory_update.lua";
                DefaultRedisScript<Long> luaScript = Singleton.get(luaScriptPath, () -> {
                    DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
                    redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource(luaScriptPath)));
                    redisScript.setResultType(Long.class);
                    return redisScript;
                });
                // 库存扣减、场号占用
                stringRedisTemplate.execute(
                        luaScript,
                        Lists.newArrayList(
                                String.format(RedisCacheConstant.VENUE_TIME_PERIOD_STOCK_KEY, executeReserveMqDTO.getTimePeriodId()),
                                String.format(RedisCacheConstant.VENUE_TIME_PERIOD_FREE_INDEX_BIT_MAP_KEY, executeReserveMqDTO.getTimePeriodId())
                        ),
                        executeReserveMqDTO.getFreeCourtIndex().toString()
                );
            }
            // todo 需要实现binlog版本

            // 调用远程服务创建订单
            OrderGenerateReqDTO orderGenerateReqDTO = OrderGenerateReqDTO.builder()
                    .timePeriodId(timePeriodDO.getId())
                    .partitionId(timePeriodDO.getPartitionId())
                    .periodDate(timePeriodDO.getPeriodDate())
                    .beginTime(timePeriodDO.getBeginTime())
                    .endTime(timePeriodDO.getEndTime())
                    .courtIndex(executeReserveMqDTO.getFreeCourtIndex())
                    .userId(executeReserveMqDTO.getUserId())
                    .userName(executeReserveMqDTO.getUserName())
                    .venueId(executeReserveMqDTO.getVenueId())
                    .payAmount(timePeriodDO.getPrice())
                    .build();

            Result<OrderDO> result;
            try {
                result = orderFeignService.generateOrder(orderGenerateReqDTO);
                if (result == null || !result.isSuccess()) {
                    // --if-- 订单生成失败，抛出异常，上面的库存扣减也会回退
                    throw new ServiceException(BaseErrorCode.ORDER_GENERATE_ERROR);
                }
            } catch (Exception e) {
                // --if-- 订单生成服务调用失败
                // 恢复缓存中的信息
                this.restoreStockAndBookedSlotsCache(
                        timePeriodDO.getId(),
                        1L,
                        executeReserveMqDTO.getFreeCourtIndex(),
                        RedisCacheConstant.VENUE_TIME_PERIOD_STOCK_TOKEN_KEY,
                        RedisCacheConstant.VENUE_TIME_PERIOD_FREE_INDEX_BIT_MAP_TOKEN_KEY);
                // todo 如果说由于网络原因，实际上订单已经创建成功了，但是因为超时访问失败，这里库存却回滚了，此时需要将订单置为废弃状态（即删除）
                // 发送一个短暂的延时消息（时间过长，用户可能已经支付），去检查订单是否生成，如果生成，将其删除
                // 打印错误堆栈信息
                e.printStackTrace();
                // 把错误返回到前端
                throw new ServiceException(e.getMessage());
            }
            OrderDO orderDO = result.getData();
            // todo 使用 WebSocket 通知前端，订单生成成功
        } catch (Exception ex) {
            status.setRollbackOnly();
            throw ex;
        }
    });
}

MQ

生产者

import cn.hutool.core.util.StrUtil;
import com.vrs.constant.RocketMqConstant;
import com.vrs.domain.dto.mq.ExecuteReserveMqDTO;
import com.vrs.templateMethod.AbstractCommonSendProduceTemplate;
import com.vrs.templateMethod.BaseSendExtendDTO;
import com.vrs.templateMethod.MessageWrapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * 执行预订流程 生产者
 *
 * @Author dam
 * @create 2024/9/20 16:00
 */
@Slf4j
@Component
public class ExecuteReserveProducer extends AbstractCommonSendProduceTemplate<ExecuteReserveMqDTO> {

    @Override
    protected BaseSendExtendDTO buildBaseSendExtendParam(ExecuteReserveMqDTO messageSendEvent) {
        return BaseSendExtendDTO.builder()
                .eventName("执行时间段预定")
                .keys(String.valueOf(messageSendEvent.getTimePeriodId()))
                .topic(RocketMqConstant.VENUE_TOPIC)
                .tag(RocketMqConstant.TIME_PERIOD_EXECUTE_RESERVE_TAG)
                .sentTimeout(2000L)
                .build();
    }

    @Override
    protected Message<?> buildMessage(ExecuteReserveMqDTO messageSendEvent, BaseSendExtendDTO requestParam) {
        String keys = StrUtil.isEmpty(requestParam.getKeys()) ? UUID.randomUUID().toString() : requestParam.getKeys();
        return MessageBuilder
                .withPayload(new MessageWrapper(keys, messageSendEvent))
                .setHeader(MessageConst.PROPERTY_KEYS, keys)
                .setHeader(MessageConst.PROPERTY_TAGS, requestParam.getTag())
                .build();
    }
}

消费者

import com.vrs.annotation.Idempotent;
import com.vrs.constant.RocketMqConstant;
import com.vrs.domain.dto.mq.ExecuteReserveMqDTO;
import com.vrs.enums.IdempotentSceneEnum;
import com.vrs.service.TimePeriodService;
import com.vrs.templateMethod.MessageWrapper;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

/**
 * 执行预订流程 消费者
 * @Author dam
 * @create 2024/9/20 21:30
 */
@Slf4j(topic = RocketMqConstant.VENUE_TOPIC)
@Component
@RocketMQMessageListener(topic = RocketMqConstant.VENUE_TOPIC,
        consumerGroup = RocketMqConstant.VENUE_CONSUMER_GROUP + "-" + RocketMqConstant.TIME_PERIOD_EXECUTE_RESERVE_TAG,
        messageModel = MessageModel.CLUSTERING,
        // 监听tag
        selectorType = SelectorType.TAG,
        selectorExpression = RocketMqConstant.TIME_PERIOD_EXECUTE_RESERVE_TAG
)
@RequiredArgsConstructor
public class ExecuteReserveListener implements RocketMQListener<MessageWrapper<ExecuteReserveMqDTO>> {

    private final TimePeriodService timePeriodService;

    /**
     * 消费消息的方法
     * 方法报错就会拒收消息
     *
     * @param messageWrapper 消息内容，类型和上面的泛型一致。如果泛型指定了固定的类型，消息体就是我们的参数
     */
    @Idempotent(
            uniqueKeyPrefix = "time_period_execute_reserve:",
            key = "#messageWrapper.getMessage().getTimePeriodId()+''",
            scene = IdempotentSceneEnum.MQ,
            keyTimeout = 3600L
    )
    @SneakyThrows
    @Override
    public void onMessage(MessageWrapper<ExecuteReserveMqDTO> messageWrapper) {
        // 开头打印日志，平常可 Debug 看任务参数，线上可报平安（比如消息是否消费，重新投递时获取参数等）
        log.info("[消费者] 执行时间段预定，时间段ID：{}", messageWrapper.getMessage().getTimePeriodId());
        timePeriodService.mqExecutePreserve(messageWrapper.getMessage());
    }
}

测试结果

在这里插入图片描述

样本数量：共有40,000个样本，这表示在测试期间进行了40,000次操作或请求。
响应时间：
1. 平均值：1801毫秒，表示所有请求的平均响应时间。
2. 中位数：1346毫秒，表示50%的请求响应时间低于这个值。
3. 90%百分位：2048毫秒，表示90%的请求响应时间低于这个值。
4. 95%百分位：3410毫秒，表示95%的请求响应时间低于这个值。
5. 99%百分位：15133毫秒，表示99%的请求响应时间低于这个值。
6. 最小值：15毫秒，表示最快的请求响应时间。
7. 最大值：22121毫秒，表示最慢的请求响应时间。
异常率：0.22%，表示在所有请求中，有0.22%的请求出现了异常。
吞吐量：每秒可以处理1045.8个请求
网络流量：
1. 接收速率：221.51 KB/sec，表示系统每秒接收的数据量。
2. 发送速率：509.96 KB/sec，表示系统每秒发送的数据量。

总结：

系统的平均响应时间为1801毫秒，中位数为1346毫秒，表明大多数请求的响应时间在可接受范围内。
99%的请求响应时间在15133毫秒以内，但有少数请求的响应时间较长，最大值达到了22121毫秒。
系统的吞吐量为1045.8次请求/秒，处理能力较高，相较于接口V1，性能强了一倍

说明

文章内容并非最新代码实现，若需要知道最新实现，麻烦移步开源仓库： HelloDam/场快订（场馆预定 SaaS 平台）

原文地址：https://blog.csdn.net/laodanqiu/article/details/145281039

免责声明：本站文章内容转载自网络资源，如侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：centos9编译安装opensips 二【进阶篇-定制目录+模块】
下一篇：【优选算法】4----盛最多水的容器

ubuntu24.04安装mysql 5.7.44版本
使用下面命令，查看配置信息是否，这一步非常重要，配置文件不生效，会导致系统区分大小写表。创建该文件，退出的时候，就算不输入内容，也要强制保存一下 :wq!命令来更新init系统，以便在系统启动时自动启
阅读更多2025-01-22
SQLmap 自动注入 -02
SQLmap 是SQL注入的超级工具，这次参数是 -dbs, 还要-users, --current-user
阅读更多2025-01-22
AI大模型-提示工程学习笔记12-检索增强生成(RAG)
是一种结合检索（Retrieval）和生成（Generation）的技术，旨在通过从外部知识库中检索相关信息，增强大语言模型（LLM）的生成能力。尽管 RAG 面临一些挑战（如检索质量、知识库维护成本
阅读更多2025-01-22
在亚马逊云科技上用AI提示词优化功能写出漂亮提示词（下）
提示工程（Prompt Engineering）对各位小伙伴们来说是再熟悉不过了，提示词工程技术是通过编写指令词，指导开发者们调用AI基础模型（FMs）获得期望的响应。但是经常写提示词的朋友们会知道，
阅读更多2025-01-22
他把智能科技引入现代农业领域
江苏田倍丰农业科技有限公司（以下简称“田倍丰”）是一家专注于粮油种植的农业科技公司，为拥有300亩以上田地的大户提供全面的解决方案。田倍丰通过与当地政府合作，将土地承包给大户，并提供农资和技术，实现利
阅读更多2025-01-22
深圳市云盟智慧科技有限公司智慧停车管理系统 SQL注入漏洞复现（附脚本）
深圳市云盟智慧科技有限公司智慧停车管理系统 /GetVideo 接口存在SQL注入漏洞，未经身份验证的恶意攻击者利用 SQL 注入漏洞获取数据库中的信息（例如管理员后台密码、站点用户个人信息）之外，攻
阅读更多2025-01-22
江天科技主要产品销售单价下滑，应收账款、存货周转率大幅下降
报告期各期，公司薄膜类不干胶标签实现营收分别为2.37亿、2.75亿、3.90亿、2.21亿，营收占比分别为68.20%、71.69%、76.77%、79.23%；此外，报告期各期，江天科技毛利率分别
阅读更多2025-01-22
PAT乙级真题 — 1060 爱丁顿数（java）
英国天文学家爱丁顿很喜欢骑车。据说他为了炫耀自己的骑车功力，还定义了一个“爱丁顿数” E ，即满足有 E 天骑车超过 E 英里的最大整数 E。据说爱丁顿自己的 E 等于87。现给定某人 N 天的骑车距
阅读更多2025-01-22
AutoSAR CP RTE 规范核心内容简介以及BswScheduler工作原理解析
本规范详细介绍了AUTOSAR运行时环境（RTE）和基本软件调度器（BswScheduler）的软件规范。RTE概述：AUTOSAR概念：RTE生成过程：RTE功能规范：RTE与基本软件调度器的交互：
阅读更多2025-01-22
UE5 开启“Python Remote Execution“
在启用 Unreal Engine 5（UE5）的“Python 远程执行”功能后，UE5 会启动一个 UDP 组播套接字服务，以监听来自外部应用程序的 Python 命令。通过这种方式，UE5 实现
阅读更多2025-01-22