`
Adan-Chiu
  • 浏览: 21950 次
社区版块
存档分类
最新评论

高并发业务

 
阅读更多

 

搭建商品秒杀开发环境和超卖现象

        适逢618狂欢大冲刺，嗨购不停歇，现在模拟一件秒杀商品浪琴瑰丽女表，原价10000元，秒杀价7999块，共10000件，有2万人同时抢夺的场景。在高并发的场景下，除了数据的一致性外，还要关注性能问题，一般而言，超过5秒用户体验就不太好了，所以要考虑数据一致性和系统的性能。

1.技术栈采用spring-boot-starter-data-jpa，导入相关依赖

compile 'org.springframework.boot:spring-boot-starter-data-jpa'
compile 'org.springframework.boot:spring-boot-starter-web'
compile 'org.springframework.boot:spring-boot-starter-thymeleaf'
compile 'org.springframework.boot:spring-boot-devtools'
runtimeOnly 'mysql:mysql-connector-java'
compile 'com.alibaba:druid:1.1.17'

2.定义相关实体

/**
  * 秒杀商品实体
  */
@Entity
public class SpikeProductInfo {
    /**
     * 秒杀商品id
     */
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Integer productId;
    /**
     * 商品名称
     */
    @Column(length = 50,nullable = false)
    private String productName;
    /**
     * 市场价
     */
    private float marketPrice;
    /**
     * 销售价
     */
    private float sellPrice;
    /**
     * 库存
     */
    private int stock;
    /**
     * 商品总数
     */
    private int total;
    /**
     * 商品备注
     */
    @Lob @Basic(fetch = FetchType.LAZY)
    private String note;
//************ setter and getter **********/
}
@Entity
@Table(name = "t_order")
public class Order {
    /**
     * 订单id，实际应该按照业务需求生成订单id，比如日期+用户编号+其它规则，
     * 这里使用Hibernate的UUID来保证订单编号的唯一性（没有业务规则）
     * 使用一个128-bit的UUID算法生成字符串类型的标识符，UUID被编码成一个32位16进制数字的字符串。UUID包含：IP地址、JVM启动时间、系统时间（精确到1/4秒）和一个计数器值（JVM中唯一）
     */
    @Id
    @GenericGenerator(name = "orderGenerator",strategy = "uuid")
    @GeneratedValue(generator = "orderGenerator")
    @Column(length = 32)
    private String orderId;
    /**
     * 下单时间
     */
    private LocalDateTime createDate = LocalDateTime.now();
    /**
     * 产品名称,产品相关属性之所以不采用关联是怕活动结束后价格等产生歧义
     */
    @Column(length = 50,nullable = false)
    private String productName;
    /**
     *  产品id
     */
    @Column(nullable = false)
    private Integer productId;
    /**
     *  产品销售价
     */
    private Float productPrice = 0f;
//************* setter and getter ***********//
}

3.搭建Repository层和Service层

public interface SpikeProductInfoRepository extends JpaRepository<SpikeProductInfo,Integer> {
    @Modifying @Query("update SpikeProductInfo o set o.stock = o.stock - 1 where o.productId = :productId")
    int decreaseStock(@Param("productId") Integer productId);
}
public interface OrderRepository extends JpaRepository<Order,String> {
}
public interface SpikeProductInfoService {
    /**
     * 获取秒杀商品信息
     * @param productId 商品id
     * @return 秒杀商品具体信息
     */
    SpikeProductInfo getSpikeProductInfo(Integer productId);

    /**
     * 扣减秒杀商品库存
     * @param productId 商品id
     * @return 更新记录条数
     */
    int decreaseStock(Integer productId);

    /**
     * 保存一个秒杀商品信息
     * @param productInfo 秒杀商品信息
     * @return 秒杀商品信息，主键回填
     */
    SpikeProductInfo save(SpikeProductInfo productInfo);
}
public interface OrderService {
    /**
     * 插入订单信息
     * @param productId 商品id
     * @return 订单详情信息
     */
    Order createOrder(Integer productId);
}

业务实现类：

@Service
@Transactional
public class SpikeProductInfoServiceImpl implements SpikeProductInfoService {
    @Autowired private SpikeProductInfoRepository repository;
    @Override 
    public SpikeProductInfo getSpikeProductInfo(Integer productId) {
        return repository.findById(productId).get();
    }

    @Override
    public int decreaseStock(Integer productId) {
        return repository.decreaseStock(productId);
    }

    @Override
    public SpikeProductInfo save(SpikeProductInfo productInfo) {
        return repository.save(productInfo);
    }
}
@Service
@Transactional
public class OrderServiceImpl implements OrderService {
    @Autowired private OrderRepository repository;
    @Autowired private SpikeProductInfoServiceImpl productInfoService;
    @Override
    public Order createOrder(Integer productId) {
        //1.查询该商品库存
        var product = productInfoService.getSpikeProduct(productId);
        if(product.getStock() > 0){//商品库存大于0下订单
            var order = new Order();
            order.setProductId(productId);
            order.setProductName(product.getProductName());
            order.setProductPrice(product.getSellPrice());
            var ret = repository.save(order);//下单（秒杀）
productInfoService.decreaseStock(productId);//减库存
            return ret;
        }
        throw new RuntimeException("秒杀结束");
    }
}

4.开发控制器和超卖现象测试

@RestController
@RequestMapping("product")
public class SpikeProductInfoController {
    @Autowired private SpikeProductInfoService service;
    @Autowired private OrderService orderService;

    @RequestMapping("/save")
    public String save(SpikeProductInfo productInfo){
        service.save(productInfo);
        return productInfo.getProductName() + "开始秒杀,市场价为："
                +productInfo.getMarketPrice()+"秒杀价为："
                +productInfo.getSellPrice()+"还剩：" + productInfo.getStock()+"件";
    }
    @RequestMapping("/query")
    public SpikeProductInfo querySpikeProductInfo(Integer productId){
        return service.getSpikeProductInfo(productId);
    }
    @RequestMapping("/spike")
    public String spike(Integer productId){
        orderService.createOrder(productId);
        return "秒杀成功";
    }
}

       这里使用Jmeter或Apache的ab压测工具去模拟20000人同时抢该商品的场景，而该商品库存仅仅为1000件（也可以使用ajax的异步请求来进行模拟，代码如下），在这样高并发场景下会有什么问题发生呢？注意点两个:一个是数据的一致性，一个是性能问题。

<script src="menu/js/jquery-3.1.1.min.js"></script>
    <script>
        $(function(){
            for(var i = 0; i < 20000; i++){
                $.post({
                    url: "product/spike?productId=2",
                    success: function(result){
                    }
                });
            }
        });
    </script>

        观察数据库的数据，会发现超卖现象，查询该商品信息，发现该商品信息现有库存为-5，超出了之前的限定，这就是高并发的超卖现象，这是一个错误的逻辑再来看一下性能问题

       通过第一个秒杀订单和最后一个秒杀订单来看看其时间间隔来查看其执行时间

       一共使用了72秒多的时间，完成了10005个商品的秒杀，由于环境的原因，性能还是不错的，但是逻辑上存在超卖的错误，需要解决超卖的问题。

       超卖现象是由多线程下数据不一致造成的，对于此类问题，当前互联网主要通过悲观锁和乐观锁来处理，以保证数据的一致性，这两种方法的性能是不一样的。

悲观锁

       悲观锁是一种利用数据库内部机制提供的锁的方法，也就是对更新的数据加锁，这样在并发期间一旦有一个事务持有了数据库记录的锁，其它线程将不能再对数据进行更新了。spring-data-jpa对悲观锁提供了支持，实现方式有两种：

  •  添加@Lock注解，并设置值为LockModeType.PESSIMISTIC_WRITE（附录中对该注解进行解释） 
    @Override @Lock(LockModeType.PESSIMISTIC_WRITE)
Optional<SpikeProductInfo> findById(Integer integer);
  • 本地SQL，在SQL语句中加入for update 
    @Query(value = "select o.* from spike_product_info o where o.productId = :id for update", nativeQuery = true)
Optional<SpikeProductInfo>findForUpdate(@Param("id") Integer integer);
     注意，在SQL中加入的for update语句，意味着将持有对数据库记录的行更新锁（这里使用主键查询，所以只会对行加锁，如果使用的非主键，要考虑是否对全表加锁的问题）意味着在高并发的场景下，当一条事务持有了这个更新锁才能往下操作，其它的线程如果要更新这条记录，都需要等待，这样就不会出现超卖现象引发的数据一致性问题了。再新增一条秒杀商品信息，Rolex的绿水鬼，原价135000元，秒杀价35000块，共10000件，有2万人同时抢夺的场景进行测试。

       这里已经解决超卖问题了，结果是正确的，但是对于互联网而言，除了结果正确，还需要考虑性能问题。对于悲观锁来说，当一条线程抢占了资源后，其它的线程将得不到资源，那么这个时候CPU就会将这些得不到资源的线程挂起，挂起的线程会消耗CPU的资源，尤其在高并发的请求中，只能有一个事务占据资源，其它事务被挂起等待持有资源（锁）的事务提交并释放资源。一旦持有资源的线程提交了事务，那么锁就会被释放，这个时候被挂起的线程就会开始竞争秒杀资源，那么竞争到的线程会被CPU恢复到运行状态，继续运行。于是频繁挂起，等待持有锁线程释放资源，一旦资源释放后，就开始抢夺，恢复线程，直到商品秒杀完。在高并发环境中这将十分消耗资源。有些时候我们把悲观锁称为独占锁或阻塞锁，因为只有一个线程可以独占这个资源，所以会造成其它线程的阻塞。会造成并发能力的下降，导致CPU频繁切换线程上下文。造成性能低下。为了克服这个问题，提出了乐观锁机制。

乐观锁

       乐观锁是一种不会阻塞其它线程并发的机制，它不会使用数据库的锁进行实现，由于不阻塞其它线程，所以并不会引发线程频繁挂起和恢复，这样便可以提高并发能力。所以又称为非阻塞锁。乐观锁使用的是CAS原理。

      CAS原理概述

      CAS操作包含三个操作数—— 内存位置的值（V）、预期原值（A）和新值（B）。如果内存位置的值与预期原值相匹配，那么处理器会自动将该位置更新为新值。否则，处理器不做任何操作。无论哪种情况，它都会在CAS指令之前返回该位置的值。CAS有效地说明了“我认为位置V应该包含值A；如果包含该值，则将B放到这个位置；否则，不要更改该位置，只告诉我这个位置现在的值即可。“

CAS是一种有名的无锁算法。无锁编程，即不适用锁的情况下实现多线程之间的变量同步，也就是在没有现成被阻塞的情况下实现变量的同步。

总结如下：

  • CAS（Compare And Swap）比较并替换，是线程并发运行时用到的一种技术
  • CAS是原子操作，保证并发安全，而不能保证并发同步
  • CAS是CPU的一个指令（需要JNI调用Native方法，才能调用CPU的指令）
  • CAS是非阻塞的、轻量级的乐观锁

       CAS原理并不排斥并发，也不独占资源，只是在线程开始阶段就读入线程共享数据，保存为旧值。当处理完逻辑，需要更新数据的时候，会进行一次比较，即比较各个线程当前共享的数据是否和旧值保持一致。如果一致就更新数据，如果不一致，则认为该数据已经被其它线程修改了，那么就不再更新数据，可以考虑重试或者放弃。有时候可以重试，这样就是一个可重入锁，CAS原理会有一个问题，那就是ABA问题。

 线程C、D；线程D将A修改为B后又修改为A，此时C线程以为A没有改变过

时刻 线程C 线程D 备注
T0 —— —— 初始化X = A
T1 读入X = A —— ——
T2 —— 读入X = A ——
T3 处理线程C的业务逻辑 X = B 修改共享变量为B
T4   处理线程D的业务逻辑第一段 线程C在X=B的情况下运行逻辑
T5   X = A 还原变量为A
T6 因为判断X=A，更新数据 处理线程D的业务逻辑第二段 此时线程C无法知道线程D是否修改过X，引发业务逻辑错误
T7 —— 更新数据 ——

       ABA问题的发生，是因为业务逻辑存在回退的可能性。如果加入一个非业务逻辑的属性，比如在一个数据中加入版本号（version），对于版本号有一个约定，就是只要修改变量X的数据，强制版本号只能递增而不会回退，即使其它业务数据回退，它也会递增，那么ABA问题就解决了。

spring-data-jpa对乐观锁的支持

1.在秒杀商品实体类中新增一个字段version用于版本控制

    /**
     * 版本号的类型支持int, short, long三种基本数据类型和他们的包装类以及Timestamp
     */
    @Version
    private int version;

      spring-data-jpa对锁有自己优雅的实现方式，采用@Lock注解里有6中锁模式（包含悲观锁及乐观锁），乐观锁也可以采用@Version注解来进行版本控制，不需要程序自己去维护版本号。如果没有成功更新数据则抛出ObjectOptimisticLockingFailureException异常回滚保证数据的一致性。如果想要实现重入流程可以捕获ObjectOptimisticLockingFailureException这个异常，下面用sql语句控制的方式实现乐观锁以及重入。

 

@Modifying
@Query("update SpikeProductInfo o set o.stock = o.stock - 1,o.version = o.version + 1 where o.productId = :productId and o.version = :version")
int decreaseStockOptimisticLock(@Param("productId") Integer integer,@Param("version") int version);

        可以看到update的where有一个判断version的条件，并且会set version = version + 1。这就保证了只有当数据库里的版本号和要更新的实体类的版本号相同的时候才会更新数据。

Service秒杀代码

@Service
@Transactional
public class OrderServiceImpl implements OrderService {
    @Autowired private OrderRepository repository;
    @Autowired private SpikeProductInfoRepository productInfoRepository;
    @Override
    public Order createOrder(Integer productId) {
        //1.查询该商品库存
        var product = productInfoRepository.findById(productId).get();
        if(product.getStock() > 0){//商品库存大于0下订单
            var update = productInfoRepository.decreaseStockOptimisticLock(productId,product.getVersion());//减库存
            //如果没有数据更新，说明其它线程已经修改过数据，本次抢购失败
            if(update == 0) throw new ObjectOptimisticLockingFailureException(product.getClass(),product);
            var order = new Order();
            order.setProductId(productId);
            order.setProductName(product.getProductName());
            order.setProductPrice(product.getSellPrice());
            var ret = repository.save(order);//下单（秒杀）
            return ret;
        }
        throw new RuntimeException("秒杀结束");
    }
}
       version值一开始就保存到了对象中，当扣减的时候，再次传递给SQL，让SQL对数据库的version和当前线程的旧值version进行比较。如果一致则秒杀下单，否则就抛出异常。

 

乐观锁重入机制

因为乐观锁造成大量更新失败的问题，可以使用两种机制执行乐观锁重入以助于秒杀的成功率。

  • 按时间戳重入：在一定的时间戳内（比如100毫秒），不成功的会循环到成功为止，直到超过时间戳。不成功退出
  • @Service
@Transactional
public class OrderServiceImpl implements OrderService {
    @Autowired private OrderRepository repository;
    @Autowired private SpikeProductInfoRepository productInfoRepository;
    @Override
    public Order createOrder(Integer productId) {
        //记录开始时间
        SpikeProductInfo product = null;
        var start = System.currentTimeMillis();
        while(true){
            //获取循环当前时间
            var end = System.currentTimeMillis();
            if(end - start > 100) throw new ObjectOptimisticLockingFailureException(product.getClass(),product);
            //查询该商品库存,注意version值
            product = productInfoRepository.findById(productId).get();
            if(product.getStock() > 0){//商品库存大于0下订单
               var update = productInfoRepository.decreaseStockOptimisticLock(productId,product.getVersion());//减库存
               //如果没有数据更新，说明其它线程已经修改过数据，本次抢购失败
               if(update == 0) continue;
               var order = new Order();
               order.setProductId(productId);
               order.setProductName(product.getProductName());
               order.setProductPrice(product.getSellPrice());
               var ret = repository.save(order);//下单（秒杀）
               return ret;
           }
           throw new RuntimeException("秒杀结束");
        }
    }
} 
     时间戳不是很稳定，会随着系统的空闲或者繁忙导致重试次数不一。那么就要考虑限制重试次数。
  • 按次数重入，比如限定3次，程序尝试超过3次秒杀失败后就判断请求失效 
    @Service
@Transactional
public class OrderServiceImpl implements OrderService {
    @Autowired private OrderRepository repository;
    @Autowired private SpikeProductInfoRepository productInfoRepository;
    @Override
    public Order createOrder(Integer productId) {
        for(var i = 0; i < 3; i++){
            //查询该商品库存,注意version值
            product = productInfoRepository.findById(productId).get();
            if(product.getStock() > 0){//商品库存大于0下订单
               var update = productInfoRepository.decreaseStockOptimisticLock(productId,product.getVersion());//减库存
               //如果没有数据更新，说明其它线程已经修改过数据，本次抢购失败
               if(update == 0) continue;
               var order = new Order();
               order.setProductId(productId);
               order.setProductName(product.getProductName());
               order.setProductPrice(product.getSellPrice());
               var ret = repository.save(order);//下单（秒杀）
               return ret;
           }
           throw new RuntimeException("秒杀结束");
        }
    }
}

内存共享数据->分布式锁

       使用setnx http://redis.cn/commands/setnx.html

SETNX key value

 

起始版本：1.0.0

时间复杂度：O(1)

将key设置值为value，如果key不存在，这种情况下等同SET命令。 当key存在时，什么也不做。SETNX是”SET if Not eXists”的简写。

返回值

Integer reply, 特定值:

  • 1 如果key被设置了
  • 0 如果key没有被设置

例子

redis> SETNX mykey "Hello"
(integer) 1
redis> SETNX mykey "World"
(integer) 0
redis> GET mykey
"Hello"
redis>

Design pattern: Locking with !SETNX

设计模式：使用!SETNX加锁

/**
 * 使用Redis作为分布式锁
 */
@Component
public class RedisLock {
    @Autowired private StringRedisTemplate template;
    /**
     * 加锁
     * @param key key
     * @param value 当前时间 + 超时时间
     * @return true->加锁成功，false->加锁失败
     */
    public boolean lock(String key,String value){
        //setnx命令
        if(template.opsForValue().setIfAbsent(key,value)){
            return true;
        }
        return false;
    }
    /**
     * 解锁
     * @param key
     * @param value
     */
    public void unlock(String key,String value){
        var currentValue = template.opsForValue().get(key);
        if (!StringUtils.isEmpty(currentValue)
                && currentValue.equals(value))
            template.delete(key);
    }
}

 客户端使用==伪代码

@Service
@Transactional
public class OrderServiceImpl2 implements OrderService {
    @Autowired private RedisLock redisLock;
    private static final int TIMEOUT = 1000; //超时时间1秒
    @Override
    public Order createOrder(Integer productId) {
        var currentTime = System.currentTimeMillis();
        //加锁
        if(!redisLock.lock(productId + "",currentTime + TIMEOUT + "")
            throw new SpikeException(555,"秒杀失败，换个姿势再来一次吧");
        //查询该商品库存
        var product = queryProduct(productId);
        if(product.getStock() > 0){//商品库存大于0下订单
            //创建订单---下订单
            var ret = 下订单;
            //扣减库存
            product.stock--;
        }else
            throw new SpikeException(520,"秒杀结束");
        //解锁
        redisLock.unlock(productId + "",currentTime + TIMEOUT + "");
    }
}

处理死锁

以上加锁算法存在一个问题：如果客户端出现故障，崩溃或者其他情况无法释放该锁会发生死锁的情况，下面加锁进行处理

/**
 * 使用Redis作为分布式锁
 */
@Component
public class RedisLock {
    @Autowired private StringRedisTemplate template;
    /**
     * 加锁
     * @param key key
     * @param value 当前时间 + 超时时间
     * @return true->加锁成功，false->加锁失败
     */
    public boolean lock(String key,String value){
        //setnx命令
        if(template.opsForValue().setIfAbsent(key,value)){
            return true;
        }
        //从redis中获取当前key的值 //确保一个线程拿到锁
        var currentValue = template.opsForValue().get(key);
        //如果锁过期
        if(!StringUtils.isEmpty(currentValue)
                && Long.parseLong(currentValue) < System.currentTimeMillis()){
            //获取上一个锁的时间，使用getset命令，该命令将新值替换掉key对应的旧值，并将旧值返回
/*
 * 假设AB两个线程到达，此时两个线程拿到的currentValue都为template.opsForValue().get(key);，oldValue即为currentValue
 * 当其中一个线程执行template.opsForValue().getAndSet(key,value)后，第二个线程线程get的时候已经是第一个线程修改后的值，那么条件就不成立，确保只有一个线程获得锁
 */            
var oldValue = template.opsForValue().getAndSet(key,value);
            if(!StringUtils.isEmpty(oldValue) && oldValue.equals(currentValue)){
                return true; 
            }
        }
        return false;
    }

    /**
     * 解锁,删除key
     * @param key
     * @param value
     */
    public void unlock(String key,String value){
        var currentValue = template.opsForValue().get(key);
        if (!StringUtils.isEmpty(currentValue)
                && currentValue.equals(value))
            template.delete(key);
    }
}

GETSET key value

起始版本：1.0.0

时间复杂度：O(1)

自动将key对应到value并且返回原来key对应的value。如果key存在但是对应的value不是字符串，就返回错误。

使用Redis进行商品秒杀

       to be continue；

分享到：
Spring 缓存机制
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

Magicbox
Global site tag (gtag.js) - Google Analytics