分布式秒杀技术学习（2） - 产品查询

it2026-08-14 9

文章目录

前言一、秒杀流程1. 大致秒杀流程2. 数据库准备3. 实体类、DAO层二、查询商品1.直接查询数据库2. 使用redis缓存a. 缓存雪崩b. 缓存击穿c. 缓存穿透 3. 使用本地缓存总结

前言

在上一章当中，我们完成了环境的搭建。从这章开始，我们一步一步来实现秒杀业务的代码编写。

一、秒杀流程

1. 大致秒杀流程

整个秒杀的流程如下：

查询商品，如果商品不存在，当然直接秒杀失败如果商品存在，则要查询商品的库存并进行预扣提交订单，提醒用户秒杀成功并付款付款成功，秒杀成功入库并结束付款失败，秒杀库存+1，秒杀失败在编码的过程中，需要注意后面的流程可能会对前面的流程有影响，比如付款失败后，会对库存增加操作。这与第2步的预扣是环环相扣的。

2. 数据库准备

首先创建两个表，脚本如下

-- ---------------------------- -- Table structure for goods -- ---------------------------- DROP TABLE IF EXISTS `goods`; CREATE TABLE `goods` ( `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Id', `goodsname` varchar(45) NOT NULL COMMENT '商品名称', `count` int NOT NULL COMMENT '商品数量', PRIMARY KEY (`id`) ) ENGINE = InnoDB AUTO_INCREMENT = 4 DEFAULT CHARSET = utf8 COMMENT ='商品信息表'; -- ---------------------------- -- Records of goods -- ---------------------------- INSERT INTO `goods` VALUES ('1', '华为手机', '1000'); INSERT INTO `goods` VALUES ('2', '华为手机', '1000'); DROP TABLE IF EXISTS `goods_user`; CREATE TABLE `goods_user` ( `goodsId` int(11) NOT NULL COMMENT '商品编号', `userId` varchar(45) NOT NULL COMMENT '用户编号', key (`goodsId`, `userId`) ) ENGINE = InnoDB DEFAULT CHARSET = utf8 COMMENT ='商品用户秒杀结果表';

3. 实体类、DAO层

创建实体类 package com.example.kill.entity; import java.io.Serializable; public class Goods implements Serializable { private Integer id; private String goodsname; private Integer count; public Integer getId() { return id; } public void setId(Integer id) { this.id = id; } public String getGoodsname() { return goodsname; } public void setGoodsname(String goodsname) { this.goodsname = goodsname; } public Integer getCount() { return count; } public void setCount(Integer count) { this.count = count; } } package com.example.kill.entity; import java.io.Serializable; public class GoodsUser implements Serializable { private Integer goodsId; private String userId; public Integer getGoodsId() { return goodsId; } public void setGoodsId(Integer goodsId) { this.goodsId = goodsId; } public String getUserId() { return userId; } public void setUserId(String userId) { this.userId = userId; } }

需要注意实现java.io.Serializable接口，因为后面需要进行缓存操作，可能涉及到序列化与反序列化。

创建mapper接口和xml文件 package com.example.kill.mapper; import com.example.kill.entity.Goods; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; import java.util.List; @Mapper public interface GoodsMapper { Goods selectByPrimaryKey(Integer id); int updateSecKill(@Param("id") Integer id); } package com.example.kill.mapper; import com.example.kill.entity.GoodsUser; import org.apache.ibatis.annotations.Mapper; @Mapper public interface GoodsUserMapper { GoodsUser findByUserId(String userId); } <?xml version="1.0" encoding="UTF-8"?> <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd"> <mapper namespace="com.example.kill.mapper.GoodsMapper"> <resultMap id="BaseResultMap" type="com.example.kill.entity.Goods">  <id column="id" jdbcType="INTEGER" property="id"/> <result column="goodsname" jdbcType="VARCHAR" property="goodsname"/> <result column="count" jdbcType="INTEGER" property="count"/> </resultMap> <select id="selectByPrimaryKey" parameterType="java.lang.Integer" resultMap="BaseResultMap"> select id, goodsname, count from goods where id = #{id,jdbcType=INTEGER} </select> <update id="updateSecKill"> update goods SET count = count - 1 where id = #{id,jdbcType=INTEGER} and count > 0 </update> </mapper> <?xml version="1.0" encoding="UTF-8"?> <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd"> <mapper namespace="com.example.kill.mapper.GoodsUserMapper"> <resultMap id="BaseResultMap" type="com.example.kill.entity.GoodsUser">  <id column="goodsId" jdbcType="INTEGER" property="goodsId"/> <result column="userId" jdbcType="VARCHAR" property="userId"/> </resultMap> <select id="findByUserId" resultMap="BaseResultMap"> select goodsId, userId from goods_user where userId = #{userId,jdbcType=VARCHAR} </select> </mapper> 业务层接口以及实现暂时提供两个接口一个是查询商品，一个是秒杀商品 package com.example.kill.service; import com.example.kill.entity.Goods; public interface KillService { Goods detail(Integer goodsId); boolean secKillByDb(Integer goodsId, String sessionUserId); }

实现类

package com.example.kill.service.impl; import com.example.kill.entity.Goods; import com.example.kill.entity.GoodsUser; import com.example.kill.mapper.GoodsMapper; import com.example.kill.mapper.GoodsUserMapper; @Service public class KillServiceImpl implements KillService { private static final Logger logger = LoggerFactory.getLogger(KillService.class); @Autowired private GoodsMapper goodsMapper; @Autowired private GoodsUserMapper goodsUserMapper; @Override public Goods detail(Integer goodsId) { Goods goods = null; String killGoodsId = KillConstants.KILLGOOD_DETAIL + goodsId; return goods; } @Override public boolean secKillByDb(Integer goodsId, String userId) { return true; } } 控制层

在正常项目中需要考虑秒杀时间的问题，此处简化处理。用户需要从登陆的session信息中获取到，这里直接随机取uuid值。秒杀的方法中首先查询商品，如果商品不存在，就会抛出异常。如果商品存在，则执行秒杀服务。

package com.example.kill.controller; import com.example.kill.constant.HttpResponseBody; import com.example.kill.entity.Goods; import com.example.kill.service.KillService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import java.util.UUID; @RestController public class KillController { private static final Logger logger = LoggerFactory.getLogger(KillController.class); @Autowired private KillService killService; @RequestMapping("kill/killByDb/{goodsId}") public HttpResponseBody killByDb(@PathVariable("goodsId") Integer goodsId) { Goods goods = killService.detail(goodsId); if (goods == null) { return HttpResponseBody.failResponse("产品不存在"); } // if (goods.getBegainTime().getTime() > System.currentTimeMillis()){ // return HttpResponseBody.failResponse("抢购还未开始"); // } // if (goods.getEndTime().getTime() < System.currentTimeMillis()){ // return HttpResponseBody.failResponse("抢购已结束"); // } if (!killService.secKillByDb(goodsId, getSessionUserId())) { return HttpResponseBody.failResponse("抢购失败"); } return HttpResponseBody.successResponse("ok", goods); } private String getSessionUserId() { return UUID.randomUUID().toString(); } } 一些公共类和配置类 package com.example.kill.constant; import java.io.Serializable; public class HttpResponseBody<E> implements Serializable { /** * Comment for <code>serialVersionUID</code> */ private static final long serialVersionUID = -5406928126690333502L; private String code; private String msg; private E data; public HttpResponseBody() { } public HttpResponseBody(String code, String msg, E data) { this.code = code; this.msg = msg; this.data = data; } public HttpResponseBody(String code, String msg) { this.code = code; this.msg = msg; } public static HttpResponseBody successResponse(String message) { return new HttpResponseBody("秒杀成功", message); } public static <T> HttpResponseBody<T> successResponse(String message, T singleData) { return new HttpResponseBody<>("秒杀成功", message, singleData); } public static HttpResponseBody failResponse(String message) { return new HttpResponseBody("秒杀失败", message); } public String getCode() { return code; } public void setCode(String code) { this.code = code; } public String getMsg() { return msg; } public void setMsg(String msg) { this.msg = msg; } public E getData() { return data; } public void setData(E data) { this.data = data; } } package com.example.kill.config; import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.StringRedisSerializer; @Configuration public class RedisConfig { @Bean public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) { RedisTemplate<String, Object> template = new RedisTemplate<>(); template.setConnectionFactory(factory); //使用jdk的序列化 template.setValueSerializer(new FastJsonRedisSerializer<>(Object.class)); //使用StringRedisSerializer来序列化和反序列化redis的key值 template.setKeySerializer(new StringRedisSerializer()); template.afterPropertiesSet(); return template; } }

二、查询商品

首先需要完成的就是商品的查询操作

1.直接查询数据库

@Override public Goods detail(Integer goodsId) { String killGoodsId = KillConstants.KILLGOOD_DETAIL + goodsId; return goodsMapper.selectByPrimaryKey(goodsId); }

这个是最简单的实现。

在上面的测试中，吞吐量相对于之前的测试还算可以。但是这并不能说明直接请求数据库层是可以的，我们测试时使用的线程为1000，如果数量继续加大，所有的查询都打到数据库，可能就立刻导致数据库的崩溃。服务层核心设计思想，尽量把大量的请求不要瞬时落到数据库层，读请求， cache 来扛， memcached， redis 单机扛 10W+应该没问题。

2. 使用redis缓存

使用redis缓存能有效提高服务的抗压能力。但有几点需要注意：

a. 缓存雪崩

问题：缓存雪崩指的是在某一个时刻大流量怼到系统，这时候系统出现了大量的key同时失效，这样导致了大量的请求到了数据层，导致数据库奔溃从而导致整个系统雪崩的问题。解决方案：key采用不同的实现方案，尽量避免key设置同样的失效时间

b. 缓存击穿

问题：击穿指的是，在某一个时刻大并发下请求同一个key，而这个key恰好在这个时候失效了，这时候大量的请求就会怼到数据库从而导致系统崩溃。解决方案：缓存要做预热，且缓存的失效时间要大于业务生命周期时间，比如一个秒杀业务，1个小时内抢完，那么这个key的失效时间要大于1小时，请求数据库的逻辑需要加锁，避免大量请求落到数据库层。可能这个锁的逻辑块永远都不会执行，因为缓存是存在于redis的，但是要考虑健壮性考虑。

c. 缓存穿透

问题：缓存穿透更多的是一种恶意访问，黑客故意大量访问一个redis中没有，数据也没有的数据，这样同样会导致大量请求落到数据库，所以访问数据库加锁是必要的。但是这里又有一个问题，恶意访问会占用redis的连接资源，所以这里需要使用拦截手段，把请求都拦截在redis之外，比如使用布隆过滤器，比如用本地缓存都可以拦截对redis的恶意访问。解决方案1：如果数据不经常变动，采用布隆过滤器是合适的，其核心目的是减少对redis的恶意访问，减少对redis的连接资源占用。参考博客：缓存解决方案2：如果是redis没有，数据库也没有，可以吧一个null字符串存储到redis并且存一份到本地缓存，存本地缓存的目的也是为了减少对redis的访问压力。

考虑以上的问题和解决方案，此次实现缓存的失效时间为2天（避免缓存击穿问题），另外如果数据库也不包含数据，会在redis缓存中保存一个null的字符串（防止缓存穿透问题）。

@Override public Goods detail(Integer goodsId) { Goods goods = null; String killGoodsId = KillConstants.KILLGOOD_DETAIL + goodsId; Object killGoodCache = redisTemplate.opsForValue().get(killGoodsId); if (null != killGoodCache) { logger.info(Thread.currentThread().getName() + "---redis缓存中得到的数据---------"); return JSONObject.parseObject(killGoodCache.toString(), Goods.class); } synchronized (KillServiceImpl.class) { // 1、从缓存里面拿 killGoodCache = redisTemplate.opsForValue().get(killGoodsId); if (null != killGoodCache) { logger.info(Thread.currentThread().getName() + "---redis缓存中得到的数据---------"); return JSONObject.parseObject(killGoodCache.toString(), Goods.class); } // 2、去数据库里面查询数据 goods = goodsMapper.selectByPrimaryKey(goodsId); if (null != goods) { redisTemplate.opsForValue().set(killGoodsId, goods, 2, TimeUnit.DAYS); } else { //防止缓存穿透缓存时间一定要短，空数据没有必要占用redis内存 redisTemplate.opsForValue().set(killGoodsId, "null", 5, TimeUnit.MINUTES); } return goods; }

该处使用的url网络请求的数据。

3. 使用本地缓存

在查询秒杀商品详情的时候，我们首先是把秒杀商品做缓存预热加载到了redis中了，但是我们忽略了一个问题，当流量达到一定量的时候，比如流量超过了10W+的并发时，这时候redis也是扛不住的，这时候就需要考虑使用本地缓存了，而且使用本地缓存也可以加快程序响应速度，毕竟如果详情信息从redis中拿的话，还是需要占用网络带宽的，其速度不可能比使用本地缓存更快。此处我们使用ehcache作为本地缓存。

import net.sf.ehcache.Cache; import net.sf.ehcache.CacheManager; import net.sf.ehcache.Element; @Autowired private GoodsMapper goodsMapper; @Autowired private RedisTemplate redisTemplate; @Autowired private GoodsUserMapper goodsUserMapper; @Autowired private CacheManager cacheManager; @Override public Goods detail(Integer goodsId) { Goods goods = null; String killGoodsId = KillConstants.KILLGOOD_DETAIL + goodsId; Cache killgoodsLocalCache = cacheManager.getCache("killgoodDetail"); if (null != killgoodsLocalCache.get(killGoodsId)) { logger.info(Thread.currentThread().getName() + "--------ehcache缓存里面的到数据-------"); goods = (Goods) killgoodsLocalCache.get(killGoodsId).getObjectValue(); return goods; } Object killGoodCache = redisTemplate.opsForValue().get(killGoodsId); if (null != killGoodCache) { logger.info(Thread.currentThread().getName() + "---redis缓存中得到的数据---------"); return JSONObject.parseObject(killGoodCache.toString(), Goods.class); } synchronized (KillServiceImpl.class) { if (null != killgoodsLocalCache.get(killGoodsId)) { logger.info(Thread.currentThread().getName() + "--------ehcache缓存里面的到数据-------"); goods = (Goods) killgoodsLocalCache.get(killGoodsId).getObjectValue(); return goods; } killGoodCache = redisTemplate.opsForValue().get(killGoodsId); if (null != killGoodCache) { logger.info(Thread.currentThread().getName() + "---redis缓存中得到的数据---------"); return JSONObject.parseObject(killGoodCache.toString(), Goods.class); } goods = goodsMapper.selectByPrimaryKey(goodsId); if (null != goods) { killgoodsLocalCache.putIfAbsent(new Element(killGoodsId, goods)); redisTemplate.opsForValue().set(killGoodsId, goods, 2, TimeUnit.DAYS); } else { //防止缓存穿透缓存时间一定要短，空数据没有必要占用redis内存 redisTemplate.opsForValue().set(killGoodsId, "null", 5, TimeUnit.MINUTES); } } return goods; }

先查找本地缓存，如果有直接返回，再找redis，如果有直接返回，再去查询数据库，并把结果缓存到本地缓存和redis缓存，这里需要注意，本地缓存中缓存的数据量一定不要大，而且ehcache中缓存的数据的失效时间一定要短于redis中的数据的有效时间，目的是为了减少GC的压力。

另外 ehcache后续还有堆外模式，而且是一种大堆的堆外模式，这块有兴趣的同学可以去了解一下，这样很多缓存数据都可以存储到 ehcache 中，可以大大提高程序的响应速度。其实有没有发现，这里 ehcache 的使用是不是跟布隆过滤器有点像，都是为了减少对 redis 的访问压力。而且我们可以把原本会落到 redis 的请求量分散到了各个应用服务器了，而应用服务器的数量肯定是要多余 redis 服务器的，这样就把压力均摊到了各个应用服务器了。

相关ehcache的配置：

package com.example.kill.config; import net.sf.ehcache.CacheManager; import org.springframework.cache.ehcache.EhCacheManagerFactoryBean; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Primary; import org.springframework.core.io.ClassPathResource; @Configuration public class EhcacheConfig { @Primary @Bean(name = "ehCacheManager") public CacheManager cacheManager(EhCacheManagerFactoryBean bean) { CacheManager cm = bean.getObject(); return cm; } @Bean public EhCacheManagerFactoryBean ehCacheManagerFactoryBean() { EhCacheManagerFactoryBean cacheManagerFactoryBean = new EhCacheManagerFactoryBean(); cacheManagerFactoryBean.setConfigLocation(new ClassPathResource("ehcache.xml")); cacheManagerFactoryBean.setShared(true); // 设置完属性后，cacheManagerFactoryBean会执行afterProertiesSet()方法， // 所以不能在这里直接执行cacheManagerFactoryBean.getObject(),直接执行的话，因为在afterPropertiesSet()方法之前执行，所以：getObject()会得到null值 return cacheManagerFactoryBean; } }

src/main/resources/ehcache.xml

<?xml version="1.0" encoding="UTF-8"?> <ehcache xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="http://ehcache.org/ehcache.xsd" updateCheck="false"> <diskStore path = "/home/cache"/>   <defaultCache eternal="false" maxElementsInMemory="900" overflowToDisk="false" diskPersistent="false" timeToIdleSeconds="0" timeToLiveSeconds="60" memoryStoreEvictionPolicy="LRU" />  <cache name="killgoodDetail" eternal="false" maxElementsInMemory="5000" overflowToDisk="true" diskPersistent="false" timeToIdleSeconds="0" timeToLiveSeconds="86400" memoryStoreEvictionPolicy="LRU" /> </ehcache>

总结

本章通过一个简单的实例介绍了秒杀过程中的一些问题以及redis作为二级缓存、ehcache作为一级本地缓存来应对高并发。在实际过程中，问题只会更多。比如：

秒杀和抢购收到了“海量”的请求，实际上里面的水分是很大的。不少用户，为了“抢“到商品，会使用“刷票工具”等类型的辅助工具，帮助他们发送尽可能多的请求到服务器。还有一部分高级用户，制作强大的自动请求脚本。这种做法的理由也很简单，就是在参与秒杀和抢购的请求中，自己的请求数目占比越多，成功的概率越高

同一账号发送多次请求这种加锁就可以解决， redis 锁，如果其中一个请求获取锁成功就允许秒杀，其他都拒绝多个账号发送多次请求黄牛党，僵尸号在秒杀开始的时候疯狂请求可以做 ip 限制，如果某个 ip 请求频率过高，可以弹验证码或者实名验证，其实就是判断是否是活人。多个账号，不同 ip 这种就比较复杂了，典型的就是高级黑客手里掌握了 N 多肉鸡服务器，那么 ip 就不一样了这种请求就跟普通用户请求基本上类似了这种必须要做业务限制，提高秒杀的参与门槛，比如只有星级用户、账号等级等等业务手段来做限制。而僵尸账号往往账号等级和活跃度都不高

最新回复(0)