根據(jù)上一篇Demo測(cè)試情況反映，當(dāng)有多個(gè)線程同時(shí)搶購(gòu)時(shí)俊马，會(huì)發(fā)生超發(fā)現(xiàn)象丁存，所謂超發(fā)現(xiàn)象，就是原本設(shè)置庫(kù)存為30000件柴我，但是解寝，當(dāng)搶購(gòu)?fù)瓿珊蟀l(fā)現(xiàn)庫(kù)存余量變成了負(fù)數(shù)，即發(fā)貨量大于庫(kù)存量的情況：

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

造成這種現(xiàn)象的原因：當(dāng)多個(gè)線程請(qǐng)求數(shù)據(jù)庫(kù)查詢(xún)庫(kù)存余量時(shí)艘儒，顯示有余量聋伦，但是當(dāng)進(jìn)行扣減庫(kù)存時(shí)，庫(kù)存已經(jīng)用完了界睁，但那個(gè)線程并不知道觉增，依舊去扣減庫(kù)存，造成庫(kù)存為負(fù)數(shù)的情況翻斟，于是乎就出現(xiàn)了超發(fā)現(xiàn)象抑片。

測(cè)試方法：根據(jù)書(shū)上是html中使用js，for循環(huán)異步請(qǐng)求杨赤，發(fā)現(xiàn)并不會(huì)造成超發(fā)現(xiàn)象敞斋，后改為在瀏覽器中同時(shí)開(kāi)啟多個(gè)窗口訪問(wèn)/test進(jìn)行搶購(gòu)，模擬多個(gè)用戶(hù)搶購(gòu)的場(chǎng)景疾牲，內(nèi)存爆炸...

為了解決這種問(wèn)題植捎，下面將介紹三種解決方法：

1、悲觀鎖

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

修改ProductMapper中的SQL語(yǔ)句：

@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語(yǔ)句末尾添加了for update，這樣避消，在數(shù)據(jù)庫(kù)事務(wù)執(zhí)行的過(guò)程中低滩，就會(huì)鎖定查詢(xún)出來(lái)的數(shù)據(jù)，其他事務(wù)將不能再對(duì)其進(jìn)行讀寫(xiě)岩喷，單個(gè)請(qǐng)求直至數(shù)據(jù)庫(kù)事務(wù)完成恕沫，才會(huì)釋放這個(gè)鎖，下圖可見(jiàn)其stock為0纱意，沒(méi)有發(fā)生超發(fā)現(xiàn)象婶溯，但執(zhí)行效率下降了，通過(guò)購(gòu)買(mǎi)記錄可以得知偷霉，相比之前沒(méi)加鎖慢了1/5迄委。

庫(kù)存

2、樂(lè)觀鎖

為了解決悲觀鎖帶來(lái)的性能下降的問(wèn)題腾它，我們來(lái)討論一下樂(lè)觀鎖的原理：

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

流程圖

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

T1時(shí)刻：線程1讀取商品庫(kù)存為A

T2時(shí)刻：線程2讀取商品庫(kù)存為A

T3時(shí)刻：線程2計(jì)算購(gòu)買(mǎi)商品總價(jià)格

T4時(shí)刻：當(dāng)前庫(kù)存為A，與線程2保存的舊值一致瞒滴，因此線程2可減庫(kù)存（當(dāng)前庫(kù)存A--->B）曲梗，此時(shí)線程1在當(dāng)前庫(kù)存為B的情況下計(jì)算剩余商品價(jià)格（單價(jià)*B）。

T5時(shí)刻：線程2取消購(gòu)買(mǎi)妓忍，線程2回退（當(dāng)前庫(kù)存B--->A）虏两，線程1計(jì)算的剩余商品價(jià)格錯(cuò)誤。

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

