案例分享：使用RabbitMQ消息队列和Redis缓存优化Spring Boot秒杀功能

news/2024/11/17 1:44:20/

作者介绍：✌️大厂全栈码农|毕设实战开发，专注于大学生项目实战开发、讲解和毕业答疑辅导。

 推荐订阅精彩专栏 👇🏻 避免错过下次更新

Springboot项目精选实战案例

更多项目：CSDN主页YAML墨韵

学如逆水行舟，不进则退。学习如赶路，不能慢一步。

目录

1、描述

2、pom.xml文件

3、创建redis工具类

4、创建rabbitmq的配置类

5、创建数据库表

用户信息 

商品信息 

秒杀信息

订单信息

秒杀订单 

6、具体实现过程

用户登录

秒杀商品数量初始化

rabbitmq队列

生产者代码

消费者代码

订单模块

注意事项：

秒杀功能作为大型交易平台的常见活动，落地实现的时候需要应对大量并发请求，同时保证请求的快速、准确处理。本文通过案例分析，讲解如何结合Springboot 3和RabbitMQ和Redis来构建秒杀请求的异步处理队列，并通过性能测试对异步处理方案进行优化。

1、描述

交易平台秒杀功能的业务流程

秒杀活动一般发生在一些特定的时间点，如节日特卖或者是限量产品的销售。这样的活动通常会吸引大批用户的参与。由于参与的用户量大，再加上秒杀商品的数量有限，因此这种活动对后端系统的架构设计提出了很大挑战。一般来说，秒杀活动的流程可以分为以下几个步骤：

  1. 秒杀宣传与倒计时：商家通过广告和营销方式进行秒杀活动的宣传，并在秒杀页面显示一个倒计时，告知用户秒杀活动开始的时间。

  2. 用户抢单：一般来说，用户需要在秒杀页面上点击“立即秒杀”按钮，发起秒杀请求。

  3. 系统处理秒杀请求：由于秒杀活动瞬间会产生大量用户请求，所以系统要有相应的优化措施。这里我们采用的是异步处理的方式。具体来说，就是当用户发起秒杀请求后，实际上用户的请求被发送到RabbitMQ的消息队列中，然后通过后台服务进行异步处理。

  4. 检查库存：在进行后续操作之前，系统会检查当前商品的库存量，以确保没有超卖。

  5. 扣减库存并生成订单：这是一个原子操作，即系统需要在一次操作中完成库存扣减和订单生成。也就是说，当我们更新库存数量的同时，也会在订单表中创建新的订单记录。

  6. 支付处理：成功生成订单的用户被引导至支付页面进行付款操作。支付通常有一定的时间限制，如果超过时间未支付，订单会被自动取消，并将库存加回。

  7. 订单处理：用户成功支付后，系统会进行后续的订单处理工作，比如发货等。

以上是一个通用的秒杀活动业务流程。实际上，在面对大流量的情况下，需要利用各种手段进行优化，以保证服务的稳定性和用户的体验，比如引入消息队列进行异步处理，使用缓存等。具体的优化手段，还需要根据业务场景和系统状况灵活选择和实施。

2、pom.xml文件

RabbitMQ的消息队列配置请看：Spring Boot与RabbitMQ整合：实现高可用消息队列服务

redis的基本配置和实战请看：Spring Boot与Redis深度整合：实战指南

我们需要在SpringBoot应用中集成RabbitMQ和Redis。我们在pom.xml文件中添加依赖：

<dependencies>  <!-- Spring Boot Starter for AMQP (RabbitMQ) -->  <dependency>  <groupId>org.springframework.boot</groupId>  <artifactId>spring-boot-starter-amqp</artifactId>  </dependency>  <!-- Spring Boot Starter for Redis -->  <dependency>  <groupId>org.springframework.boot</groupId>  <artifactId>spring-boot-starter-data-redis</artifactId>  </dependency>  <!-- 其他依赖... -->  
</dependencies>

在application.yml中配置RabbitMQ和Redis：

spring:  rabbitmq:  host: your-rabbitmq-host  port: 5672  username: guest  password: guest  redis:  host: your-redis-host  port: 6379  password: your-redis-password # 如果设置了密码的话

