根據(jù)上一篇Demo測試情況反映，當有多個線程同時搶購時，會發(fā)生超發(fā)現(xiàn)象，所謂超發(fā)現(xiàn)象，就是原本設置庫存為30000件，但是，當搶購完成后發(fā)現(xiàn)庫存余量變成了負數(shù)，即發(fā)貨量大于庫存量的情況：

超發(fā)現(xiàn)象

造成這種現(xiàn)象的原因：當多個線程請求數(shù)據(jù)庫查詢庫存余量時，顯示有余量，但是當進行扣減庫存時，庫存已經(jīng)用完了，但那個線程并不知道，依舊去扣減庫存，造成庫存為負數(shù)的情況，于是乎就出現(xiàn)了超發(fā)現(xiàn)象。

測試方法：根據(jù)書上是html中使用js，for循環(huán)異步請求，發(fā)現(xiàn)并不會造成超發(fā)現(xiàn)象，后改為在瀏覽器中同時開啟多個窗口訪問/test進行搶購，模擬多個用戶搶購的場景，內(nèi)存爆炸...

為了解決這種問題，下面將介紹三種解決方法：

1、悲觀鎖

發(fā)生超發(fā)現(xiàn)象的根本原因是共享數(shù)據(jù)被多個線程所修改，無法保證其執(zhí)行順序，如果一個數(shù)據(jù)庫事務讀取到一個產(chǎn)品后，就將數(shù)據(jù)直接鎖定，不允許其他線程進行讀寫操作，直至當前數(shù)據(jù)庫事務完成才釋放這條數(shù)據(jù)的鎖，就不會發(fā)生超發(fā)現(xiàn)象，但是執(zhí)行效率性能將大大下降。

修改ProductMapper中的SQL語句：

@Mapper
public interface ProductMapper {

    @Select("SELECT id, product_name as productName, stock, price, version, note FROM t_product where id=#{id} for update")
    ProductPo getProduct(Long id);

    @Update("UPDATE t_product SET stock = stock - #{quantity} WHERE id = #{id}")
    int decreaseProduct(@Param("id") Long id, @Param("quantity") int quantity);
}

在select語句末尾添加了for update，這樣，在數(shù)據(jù)庫事務執(zhí)行的過程中，就會鎖定查詢出來的數(shù)據(jù)，其他事務將不能再對其進行讀寫，單個請求直至數(shù)據(jù)庫事務完成，才會釋放這個鎖，下圖可見其stock為0，沒有發(fā)生超發(fā)現(xiàn)象，但執(zhí)行效率下降了，通過購買記錄可以得知，相比之前沒加鎖慢了1/5。

庫存

2、樂觀鎖

為了解決悲觀鎖帶來的性能下降的問題，我們來討論一下樂觀鎖的原理：

樂觀鎖是一種不使用數(shù)據(jù)庫鎖和不阻塞線程并發(fā)的方案，下圖是以本Demo為例的樂觀鎖流程：

流程圖

這種方案就是多線程的概念CAS（Compare and Swap）,然而這樣的方案會引發(fā)一種ABA問題：

T1時刻：線程1讀取商品庫存為A

T2時刻：線程2讀取商品庫存為A

T3時刻：線程2計算購買商品總價格

T4時刻：當前庫存為A，與線程2保存的舊值一致，因此線程2可減庫存（當前庫存A--->B），此時線程1在當前庫存為B的情況下計算剩余商品價格（單價*B）。

T5時刻：線程2取消購買，線程2回退（當前庫存B--->A），線程1計算的剩余商品價格錯誤。

T6時刻：線程1比較舊值與當前數(shù)據(jù)庫庫存，發(fā)現(xiàn)都為A，返回之前計算好的（單價*B）結(jié)果，造成了錯誤。