從上面的分析中看到一個(gè)現(xiàn)象A--->B--->A的過(guò)程祖凫，就是所謂的ABA問(wèn)題琼蚯，解決此問(wèn)題的方法為加入版本號(hào)的限制，只要在操作過(guò)程中修改共享值惠况，無(wú)論業(yè)務(wù)正常遭庶，回退，還是異常稠屠，版本號(hào)只能遞增峦睡，不能回退遞減。每次通過(guò)比較數(shù)據(jù)的版本號(hào)來(lái)查看此數(shù)據(jù)是否被修改過(guò)权埠。

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

產(chǎn)品表

發(fā)現(xiàn)stock并沒(méi)有降為0榨了，原因是加入了版本號(hào)的判斷，所以大量的請(qǐng)求得到了失敗的結(jié)果攘蔽，而且失敗率有點(diǎn)高龙屉。要解決這個(gè)方法，就設(shè)定為如果失敗秩彤，就重試叔扼，直至成功，但是這樣又會(huì)造成大量SQL執(zhí)行漫雷，影響性能瓜富，所以一般可以使用限制時(shí)間或者重入次數(shù)的方法來(lái)克服。

時(shí)間戳限制重入的樂(lè)觀鎖：

將一個(gè)請(qǐng)求限制在100ms的生存期降盹，如果在100ms內(nèi)發(fā)生版本號(hào)沖突而導(dǎo)致不能更新的与柑，則會(huì)重新嘗試請(qǐng)求，否則請(qǐng)求失敗蓄坏。

修改service下PurchaseserviceImpl的purchase方法

 @Override
     // 啟動(dòng)Spring數(shù)據(jù)庫(kù)事務(wù)機(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);
             // 比較庫(kù)存和購(gòu)買(mǎi)數(shù)量
             if (product.getStock() < quantity) {
                 // 庫(kù)存不足
                 return false;
             }
             // 扣減庫(kù)存
             int result = productMapper.decreaseProduct(productId, quantity, product.getVersion());

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

這種方法在測(cè)試中效果并不是很好涡戳，執(zhí)行速度很慢结蟋，沖突現(xiàn)象并沒(méi)有減少，反而增多渔彰，可能是我測(cè)試方法并不好嵌屎，只開(kāi)了三個(gè)網(wǎng)頁(yè)來(lái)模擬并發(fā)，不太懂JS恍涂，Demo用的JS是發(fā)送異步請(qǐng)求的宝惰，但用單窗口測(cè)試了好多次都沒(méi)出現(xiàn)超發(fā)現(xiàn)象，只能人肉模擬并發(fā)再沧。

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

     @Override
     // 啟動(dòng)Spring數(shù)據(jù)庫(kù)事務(wù)機(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);
             // 比較庫(kù)存和購(gòu)買(mǎi)數(shù)量
             if (product.getStock() < quantity) {
                 // 庫(kù)存不足
                 return false;
             }
             // 扣減庫(kù)存
             int result = productMapper.decreaseProduct(productId, quantity, product.getVersion());

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

這種方式比上一種限定時(shí)間好，速度和單純使用樂(lè)觀鎖差不多淤堵，并且消除了沖突寝衫。

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

在高并發(fā)環(huán)境中粘勒，直接操作數(shù)據(jù)庫(kù)的方式過(guò)于緩慢竞端，因?yàn)閿?shù)據(jù)庫(kù)是一個(gè)寫(xiě)入磁盤(pán)的過(guò)程，這個(gè)速度沒(méi)有寫(xiě)入內(nèi)存的Redis快庙睡，Redis的機(jī)制也能夠幫助我們克服超發(fā)現(xiàn)象，但是技俐，因?yàn)槠涿罘绞竭\(yùn)算能力比較薄弱乘陪，所以往往采用Redis Lua去代替它原有的命令方式。Redis Lua在Redis的執(zhí)行中是局內(nèi)原子性的雕擂，但他被執(zhí)行時(shí)不會(huì)被其他客戶(hù)端發(fā)送過(guò)來(lái)的命令打斷啡邑，通過(guò)這樣一種機(jī)制，可以在需要高并發(fā)的環(huán)境下考慮使用Redis去代替數(shù)據(jù)庫(kù)作為響應(yīng)用戶(hù)的數(shù)據(jù)載體井赌。但是Redis存儲(chǔ)具有不穩(wěn)定性谤逼，所以還需要有一定的機(jī)制將Redis存儲(chǔ)的數(shù)據(jù)刷入數(shù)據(jù)庫(kù)。