3、创建redis工具类

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;@Component
public class RedisUtil {@Autowiredprivate RedisTemplate<String, Object> redisTemplate;/*** 指定缓存失效时间* @param key  键* @param time 时间(秒)*/public boolean expire(String key, long time) {try {if (time > 0) {redisTemplate.expire(key, time, TimeUnit.SECONDS);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 根据key 获取过期时间* @param key 键 不能为null* @return 时间(秒) 返回0代表为永久有效*/public long getExpire(String key) {return redisTemplate.getExpire(key, TimeUnit.SECONDS);}/*** 判断key是否存在* @param key 键* @return true 存在 false不存在*/public boolean hasKey(String key) {try {return redisTemplate.hasKey(key);} catch (Exception e) {e.printStackTrace();return false;}}/*** 删除缓存* @param key 可以传一个值 或多个*/@SuppressWarnings("unchecked")public void del(String... key) {if (key != null && key.length > 0) {if (key.length == 1) {redisTemplate.delete(key[0]);} else {redisTemplate.delete((Collection<String>) CollectionUtils.arrayToList(key));}}}// ============================String=============================/*** 普通缓存获取* @param key 键* @return 值*/public Object get(String key) {return key == null ? null : redisTemplate.opsForValue().get(key);}/*** 普通缓存放入* @param key   键* @param value 值* @return true成功 false失败*/public boolean set(String key, Object value) {try {redisTemplate.opsForValue().set(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 普通缓存放入并设置时间* @param key   键* @param value 值* @param time  时间(秒) time要大于0 如果time小于等于0 将设置无限期* @return true成功 false 失败*/public boolean set(String key, Object value, long time) {try {if (time > 0) {redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);} else {set(key, value);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 递增* @param key   键* @param delta 要增加几(大于0)*/public long incr(String key, long delta) {if (delta < 0) {throw new RuntimeException("递增因子必须大于0");}return redisTemplate.opsForValue().increment(key, delta);}/*** 递减* @param key   键* @param delta 要减少几(小于0)*/public long decr(String key, long delta) {if (delta < 0) {throw new RuntimeException("递减因子必须大于0");}return redisTemplate.opsForValue().increment(key, -delta);}// ================================Map=================================/*** HashGet* @param key  键 不能为null* @param item 项 不能为null*/public Object hget(String key, String item) {return redisTemplate.opsForHash().get(key, item);}/*** 获取hashKey对应的所有键值* @param key 键* @return 对应的多个键值*/public Map<Object, Object> hmget(String key) {return redisTemplate.opsForHash().entries(key);}/*** HashSet* @param key 键* @param map 对应多个键值*/public boolean hmset(String key, Map<String, Object> map) {try {redisTemplate.opsForHash().putAll(key, map);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** HashSet 并设置时间* @param key  键* @param map  对应多个键值* @param time 时间(秒)* @return true成功 false失败*/public boolean hmset(String key, Map<String, Object> map, long time) {try {redisTemplate.opsForHash().putAll(key, map);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 向一张hash表中放入数据,如果不存在将创建** @param key   键* @param item  项* @param value 值* @return true 成功 false失败*/public boolean hset(String key, String item, Object value) {try {redisTemplate.opsForHash().put(key, item, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 向一张hash表中放入数据,如果不存在将创建** @param key   键* @param item  项* @param value 值* @param time  时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间* @return true 成功 false失败*/public boolean hset(String key, String item, Object value, long time) {try {redisTemplate.opsForHash().put(key, item, value);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 删除hash表中的值** @param key  键 不能为null* @param item 项 可以使多个 不能为null*/public void hdel(String key, Object... item) {redisTemplate.opsForHash().delete(key, item);}/*** 判断hash表中是否有该项的值** @param key  键 不能为null* @param item 项 不能为null* @return true 存在 false不存在*/public boolean hHasKey(String key, String item) {return redisTemplate.opsForHash().hasKey(key, item);}/*** hash递增 如果不存在,就会创建一个 并把新增后的值返回** @param key  键* @param item 项* @param by   要增加几(大于0)*/public double hincr(String key, String item, double by) {return redisTemplate.opsForHash().increment(key, item, by);}/*** hash递减** @param key  键* @param item 项* @param by   要减少记(小于0)*/public double hdecr(String key, String item, double by) {return redisTemplate.opsForHash().increment(key, item, -by);}// ============================set=============================/*** 根据key获取Set中的所有值* @param key 键*/public Set<Object> sGet(String key) {try {return redisTemplate.opsForSet().members(key);} catch (Exception e) {e.printStackTrace();return null;}}/*** 根据value从一个set中查询,是否存在** @param key   键* @param value 值* @return true 存在 false不存在*/public boolean sHasKey(String key, Object value) {try {return redisTemplate.opsForSet().isMember(key, value);} catch (Exception e) {e.printStackTrace();return false;}}/*** 将数据放入set缓存** @param key    键* @param values 值 可以是多个* @return 成功个数*/public long sSet(String key, Object... values) {try {return redisTemplate.opsForSet().add(key, values);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 将set数据放入缓存** @param key    键* @param time   时间(秒)* @param values 值 可以是多个* @return 成功个数*/public long sSetAndTime(String key, long time, Object... values) {try {Long count = redisTemplate.opsForSet().add(key, values);if (time > 0)expire(key, time);return count;} catch (Exception e) {e.printStackTrace();return 0;}}/*** 获取set缓存的长度** @param key 键*/public long sGetSetSize(String key) {try {return redisTemplate.opsForSet().size(key);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 移除值为value的** @param key    键* @param values 值 可以是多个* @return 移除的个数*/public long setRemove(String key, Object... values) {try {Long count = redisTemplate.opsForSet().remove(key, values);return count;} catch (Exception e) {e.printStackTrace();return 0;}}// ===============================list=================================/*** 获取list缓存的内容** @param key   键* @param start 开始* @param end   结束 0 到 -1代表所有值*/public List<Object> lGet(String key, long start, long end) {try {return redisTemplate.opsForList().range(key, start, end);} catch (Exception e) {e.printStackTrace();return null;}}/*** 获取list缓存的长度** @param key 键*/public long lGetListSize(String key) {try {return redisTemplate.opsForList().size(key);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 通过索引 获取list中的值** @param key   键* @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推*/public Object lGetIndex(String key, long index) {try {return redisTemplate.opsForList().index(key, index);} catch (Exception e) {e.printStackTrace();return null;}}/*** 将list放入缓存** @param key   键* @param value 值*/public boolean lSet(String key, Object value) {try {redisTemplate.opsForList().rightPush(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存* @param key   键* @param value 值* @param time  时间(秒)*/public boolean lSet(String key, Object value, long time) {try {redisTemplate.opsForList().rightPush(key, value);if (time > 0)expire(key, time);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存** @param key   键* @param value 值* @return*/public boolean lSet(String key, List<Object> value) {try {redisTemplate.opsForList().rightPushAll(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存** @param key   键* @param value 值* @param time  时间(秒)* @return*/public boolean lSet(String key, List<Object> value, long time) {try {redisTemplate.opsForList().rightPushAll(key, value);if (time > 0)expire(key, time);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 根据索引修改list中的某条数据** @param key   键* @param index 索引* @param value 值* @return*/public boolean lUpdateIndex(String key, long index, Object value) {try {redisTemplate.opsForList().set(key, index, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 移除N个值为value** @param key   键* @param count 移除多少个* @param value 值* @return 移除的个数*/public long lRemove(String key, long count, Object value) {try {Long remove = redisTemplate.opsForList().remove(key, count, value);return remove;} catch (Exception e) {e.printStackTrace();return 0;}}
}

4、创建rabbitmq的配置类


@Configuration
public class TopicConfig {@Beanpublic TopicExchange topicExchange() {return new TopicExchange("seckill_topic", true, false);}@Beanpublic Queue seckillQueue() {return new Queue("seckillQueue", true);}@Beanpublic Binding binding() {return BindingBuilder.bind(seckillQueue()).to(topicExchange()).with("seckill.#");}
}

5、创建数据库表

用户信息 

商品信息 

秒杀信息

订单信息

秒杀订单 

6、具体实现过程

用户登录
将要秒杀的商品数量预存在redis中
接收前端传回来的用户id，被秒杀商品id
验证用户是否为当前用户，是则进行下一步
获取被秒杀的商品信息
判断用户是否成功秒杀过商品，如果秒杀过，则结束进程，否则下一步
判断被秒杀商品是否还有库存，有，下一步，否则结束进程
将用户id和被秒杀商品信息传入队列进行处理
队列处理完成后，返回结果。