從上面的分析中看到一個現(xiàn)象A--->B--->A的過程，就是所謂的ABA問題，解決此問題的方法為加入版本號的限制，只要在操作過程中修改共享值，無論業(yè)務正常，回退，還是異常，版本號只能遞增，不能回退遞減。每次通過比較數(shù)據(jù)的版本號來查看此數(shù)據(jù)是否被修改過。

@Mapper
public interface ProductMapper {

    @Select("SELECT id, product_name as productName, stock, price, version, note FROM t_product where id=#{id}")
    ProductPo getProduct(Long id);

    //********************change******************************
    @Update("UPDATE t_product SET stock = stock - #{quantity}, version = version + 1 WHERE id = #{id} and version = #{version}")
    int decreaseProduct(@Param("id") Long id, @Param("quantity") int quantity, @Param("version") int version);
}

     @Override
     // 啟動Spring數(shù)據(jù)庫事務機制
     @Transactional
     public boolean purchase(Long userId, Long productId, int quantity) {
         // 獲取產(chǎn)品
         ProductPo product = productMapper.getProduct(productId);
         // 比較庫存和購買數(shù)量
         if (product.getStock() < quantity) {
         // 庫存不足
         return false;
         }
         //**************************change*******************************
         // 扣減庫存,加入了version
         productMapper.decreaseProduct(productId, quantity, product.getVersion());
         //***************************************************************
         // 初始化購買記錄
         PurchaseRecordPo pr = this.initPurchaseRecord(userId, product, quantity);
         // 插入購買記錄
         purchaseRecordMapper.insertPurchaseRecord(pr);
         return true;
     }

產(chǎn)品表

發(fā)現(xiàn)stock并沒有降為0，原因是加入了版本號的判斷，所以大量的請求得到了失敗的結(jié)果，而且失敗率有點高。要解決這個方法，就設定為如果失敗，就重試，直至成功，但是這樣又會造成大量SQL執(zhí)行，影響性能，所以一般可以使用限制時間或者重入次數(shù)的方法來克服。

時間戳限制重入的樂觀鎖：

將一個請求限制在100ms的生存期，如果在100ms內(nèi)發(fā)生版本號沖突而導致不能更新的，則會重新嘗試請求，否則請求失敗。

修改service下PurchaseserviceImpl的purchase方法

 @Override
     // 啟動Spring數(shù)據(jù)庫事務機制
     @Transactional
     public boolean purchase(Long userId, Long productId, int quantity) {

         long start = System.currentTimeMillis();
         while (true){
             long end = System.currentTimeMillis();
             if (end - start >100){
                 return false;
             }
             // 獲取產(chǎn)品
             ProductPo product = productMapper.getProduct(productId);
             // 比較庫存和購買數(shù)量
             if (product.getStock() < quantity) {
                 // 庫存不足
                 return false;
             }
             // 扣減庫存
             int result = productMapper.decreaseProduct(productId, quantity, product.getVersion());

             // 如果數(shù)據(jù)更新失敗，說明數(shù)據(jù)在多線程中被其他線程修改
             // 導致失敗，著通過循環(huán)重入嘗試購買商品
             if (result == 0){
                 continue;
             }
             // 初始化購買記錄
             PurchaseRecordPo pr = this.initPurchaseRecord(userId, product, quantity);
             // 插入購買記錄
             purchaseRecordMapper.insertPurchaseRecord(pr);
             return true;
         }
     }

這種方法在測試中效果并不是很好，執(zhí)行速度很慢，沖突現(xiàn)象并沒有減少，反而增多，可能是我測試方法并不好，只開了三個網(wǎng)頁來模擬并發(fā)，不太懂JS，Demo用的JS是發(fā)送異步請求的，但用單窗口測試了好多次都沒出現(xiàn)超發(fā)現(xiàn)象，只能人肉模擬并發(fā)。