下面先來(lái)配置一下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沒(méi)有密碼
#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>
        <!--不依賴(lài)Redis的異步客戶(hù)端lettuce -->
        <exclusion>
            <groupId>io.lettuce</groupId>
            <artifactId>lettuce-core</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<!--引入Redis的客戶(hù)端驅(qū)動(dòng)jedis -->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>

PurchaseServiceImpl.java仇穗，使用Redis Lua響應(yīng)請(qǐng)求

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

    @Override
    public boolean purchaseRedis(Long userId, Long productId, int quantity) {
        // 購(gòu)買(mǎi)時(shí)間
        Long purchaseDate = System.currentTimeMillis();
        Jedis jedis = null;
        try {
            // 獲取原始連接
            jedis = (Jedis) stringRedisTemplate
                    .getConnectionFactory().getConnection().getNativeConnection();
            // 如果沒(méi)有加載過(guò)枝冀，則先將腳本加載到Redis服務(wù)器，讓其返回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 {
            // 關(guān)閉jedis連接
            if (jedis != null && jedis.isConnected()) {
                jedis.close();
            }
        }
    }

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

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

首先設(shè)置啟動(dòng)文件：

@SpringBootApplication(scanBasePackages = "com.wayne.springboot")
@MapperScan(annotationClass = Mapper.class, basePackages = "com.wayne.springboot")
// 啟動(dòng)springboot的定時(shí)機(jī)制谷誓，為此需要一個(gè)定時(shí)的方法來(lái)提供服務(wù)
// 把Redis的數(shù)據(jù)導(dǎo)入到數(shù)據(jù)庫(kù)
@EnableScheduling
public class SpringBootShoppingApplication{

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

一個(gè)定時(shí)的方法來(lái)提供服務(wù)把Redis的數(shù)據(jù)導(dǎo)入到數(shù)據(jù)庫(kù)：

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點(diǎn)鐘開(kāi)始執(zhí)行任務(wù)
//    @Scheduled(cron = "0 0 1 * * ?")
    // 下面是用于測(cè)試的配置，每分鐘執(zhí)行一次任務(wù)
    @Scheduled(fixedRate = 1000 * 30)
    public void purchaseTask() {
        System.out.println("定時(shí)任務(wù)開(kāi)始......");
        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);
            // 計(jì)算記錄數(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個(gè)搶紅包信息
                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 {
                    // 采用該方法采用新建事務(wù)的方式捍歪，這樣不會(huì)導(dǎo)致全局事務(wù)回滾
                    purchaseService.dealRedisPurchase(prpList);
                } catch(Exception ex) {
                    ex.printStackTrace();
                }
                // 清除列表為空户辱，等待重新寫(xiě)入數(shù)據(jù)
                prpList.clear();
            }
            // 刪除購(gòu)買(mǎi)列表

            stringRedisTemplate.delete(purchaseKey);
            // 從商品集合中刪除商品
            stringRedisTemplate.opsForSet()
                    .remove(PRODUCT_SCHEDULE_SET, productIdStr);
        }
        System.out.println("定時(shí)任務(wù)結(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("購(gòu)買(mǎi)日志，時(shí)間：" + purchaseTime.getTime());
        return pr;
    }
}

到這里基本完成费封，啟動(dòng)項(xiàng)目前先啟動(dòng)redis服務(wù)器焕妙，并初始化Redis：

hmset product_1 id 1 stock 10000 price 2.00

然后啟動(dòng)并訪問(wèn)瀏覽器localhost:8080/test，因?yàn)樵O(shè)定的間隔為30s弓摘，所以等30s去查看數(shù)據(jù)庫(kù)焚鹊。性能相比之前要快上數(shù)倍。

產(chǎn)品表

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