    @ApiModelProperty(value = "秒杀功能")@PostMapping("doseckill")public Result doSeckill(@RequestParam Long userid, @RequestParam Long goodsid) {//判断是否当前用户RUser user = (RUser) redisUtil.get("user" + userid + ":");if (user == null) {return Result.fail("请登录！");}//缓存被秒杀商品的信息if (redisUtil.get("g" + goodsid + ":") == null) {GoodsVo goods = goodsService.selectGoodsById(goodsid);redisUtil.set("g" + goodsid + ":", goods);}GoodsVo goods = (GoodsVo) redisUtil.get("g" + goodsid + ":");//判断是否重复操作，已经成功秒杀，不能再次秒杀
//        SeckillOrders seckillOrders = seckillOrdersService.getOne(new QueryWrapper<SeckillOrders>()
//                .lambda().eq(SeckillOrders::getUserId, userid)
//                .eq(SeckillOrders::getGoodId, goods.getId()));Object o = redisUtil.get("order:" + userid + goodsid);if (o != null) {return Result.fail("用户" + userid + ":每个用户只能秒杀一次.....");}//库存预减Long count = redisUtil.decr("seckill" + goodsid + ":", 1);if (count < 0) {redisUtil.incr("seckill" + goodsid + ":", 1);return Result.fail("用户" + userid + ":存库不足，秒杀失败.....");}//将秒杀请求添加到队列中MqMessage message = new MqMessage(userid, goods);String msg = JSON.toJSONString(message);product.sendSeckillMessage(msg);return Result.succ("秒杀中............");}

用户登录

校验账号和密码是否正确，两者都正确则登陆成功，否则登陆失败。如果登录成功，就将用户的信息缓存在redis中，后续如果需要使用到用户的信息，直接从redis中获取即可，不用再次访问数据库。


public RUser Login(UserDto userDto) {RUser user = new RUser();BeanUtils.copyProperties(userDto,user);QueryWrapper<RUser> qw = new QueryWrapper<>();qw.lambda().eq(RUser::getId,user.getId()).eq(RUser::getPassword,user.getPassword());RUser user1 =baseMapper.selectOne(qw);if(user1!=null){redisUtil.set("user"+user1.getId()+":",user1);System.out.println(redisUtil.set("user"+user1.getId()+":",user1));}return user1;
}

秒杀功能的流程是：获得秒杀资格后，需要进行商品数量减少、生成订单、生成秒杀订单的操作，当三者都成功运行后，才能秒杀成功。

秒杀商品数量初始化

在系统初始化时，调用这个方法，将秒杀商品的数量存储在redis中，后续使用redis进行预减功能

    @Overridepublic void getSeckillCount() {List<Seckill> list = baseMapper.selectList(null);list.forEach(System.out::println);if (list.size()>0) {list.forEach(seckill -> {redisUtil.set("seckill" + seckill.getGoodsId() + ":", seckill.getStockCount());Object o = redisUtil.get("seckill" + seckill.getGoodsId() + ":");System.out.println(o.toString());});}}

rabbitmq队列

生产者代码

把用户id和秒杀商品信息分发到队列中

public void sendSeckillMessage(String msg){System.out.println("发送秒杀信息"+msg);rabbitTemplate.convertAndSend("seckill_topic","seckill.message",msg);
}
消费者代码

从队列中获取用户id和商品信息。然后，判断该商品是否还有库存，是否已经秒杀过，如果都没有，则进行订单生产业务。

@RabbitListener(queues = "seckillQueue")
public void receive(String msg) {MqMessage message = JSONObject.parseObject(msg, MqMessage.class);Long userid = message.getUserid();GoodsVo goods = message.getGoodsVo();//判断是否还有库存int stock = goods.getStockCount();if(stock <= 0) {return;}//判断是否已经秒杀到了Orders order = ordersService.getOne(new QueryWrapper<Orders>().lambda().eq(Orders::getUserId,userid).eq(Orders::getGoodsId, goods.getId()));if(order != null) {return;}ordersService.seckill(userid,goods);
}
订单模块

订单模块：将用户购买的商品的数量减少1，然后生成订单和秒杀订单。三者都成功后，在redis中存储用户id和订单id，作为秒杀成功的记录，如果用户再次进行秒杀时，直接从redis查询是否存在秒杀成功的记录，存在即返回已经秒杀

public void seckill(Long userid, GoodsVo goods) {Seckill seckill = seckillMapper.selectOne(new QueryWrapper<Seckill>().lambda().eq(Seckill::getGoodsId,goods.getId()));seckill.setStockCount(seckill.getStockCount()-1);seckillMapper.updateById(seckill);//生成订单Orders orders = new Orders();orders.setUserId(userid).setGoodsId(goods.getId()).setGoodsCount(1).setStatu(0).setCreateDate(new Date()).setGoodsPrice(seckill.getSeckillPrice());baseMapper.insert(orders);//生成秒杀订单SeckillOrders seckillOrders = new SeckillOrders();seckillOrders.setOrderId(orders.getId());seckillOrders.setUserId(userid);seckillOrders.setGoodId(goods.getId());// 保存秒杀订单信息int i=seckillOrdersMapper.insert(seckillOrders);redisUtil.set("order:"+userid+goods.getId(),i);}

注意事项：

  1. 性能优化：在实际生产环境中，您可能需要考虑使用Redis的Lua脚本来确保库存扣减的原子性，并考虑使用分布式锁来避免超卖。
  2. 限流：在秒杀场景下，大量的请求可能会导致系统崩溃。因此，您可能需要在应用层或网络层实现限流策略。
  3. 异步处理：使用RabbitMQ进行异步处理可以确保秒杀接口的快速响应，并避免同步处理订单逻辑导致的性能瓶颈。
  4. 持久化：虽然上述案例中没有明确提到，但您应该确保将秒杀成功的订单信息持久化到数据库中，以便后续的处理和查询。
  5. 错误处理：在分布式系统中，错误处理是非常重要的。您应该确保在RabbitMQ消费者中正确处理各种异常情况，并考虑使用重试机制来确保消息的可靠传递。

 


http://www.ppmy.cn/news/1446699.html

相关文章

【论文笔记】Training language models to follow instructions with human feedback A部分

【论文笔记】Training language models to follow instructions with human feedback A部分

Training language models to follow instructions with human feedback A 部分 回顾一下第一代 GPT-1 &#xff1a; 设计思路是 “海量无标记文本进行无监督预训练少量有标签文本有监督微调” 范式&#xff1b;模型架构是基于 Transformer 的叠加解码器&#xff08;掩码自注意…
阅读更多...
探秘STM32与MPU6050传感器：姿态检测的奥秘

探秘STM32与MPU6050传感器：姿态检测的奥秘

探秘STM32与MPU6050传感器&#xff1a;姿态检测的奥秘 在嵌入式系统中&#xff0c;MPU6050传感器是一款常用的六轴惯性传感器&#xff0c;可以实现姿态检测等功能。结合STM32微控制器&#xff0c;我们可以更好地利用MPU6050传感器&#xff0c;实现姿态检测等应用。本文将带您深…
阅读更多...
k8s中，configMap与环境变量的关系

k8s中，configMap与环境变量的关系

总结 在Kubernetes中&#xff0c;ConfigMap和环境变量都可以用于向容器传递配置信息。以下是它们的区别&#xff1a; 环境变量&#xff1a;这是一种将配置信息直接注入到容器中的方法。它们在容器启动时被设置&#xff0c;并且在容器的生命周期内保持不变。ConfigMap&#xf…
阅读更多...
触发器的查看和删除

触发器的查看和删除

Oracle从入门到总裁:​​​​​​https://blog.csdn.net/weixin_67859959/article/details/135209645 如果想查看当前所有的触发器信息&#xff0c;可以使用数据字典 user_triggers&#xff0c;这个数据字典有很多字段可以查看所有触发器的名称、类型、表名、拥有者等信息。 …
阅读更多...
WAF（Web Application Firewal）

WAF（Web Application Firewal）

WAF&#xff0c;即Web Application Firewall&#xff0c;是一种专注于保护Web应用程序免受恶意攻击的安全解决方案。它工作在应用层&#xff08;OSI模型的第7层&#xff09;&#xff0c;专门设计用于检测、拦截和预防各种针对Web应用的攻击企图。以下是WAF防火墙的关键特点和功…
阅读更多...
MongoDB聚合运算符：$strLenCP

MongoDB聚合运算符：$strLenCP

MongoDB聚合运算符&#xff1a;$strLenCP $strLenCP聚合运算符返回指定字符串中 UTF-8 代码点的数量。 语法 { $strLenCP: <string expression> }<expression>为可解析为字符串的表达式&#xff0c;如果解析为null或引用了不存在的字段&#xff0c;返回错误。 …
阅读更多...
二，网络安全常用术语

二，网络安全常用术语

黑客&#xff08;hacker&#xff09;——对计算机技术非常擅长的人&#xff0c;窃取数据&#xff0c;破坏计算机系统&#xff1b;全球最知名的一个黑客组织匿名&#xff08;Anonymous&#xff09;。 脚本小子——刚刚入门安全行业&#xff0c;学习了一些技术&#xff0c;只会用…
阅读更多...
解决elemen-ui的el-table的树结构数据，新增修改不刷新问题

解决elemen-ui的el-table的树结构数据，新增修改不刷新问题

前面有写过关于后端返回全量数据&#xff0c;但前端节点过多table树卡顿问题 有兴趣可以看看这篇 https://blog.csdn.net/qq_44179024/article/details/136058117?spm1001.2014.3001.5501 前提&#xff1a;我这个是根据后端返回的全量数据来递归做的load方法&#xff0c;并没有…
阅读更多...
最新文章

Copyright @ 2022~2023