限定次數(shù)重入的樂觀鎖：

     @Override
     // 啟動Spring數(shù)據(jù)庫事務機制
     @Transactional
     public boolean purchase(Long userId, Long productId, int quantity) {

         long start = System.currentTimeMillis();
         for (int i=0; i<3; i++){
             // 獲取產(chǎn)品
             ProductPo product = productMapper.getProduct(productId);
             // 比較庫存和購買數(shù)量
             if (product.getStock() < quantity) {
                 // 庫存不足
                 return false;
             }
             // 扣減庫存
             int result = productMapper.decreaseProduct(productId, quantity, product.getVersion());

             // 如果數(shù)據(jù)更新失敗，說明數(shù)據(jù)在多線程中被其他線程修改
             // 導致失敗，著通過循環(huán)重入嘗試購買商品
             if (result == 0){
                 continue;
             }
             // 初始化購買記錄
             PurchaseRecordPo pr = this.initPurchaseRecord(userId, product, quantity);
             // 插入購買記錄
             purchaseRecordMapper.insertPurchaseRecord(pr);
             return true;
         }
         return false;
     }

這種方式比上一種限定時間好，速度和單純使用樂觀鎖差不多，并且消除了沖突。

3、Redis處理高并發(fā)

在高并發(fā)環(huán)境中，直接操作數(shù)據(jù)庫的方式過于緩慢，因為數(shù)據(jù)庫是一個寫入磁盤的過程，這個速度沒有寫入內(nèi)存的Redis快，Redis的機制也能夠幫助我們克服超發(fā)現(xiàn)象，但是，因為其命令方式運算能力比較薄弱，所以往往采用Redis Lua去代替它原有的命令方式。Redis Lua在Redis的執(zhí)行中是局內(nèi)原子性的，但他被執(zhí)行時不會被其他客戶端發(fā)送過來的命令打斷，通過這樣一種機制，可以在需要高并發(fā)的環(huán)境下考慮使用Redis去代替數(shù)據(jù)庫作為響應用戶的數(shù)據(jù)載體。但是Redis存儲具有不穩(wěn)定性，所以還需要有一定的機制將Redis存儲的數(shù)據(jù)刷入數(shù)據(jù)庫。

下面先來配置一下Redis:

application.properties

#配置redis
spring.redis.jedis.pool.min-idle=5
spring.redis.jedis.pool.max-active=10
spring.redis.jedis.pool.max-idle=10
spring.redis.jedis.pool.max-wait=2000
spring.redis.port=6379
spring.redis.host=127.0.0.1
#我的Redis沒有密碼
#spring.redis.password=123456
spring.redis.timeout=1000

pom.xml

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <version>2.1.4.RELEASE</version>
    <exclusions>
        <!--不依賴Redis的異步客戶端lettuce -->
        <exclusion>
            <groupId>io.lettuce</groupId>
            <artifactId>lettuce-core</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<!--引入Redis的客戶端驅(qū)動jedis -->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>

PurchaseServiceImpl.java，使用Redis Lua響應請求

@Service
public class PurchaseServiceImpl implements PurchaseService {


    @Autowired
    private ProductMapper productMapper = null;

    @Autowired
    private PurchaseRecordMapper purchaseRecordMapper = null;
   
    private PurchaseRecordPo initPurchaseRecord(Long userId, ProductPo product, int quantity) {
        PurchaseRecordPo pr = new PurchaseRecordPo();
        pr.setNote("購買日志，時間：" + System.currentTimeMillis());
        pr.setPrice(product.getPrice());
        pr.setProductId(product.getId());
        pr.setQuantity(quantity);
        double sum = product.getPrice() * quantity;
        pr.setSum(sum);
        pr.setUserId(userId);
        return pr;
    }

    @Autowired
    StringRedisTemplate stringRedisTemplate = null;
    String purchaseScript =
            // 先將產(chǎn)品編號保存到集合中
            " redis.call('sadd', KEYS[1], ARGV[2]) \n"
                    // 購買列表
                    + "local productPurchaseList = KEYS[2]..ARGV[2] \n"
                    // 用戶編號
                    + "local userId = ARGV[1] \n"
                    // 產(chǎn)品key
                    + "local product = 'product_'..ARGV[2] \n"
                    // 購買數(shù)量
                    + "local quantity = tonumber(ARGV[3]) \n"
                    // 當前庫存
                    + "local stock = tonumber(redis.call('hget', product, 'stock')) \n"
                    // 價格
                    + "local price = tonumber(redis.call('hget', product, 'price')) \n"
                    // 購買時間
                    + "local purchase_date = ARGV[4] \n"
                    // 庫存不足，返回0
                    + "if stock < quantity then return 0 end \n"
                    // 減庫存
                    + "stock = stock - quantity \n"
                    + "redis.call('hset', product, 'stock', tostring(stock)) \n"
                    // 計算價格
                    + "local sum = price * quantity \n"
                    // 合并購買記錄數(shù)據(jù)
                    + "local purchaseRecord = userId..','..quantity..','"
                    + "..sum..','..price..','..purchase_date \n"
                    // 保存到將購買記錄保存到list里
                    + "redis.call('rpush', productPurchaseList, purchaseRecord) \n"
                    // 返回成功
                    + "return 1 \n";
    // Redis購買記錄集合前綴
    private static final String PURCHASE_PRODUCT_LIST = "purchase_list_";
    // 搶購商品集合
    private static final String PRODUCT_SCHEDULE_SET = "product_schedule_set";
    // 32位SHA1編碼，第一次執(zhí)行的時候先讓Redis進行緩存腳本返回
    private String sha1 = null;

    @Override
    public boolean purchaseRedis(Long userId, Long productId, int quantity) {
        // 購買時間
        Long purchaseDate = System.currentTimeMillis();
        Jedis jedis = null;
        try {
            // 獲取原始連接
            jedis = (Jedis) stringRedisTemplate
                    .getConnectionFactory().getConnection().getNativeConnection();
            // 如果沒有加載過，則先將腳本加載到Redis服務器，讓其返回sha1
            if (sha1 == null) {
                sha1 = jedis.scriptLoad(purchaseScript);
            }
            // 執(zhí)行腳本，返回結(jié)果
            Object res = jedis.evalsha(sha1, 2, PRODUCT_SCHEDULE_SET,
                    PURCHASE_PRODUCT_LIST, userId + "", productId + "",
                    quantity + "", purchaseDate + "");
            Long result = (Long) res;
            return result == 1;
        } finally {
            // 關閉jedis連接
            if (jedis != null && jedis.isConnected()) {
                jedis.close();
            }
        }
    }

    @Override
    // 當運行方法啟用新的獨立事務運行
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    // 保存購買記錄，持久化到數(shù)據(jù)庫
    public boolean dealRedisPurchase(List<PurchaseRecordPo> prpList) {
        for (PurchaseRecordPo prp : prpList) {
            purchaseRecordMapper.insertPurchaseRecord(prp);
            productMapper.decreaseProduct(prp.getProductId(), prp.getQuantity());
        }
        return true;
    }
}

使用定時機制，定時將數(shù)據(jù)持久化到數(shù)據(jù)庫：

首先設置啟動文件：

@SpringBootApplication(scanBasePackages = "com.wayne.springboot")
@MapperScan(annotationClass = Mapper.class, basePackages = "com.wayne.springboot")
// 啟動springboot的定時機制，為此需要一個定時的方法來提供服務
// 把Redis的數(shù)據(jù)導入到數(shù)據(jù)庫
@EnableScheduling
public class SpringBootShoppingApplication{

    public static void main(String[] args) {
        SpringApplication.run(SpringBootShoppingApplication.class, args);
    }
}

一個定時的方法來提供服務把Redis的數(shù)據(jù)導入到數(shù)據(jù)庫：

import com.wayne.springboot.pojo.PurchaseRecordPo;
import com.wayne.springboot.service.PurchaseService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

@Service
public class TaskServiceImpl implements TaskService {
    @Autowired
    private StringRedisTemplate stringRedisTemplate = null;
    @Autowired
    private PurchaseService purchaseService = null;

    private static final String PRODUCT_SCHEDULE_SET = "product_schedule_set";
    private static final String PURCHASE_PRODUCT_LIST = "purchase_list_";
    // 每次取出1000條，避免一次取出消耗太多內(nèi)存
    private static final int ONE_TIME_SIZE = 1000;

    @Override
    // 每天半夜1點鐘開始執(zhí)行任務
//    @Scheduled(cron = "0 0 1 * * ?")
    // 下面是用于測試的配置，每分鐘執(zhí)行一次任務
    @Scheduled(fixedRate = 1000 * 30)
    public void purchaseTask() {
        System.out.println("定時任務開始......");
        Set<String> productIdList
                = stringRedisTemplate.opsForSet().members(PRODUCT_SCHEDULE_SET);
        List<PurchaseRecordPo> prpList =new ArrayList<>();
        for (String productIdStr : productIdList) {
            Long productId = Long.parseLong(productIdStr);
            String purchaseKey = PURCHASE_PRODUCT_LIST + productId;
            BoundListOperations<String, String> ops
                    = stringRedisTemplate.boundListOps(purchaseKey);
            // 計算記錄數(shù)
            long size = stringRedisTemplate.opsForList().size(purchaseKey);
            Long times = size % ONE_TIME_SIZE == 0 ?
                    size / ONE_TIME_SIZE : size / ONE_TIME_SIZE + 1;
            for (int i = 0; i < times; i++) {
                // 獲取至多TIME_SIZE個搶紅包信息
                List<String> prList = null;
                if (i == 0) {
                    prList  = ops.range(i * ONE_TIME_SIZE,
                            (i + 1) * ONE_TIME_SIZE);
                } else {
                    prList = ops.range(i * ONE_TIME_SIZE + 1,
                            (i + 1) * ONE_TIME_SIZE);
                }
                for (String prStr : prList) {
                    PurchaseRecordPo prp
                            = this.createPurchaseRecord(productId, prStr);
                    prpList.add(prp);
                }
                try {
                    // 采用該方法采用新建事務的方式，這樣不會導致全局事務回滾
                    purchaseService.dealRedisPurchase(prpList);
                } catch(Exception ex) {
                    ex.printStackTrace();
                }
                // 清除列表為空，等待重新寫入數(shù)據(jù)
                prpList.clear();
            }
            // 刪除購買列表

            stringRedisTemplate.delete(purchaseKey);
            // 從商品集合中刪除商品
            stringRedisTemplate.opsForSet()
                    .remove(PRODUCT_SCHEDULE_SET, productIdStr);
        }
        System.out.println("定時任務結(jié)束......");
    }

    private PurchaseRecordPo createPurchaseRecord(
        Long productId, String prStr) {
        String[] arr = prStr.split(",");
        Long userId = Long.parseLong(arr[0]);
        int quantity = Integer.parseInt(arr[1]);
        double sum = Double.valueOf(arr[2]);
        double price = Double.valueOf(arr[3]);
        Long time = Long.parseLong(arr[4]);
        Timestamp purchaseTime = new Timestamp(time);
        PurchaseRecordPo pr = new PurchaseRecordPo();
        pr.setProductId(productId);
        pr.setPurchaseTime(purchaseTime);
        pr.setPrice(price);
        pr.setQuantity(quantity);
        pr.setSum(sum);
        pr.setUserId(userId);
        pr.setNote("購買日志，時間：" + purchaseTime.getTime());
        return pr;
    }
}

到這里基本完成，啟動項目前先啟動redis服務器，并初始化Redis：

hmset product_1 id 1 stock 10000 price 2.00

然后啟動并訪問瀏覽器localhost:8080/test，因為設定的間隔為30s，所以等30s去查看數(shù)據(jù)庫。性能相比之前要快上數(shù)倍。

產(chǎn)品表

源碼存放在github-spring-boot-shopping