本文將圍繞高并發(fā)場(chǎng)景中的限流和秒殺需求綜合演示Spring Boot整合JPA、Redis緩存和RabbitMQ消息隊(duì)列的做法。

本項(xiàng)目將通過(guò)整合Springboot和Redis以及Lua腳本來(lái)實(shí)現(xiàn)限流和秒殺的效果，將通過(guò)RabbitMQ消息隊(duì)列來(lái)實(shí)現(xiàn)異步保存秒殺結(jié)果的效果。

一、項(xiàng)目概述

本項(xiàng)目將要實(shí)現(xiàn)的秒殺是指商家在某個(gè)時(shí)間段以非常低的價(jià)格銷售商品的一種營(yíng)銷活動(dòng)。

由于商品價(jià)格非常低，因此單位時(shí)間內(nèi)發(fā)起購(gòu)買商品的請(qǐng)求會(huì)非常多，從而會(huì)對(duì)系統(tǒng)造巨大的壓力。對(duì)此，在一些秒殺系統(tǒng)中往往會(huì)整合限流的功能，同時(shí)會(huì)通過(guò)消息隊(duì)列異步地保存秒殺結(jié)果。

本章將要實(shí)現(xiàn)的限流和秒殺功能歸納如下:

(1)通過(guò)Spring Boot的控制器類對(duì)外接收秒殺請(qǐng)求。

(2)針對(duì)請(qǐng)求進(jìn)行限流操作，比如秒殺商品的數(shù)量是10個(gè)，就限定在秒殺開始后的20秒內(nèi)只有100個(gè)請(qǐng)求能參加秒殺，該操作是通過(guò)Redis來(lái)實(shí)現(xiàn)的。

(3)通過(guò)限流檢驗(yàn)的這些請(qǐng)求將會(huì)同時(shí)競(jìng)爭(zhēng)若干個(gè)秒殺商品。該操作將通過(guò)基于Redis的Lua腳本來(lái)實(shí)現(xiàn)。

(4)為了降低數(shù)據(jù)庫(kù)的壓力，秒殺成功的記錄將通過(guò)RabbitMQ隊(duì)列以異步的方式記錄到數(shù)據(jù)庫(kù)中。

(5)同時(shí)，將通過(guò)RestTemple對(duì)象以多線程的方式模擬發(fā)送秒殺請(qǐng)求，以此來(lái)觀察本秒殺系統(tǒng)的運(yùn)行效果。

也就是說(shuō)，本系統(tǒng)會(huì)綜合用到Spring Boot、JPA、Redis和RabbitMQ，相關(guān)組件之間的關(guān)系如圖所示。

二、基于Redis的Lua腳本分析

Lua使用標(biāo)準(zhǔn)C語(yǔ)言開發(fā)而成的，它是一種輕量級(jí)的腳本語(yǔ)言，可嵌入基于Redis等的應(yīng)用程序中。Lua腳本可以駐留在內(nèi)存中，所以具有較高的性能，適用于處理高并發(fā)的場(chǎng)景。

Lua腳本的特性

Lua腳本語(yǔ)言是由巴西一所大學(xué)的Roberto lerusalimschy 、 Waldemar Celes和 LnHenrique de Figuciredo設(shè)計(jì)而成的，它具有如下兩大特性

(1)輕量性:Lua只具有一些核心和最基本的庫(kù)，所以非常輕便，非常適合嵌入由其他語(yǔ)言編寫的代碼中。

(2)擴(kuò)展性:Lua語(yǔ)言中預(yù)留了擴(kuò)展接口和相關(guān)擴(kuò)展機(jī)制，這樣在Lua語(yǔ)言中就能很方便地引入其他開發(fā)語(yǔ)言的功能，

本章給出的秒殺場(chǎng)景中會(huì)向Redis服務(wù)器發(fā)送多條指令，為了降低網(wǎng)絡(luò)調(diào)用的開銷，會(huì)把相關(guān)Redis命令放在Lua腳本里。通過(guò)調(diào)用Lua腳本只需要耗費(fèi)少量的網(wǎng)絡(luò)調(diào)用代價(jià)就能執(zhí)行多條Redis命令。

此外，秒殺相關(guān)的Redis語(yǔ)句還需要具備原子性，即這些語(yǔ)句要么全都執(zhí)行，要么全都不執(zhí)行。而Lua腳本是作為一個(gè)整體來(lái)執(zhí)行的，所以可以充分地確保相關(guān)秒殺語(yǔ)句的原子性。

在Redis中引入Lua腳本

在啟動(dòng)Redis服務(wù)器以后，可以通過(guò)redis-cli命令運(yùn)行l(wèi)ua腳本，具體步驟如下:

• 可以在C:work\redisConf\lua目錄中創(chuàng)建redisCallLua.lua文件，在其中編寫Lua腳本,注意，Lua腳本文件的擴(kuò)展名一般都是.lua。
• 在第一步創(chuàng)建的redisCallLua.lua文件中加入一行代碼，在其中通過(guò)redis.call命令執(zhí)行set name Peter的命令，

redis.call('set', 'name', 'Peter')

通過(guò)rdis.call方法在Redis中調(diào)用Lua腳本時(shí)，第一個(gè)參數(shù)是Redis命令，比如這里是set，第二個(gè)參數(shù)以及之后的參數(shù)是執(zhí)行該條Redis命令的參數(shù)。

• 通過(guò)如下的--eval命令執(zhí)行第二步定義的Lua腳本，其中C:work\redisConf\lua是這條Lua腳本所在的路徑，而redisCallLua.lua是腳本名。

redis-cli --eval C:\work\redisConf\lua\redisCallLua.lua

上述命令運(yùn)行后，得到的返回結(jié)果是空(nil)，原因是該Lua腳本只是通過(guò)set命令設(shè)置了值，并沒(méi)有返回結(jié)果。不過(guò)通過(guò)get name命令就能看到通過(guò)這條Lua腳本緩存的name值，具體是Peter。

如果Lua腳本包含的語(yǔ)句很少，那么還可以直接用eval命令來(lái)執(zhí)行該腳本，具體做法是，
先通過(guò)redis-cli語(yǔ)句連接到Redis服務(wù)器，隨后再執(zhí)行如下eval命令：

eval "redis.call('set','BookName','Spring Boot')" 0

從上述語(yǔ)句中能看到，在該條eval命令之后通過(guò)雙引號(hào)引入了待執(zhí)行的Lua腳本，在該腳本中依然是通過(guò)redis.call語(yǔ)句執(zhí)行Redis的set命令，進(jìn)行設(shè)置緩存的操作。

在該eval命令之后還指定了Lua腳本中KEYS類型參數(shù)的個(gè)數(shù)，這里是0，表示該Lua腳本沒(méi)有KEYS類型的參數(shù)。注意，這里設(shè)置的是KEYS類型的參數(shù)，而不是ARGV類型的參數(shù)，下文將詳細(xì)說(shuō)明這兩種參數(shù)的差別。

Lua腳本的返回值和參數(shù)

在Lua腳本中，可以通過(guò)retum語(yǔ)句返回執(zhí)行的結(jié)果，這部分對(duì)應(yīng)的語(yǔ)法比較簡(jiǎn)單。

同時(shí)，Redis在通過(guò)eval命令執(zhí)行Lua腳本時(shí)，可以傳入KEYS和ARGV這兩種不同類型的參數(shù)，它們的區(qū)別是，可以用KEYS參數(shù)來(lái)傳入Redis命令所需要的參數(shù)，可以用ARGV參數(shù)來(lái)傳入自定義的參數(shù)，通過(guò)如下兩個(gè)eval執(zhí)行Lua腳本的命令，可以看到這兩種參數(shù)的差別。

127.0.0.1:6379> eval "return {KEYS[1],ARGV[1],ARGV[2]" 1 keyono argvone argvtwo
1) "keyone"
2) "argvone"
3) "argvtwo"
127.0.0.1:6379> eval "return {KEYS[1].ARGV[1],ARGV[2]}" 2 keyone argvone argvtwo
1) "key1"
2) "argvtwo"

在第1行eval語(yǔ)句中，KEYS[1]表示KEYS類型的第一個(gè)參數(shù)，而ARGV[1]和ARGV[2]對(duì)應(yīng)地表示第一個(gè)和第二個(gè)ARGV類型的參數(shù)。

在第1行eval語(yǔ)句中，雙引號(hào)之后的1表示KEYS類型的參數(shù)個(gè)數(shù)是1，所以統(tǒng)計(jì)參數(shù)個(gè)數(shù)時(shí)并不把ARGV自定義類型的參數(shù)統(tǒng)計(jì)在內(nèi)，隨后的keyone, argvone和argvtwo分別對(duì)應(yīng)KEYS[1]、ARGV[1]和ARGV[2].

執(zhí)行第一行對(duì)應(yīng)的Lua腳本時(shí),會(huì)看到如第2~4行所示的輸出結(jié)果,這里輸出了KEYS[1]、
ARGV[1]和ARGV[2]這3個(gè)參數(shù)對(duì)應(yīng)的值。

第5行腳本和第1行的差別是，表示KEYS參數(shù)個(gè)數(shù)的值從1變成了2。但這里第2個(gè)參數(shù)是ARGV類型的，而不是KEYS類型的，所以這條Lua腳本語(yǔ)句會(huì)拋棄第2個(gè)參數(shù)，即ARGV[1]，通過(guò)第6行和第7行的輸出結(jié)果能驗(yàn)證這點(diǎn)。

所以，在通過(guò)eval命令執(zhí)行Lua腳本時(shí)，一定要確保參數(shù)個(gè)數(shù)和類型的正確性。同時(shí)，這里再次提醒，eval命令之后傳入的參數(shù)個(gè)數(shù)是KEYS類型參數(shù)的個(gè)數(shù)，而不是ARGV類型的。

分支語(yǔ)句

在Lua腳本中，可以通過(guò)if…else語(yǔ)句來(lái)控制代碼的執(zhí)行流程，具體語(yǔ)法如下:

if(布爾表達(dá)式) then
	布爾表達(dá)式是true時(shí)執(zhí)行的語(yǔ)句
else
	布爾表達(dá)式是false時(shí)執(zhí)行的語(yǔ)句
end

通過(guò)如下的ifDemo.lua范例，讀者可以看到在Lua腳本中使用分支語(yǔ)句的做法。

if redis.call('exists','studentID')==1 then
return 'Existed'
else
redis.call('set','StudentID','001');
return 'Not Existed'
end

在第1行中，通過(guò)if語(yǔ)句判斷redis.call命令執(zhí)行的exists語(yǔ)句是否返回1，如果是，則表示StudentID鍵存在，就會(huì)執(zhí)行第2行的returm 'Existed’語(yǔ)句返回Existed，否則走第3行的else流程，執(zhí)行第4行和第5行的語(yǔ)句，設(shè)置StudentID的值，并通過(guò)retum語(yǔ)句返回Not Existed。

由此可以看到在Lua腳本中使用if分支語(yǔ)句的做法。該腳本的運(yùn)行結(jié)果是:第一次運(yùn)行時(shí)，由于StudentID鍵不存在，因此會(huì)走else流程，從而看到Not Existed的輸出，而在第二次運(yùn)行時(shí)，由于此時(shí)該鍵已經(jīng)存在，因此會(huì)直接輸出’Existed’的結(jié)果。

三、實(shí)現(xiàn)限流和秒殺功能

本節(jié)將要?jiǎng)?chuàng)建的QuickBuyDemo項(xiàng)目中，一方面會(huì)用到上文提到的Lua腳本實(shí)現(xiàn)限流和秒殺的功能，另一方面將通過(guò)RabbitMQ消息隊(duì)列實(shí)現(xiàn)異步保存秒殺結(jié)果的功能。

創(chuàng)建項(xiàng)目并編寫配置文件

可以在IDEA集成開發(fā)環(huán)境中創(chuàng)建名為QuickBuyDemo的Maven項(xiàng)目，在該項(xiàng)目的pom.xml文件中通過(guò)如下關(guān)鍵代碼引入所需要的依賴包:

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.httpcomponents</groupId>
        <artifactId>httpclient</artifactId>
        <version>4.5.5</version>
    </dependency>
    <dependency>
        <groupId>org.apache.httpcomponents</groupId>
        <artifactId>httpcore</artifactId>
        <version>4.4.10</version>
    </dependency>
</dependencies>

這里通過(guò)第2-5行代碼引入了SpringBoot的依賴包，通過(guò)第6-9行代碼引入了RabbitMQ消息隊(duì)列相關(guān)的依賴包，通過(guò)第10-13行代碼引入了Redis相關(guān)的依賴包，通過(guò)第14-23行代碼引入了HTTP客戶端相關(guān)的依賴包，在本項(xiàng)目中將通過(guò)HTTP客戶端模擬客戶請(qǐng)求，從而驗(yàn)證秒殺效果。

在本項(xiàng)目resources目錄的application.properties配置文件中，將通過(guò)如下代碼配置消息隊(duì)列和Redis緩存:

rabbitmq.host=127.0.0.1
rabbitmq.port=5672
rabbitmq.username=guest
rabbitmq.password=guest

redis.host=localhost
redis.port=6379

在該配置文件中，通過(guò)第1~4行代碼配置了RabbitMQ的連接參數(shù)，通過(guò)第5行和第6行代碼配置了Redis的連接參數(shù)。

編寫啟動(dòng)類和控制器類

本項(xiàng)目的啟動(dòng)類如下，由于和大多數(shù)的Spring Boot項(xiàng)目啟動(dòng)類完全一致，因此不再重復(fù)講述。

package prj;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class SpringBootApp {
    public static void main(String[] args) {
        SpringApplication.run(SpringBootApp.class, args);
    }
}

本項(xiàng)目的控制器類代碼如下，在該Controller控制器類的第11-25行代碼中封裝了實(shí)現(xiàn)秒殺服務(wù)的quickBuy方法，該方法是以quickBuy/{item}/{person}格式的URL請(qǐng)求對(duì)外提供服務(wù)的，其中item參數(shù)表示商品，而person參數(shù)則表示商品的購(gòu)買人。

package prj.controller;

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 prj.receiver.BuyService;
@RestController
public class Controller {
    @Autowired
    private BuyService buyService;
    @RequestMapping("/quickBuy/{item}/{person}")
    public String quickBuy(@PathVariable String item, @PathVariable String person){
        //20秒里限流100個(gè)請(qǐng)求
        if(buyService.canVisit(item, 20,100)) {
            String result = buyService.buy(item, person);
            if (!result.equals("0")) {
                return person + " success";
            } else {
                return person + " fail";
            }
        }
        else{
            return person + " fail";
     }
}

在quickBuy方法中，首先通過(guò)第14行的buyService.canVisit方法對(duì)請(qǐng)求進(jìn)行了限流操作，這里在20秒中只允許有100個(gè)請(qǐng)求訪問(wèn)，如果通過(guò)限流驗(yàn)證，那么會(huì)繼續(xù)通過(guò)第15行的buyService.buy方法進(jìn)行秒殺操作。注意，這里的實(shí)現(xiàn)限流和秒殺功能的代碼都封裝在第10行定義的BuyService類中。

消息隊(duì)列的相關(guān)配置

在本項(xiàng)目的RabbitMQConfig類中將配置RabbitMQ的消息隊(duì)列和消息交換機(jī)，具體代碼如下：

package prj;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig{
    //定義含主題的消息隊(duì)列
    @Bean
    public Queue objectQueue() {
        return new Queue("buyRecordQueue");
    }
    //定義交換機(jī)
    TopicExchange myExchange() {
        return new TopicExchange("myExchange");
    Binding bindingObjectQueue(Queue objectQueue,TopicExchange exchange) {
        return BindingBuilder.bind(objectQueue).to(exchange).with("buyRecordQueue");
}

其中通過(guò)第9行的objectQueue方法創(chuàng)建了名為buyRecordQucue的消息隊(duì)列，該消息隊(duì)同將向用戶傳輸秒殺的結(jié)果，通過(guò)第14行的myExchange方法創(chuàng)建了名為myExhnge的清息交換機(jī)，并通過(guò)第18行的bindingObjectQueue方法根據(jù)buyRecordQucue主題綁定了上述消息以列和消息交換機(jī)。

實(shí)現(xiàn)秒殺功能的Lua腳本

在本項(xiàng)目中，實(shí)現(xiàn)秒殺效果的Lua腳本代碼如下:

local item = KEYS[1]
local person = ARGV[1]
local left = tonumber(redis.call('get',item))
if (left>=1) then
	redis.call ('decrby',item,1)
	redis.call ('rpush", 'personList',person)
	return 1
else

在該腳本中，首先通過(guò)KEYS[1]參數(shù)傳入待秒殺的商品，并賦予item對(duì)象，再通過(guò)ARGV[1]參數(shù)傳入發(fā)起秒殺請(qǐng)求的用戶，并賦子person對(duì)象。

隨后在第3行中，通過(guò)get item命令從Redis緩存中獲取該商品還有多少庫(kù)存，再通過(guò)第4行的if語(yǔ)句進(jìn)行判斷。

如果發(fā)現(xiàn)該商品剩余的庫(kù)存數(shù)量大于等于1，就會(huì)執(zhí)行第5~7行的Lua腳本，先通過(guò)decrby命令把庫(kù)存數(shù)減1，再調(diào)用rpush命令記錄當(dāng)前秒殺成功的用戶，并通過(guò)第7行的return語(yǔ)句返回1，表示秒殺成功。如果發(fā)現(xiàn)庫(kù)存數(shù)已經(jīng)小于1，那么會(huì)直接通過(guò)第9行的語(yǔ)句返且0，表示秒殺失敗。

在業(yè)務(wù)實(shí)現(xiàn)類中實(shí)現(xiàn)限流和秒殺

在BuyService.java中，將調(diào)用Redis和Lua腳本實(shí)現(xiàn)限流和秒殺的功能，具體代碼如下：

package prj.receiver;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import prj.model.buyrecord;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;
@Service
public class BuyService {
    @Resource
    private RedisTemplate redisTemplate;
    @Autowired
    private AmqpTemplate amqpTemplate;
    public boolean canVisit(String item, int limitTime, int limitNum)  {
        long curTime = System.currentTimeMillis();
        // 在zset里存入請(qǐng)求
        redisTemplate.opsForZSet().add(item, curTime, curTime);
        // 移除時(shí)間范圍外的請(qǐng)求
        redisTemplate.opsForZSet().removeRangeByScore(item,0,curTime - limitTime * 1000);
        // 統(tǒng)計(jì)時(shí)間范圍內(nèi)的請(qǐng)求個(gè)數(shù)
        Long count = redisTemplate.opsForZSet().zCard(item);
        // 統(tǒng)一設(shè)置所有請(qǐng)求的超時(shí)時(shí)間
        redisTemplate.expire(item, limitTime, TimeUnit.SECONDS);
        return limitNum >= count;
     }
    public String buy(String item, String person){
        String luaScript = "local person = ARGV[1]\n" +
                           "local item = KEYS[1] \n" +
                           "local left = tonumber(redis.call('get',item)) \n" +
                           "if (left >= 1) \n" +
                           "then redis.call('decrby',item,1) \n" +
                           " redis.call('rpush','personList',person) \n" +
                           "return 1 \n" +
                           "else \n" +
                           "return 0\n" +
                           "end\n" +
                           "\n" ;
        String key=item;
        String args=person;
        DefaultRedisScript<String> redisScript = new DefaultRedisScript<String>();
        redisScript.setScriptText(luaScript);
        //調(diào)用lua腳本，請(qǐng)注意傳入的參數(shù)
        Object luaResult = redisTemplate.execute((RedisConnection connection) -> connection.eval(
                    redisScript.getScriptAsString().getBytes(),
                    ReturnType.INTEGER,
                    1,
                    key.getBytes(),
                    args.getBytes()));
        //如果秒殺成功，向消息隊(duì)列發(fā)消息，異步插入到數(shù)據(jù)庫(kù)
        if(!luaResult.equals("0") ){
            buyrecord record = new buyrecord();
            record.setItem(item);
            record.setPerson(person);
            amqpTemplate.convertAndSend("myExchange","buyRecordQueue",record);
        }
        //根據(jù)lua腳本的執(zhí)行情況返回結(jié)果
        return luaResult.toString();
    }
}

在上述代碼中，首先通過(guò)第2-11行的import語(yǔ)句引入了本類所要用到的依賴包，隨后在第15行中定義了調(diào)用Redis會(huì)用到的redisTemplate對(duì)象，在第17行中定義了向RabbitMQ消息隊(duì)列發(fā)送消息所要用到的amqpTemplate對(duì)象。

第18行的canVisit方法實(shí)現(xiàn)了限流效果，該方法的item參數(shù)表示待限流的商品，limitTime和LimitNum參數(shù)分別表示在指定時(shí)間內(nèi)需要限流的請(qǐng)求個(gè)數(shù)。

在該方法中使用Redis的有序集合實(shí)現(xiàn)了限流效果，具體的做法是，在第21行的代碼中，通過(guò)zadd方法把表示操作類型的item作為鍵插入有序集合，插入時(shí)用表示當(dāng)前時(shí)間的curTime作為值，以保證值的唯一性，同樣再用curTime值作為有序集合中元素的score值。

隨后在第23行中，通過(guò)removeRangeByScore命令移除從0到距當(dāng)前時(shí)間limitTime范圍內(nèi)的數(shù)據(jù)，比如限流的時(shí)間范圍是20秒，那么通過(guò)這條命令就能在有序集合中移除score范圍從0到距離當(dāng)前時(shí)間20秒的數(shù)據(jù)，從而確保有序集合只保存最近20秒內(nèi)的請(qǐng)求。

在此基礎(chǔ)上，通過(guò)第25行代碼用zcard命令統(tǒng)計(jì)有序集合內(nèi)鍵為item的個(gè)數(shù)，如果通過(guò)第28行的布爾語(yǔ)句發(fā)現(xiàn)當(dāng)前個(gè)數(shù)還沒(méi)達(dá)到限流的上限，該方法就會(huì)返回true，表示該請(qǐng)求能繼續(xù)，否則返回false，表示該請(qǐng)求將會(huì)被限流。

同時(shí)，需要通過(guò)第27行的expire語(yǔ)句設(shè)置有序集合中數(shù)據(jù)的超時(shí)時(shí)間，這樣就能確保在限流以及秒殺動(dòng)作完成后這些鍵能自動(dòng)刪除。

第30行定義的buy方法將會(huì)實(shí)現(xiàn)秒殺的功能，其中先通過(guò)第31~41行代碼定義實(shí)現(xiàn)秒殺功能的Lua腳本，該腳本之前分析過(guò)，隨后再通過(guò)第47一52行代碼使用redisTemplate.execute方法執(zhí)行這段Lua腳本。

在執(zhí)行時(shí)，會(huì)通過(guò)第50行代碼指定KEYS類型參數(shù)的個(gè)數(shù)，通過(guò)第51行和第52行代碼傳入該腳本執(zhí)行時(shí)所需要用到的KEYS和ARGVS參數(shù)。

隨后會(huì)通過(guò)第54行的f語(yǔ)句判斷秒殺腳本的執(zhí)行結(jié)果，如果秒殺成功，那么會(huì)通過(guò)第55~58行代碼用amqpTemplate對(duì)象向buyRecordQueue隊(duì)列發(fā)送包含秒殺結(jié)果的record對(duì)象。最后，再通過(guò)第61行的語(yǔ)句返回秒殺的結(jié)果。

觀察秒殺效果

至此，可以通過(guò)如下步驟啟動(dòng)Redis、RabbitMQ和QuickBuyDemo項(xiàng)目，并觀察秒殺效果。

• 在命令行中通過(guò)rabbitmq-server.bat start命令啟動(dòng)RabbitMQ。
• 通過(guò)運(yùn)行redis-server.exe啟動(dòng)Redis服務(wù)器,并通過(guò)運(yùn)行redis-cli.exe啟動(dòng)Redis客戶端，隨后在Redis客戶端通過(guò)set Computer 10命令向Redis中緩存一條庫(kù)存數(shù)據(jù),表示有10個(gè)Computer可供秒殺。
• 在QuickBuyDemo項(xiàng)目中，通過(guò)運(yùn)行SpringBootApp.java啟動(dòng)類啟動(dòng)該項(xiàng)目。成功啟動(dòng)后，在瀏覽器中輸入http:localhost:8080/quickBuy/Computer/Tom發(fā)起秒殺請(qǐng)求，其中Computer參數(shù)表示秒殺的商品，而Tom則表示發(fā)起秒殺請(qǐng)求的人。

輸入后，能在瀏覽器中看到Tom success的結(jié)果，隨后到Redis客戶端窗口運(yùn)行get Computer命令，能看到Computer的庫(kù)存數(shù)量會(huì)降到9，由此可以確認(rèn)秒殺成功。同時(shí)，可以通過(guò)lindex personList 0命令觀察到成功發(fā)起秒殺請(qǐng)求的人是Tom。

四、以異步方式保存秒殺結(jié)果

如果在上述QuickBuyDemo項(xiàng)目中直接把秒殺結(jié)果插入MySQL數(shù)據(jù)庫(kù)，那么當(dāng)秒殺請(qǐng)求并發(fā)量很高時(shí)會(huì)對(duì)數(shù)據(jù)庫(kù)造成很大的壓力，所以在該項(xiàng)目中會(huì)通過(guò)消息隊(duì)列把秒殺結(jié)果傳輸?shù)紻BHandlerPrj項(xiàng)目中，用異步的方式保存數(shù)據(jù)，從而降低數(shù)據(jù)庫(kù)的負(fù)載壓力。

創(chuàng)建項(xiàng)目并設(shè)計(jì)數(shù)據(jù)庫(kù)

首先需要?jiǎng)?chuàng)建名為DBHandlerPrj的Maven項(xiàng)目，在其中實(shí)現(xiàn)異步保存秒殺數(shù)據(jù)的功能，該項(xiàng)目的pom.xml文件如下，其中通過(guò)第2-5行代碼引入了Spring Boot依賴包，通過(guò)第6-9行代碼引入了RabbitMO消息隊(duì)列的依賴包，通過(guò)第10~18行代碼引入了JPA和MySQL的依賴包。

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
</dependencies>

本項(xiàng)目將會(huì)用到如表所示的buyrecord表，該表是創(chuàng)建在本地MySQL的QuickBuy數(shù)據(jù)表(schema)中的，在其中將會(huì)保存秒殺結(jié)果。

而本項(xiàng)目的啟動(dòng)類SpringBootAppjava和QuickBuyDemo項(xiàng)目中的完全一致,所以不再重復(fù)說(shuō)明。

配置消息隊(duì)列和數(shù)據(jù)庫(kù)參數(shù)

在本項(xiàng)目resources目錄的application.yml文件中，將通過(guò)如下代碼配置消息隊(duì)列和數(shù)據(jù)庫(kù)連接參數(shù)。

server:
  port: 8090
rabbitmq:
  host: 127.0.0.1
  port: 5672
  username: guest
  password: guest
spring:
  jpa:
    show-sql: true
    hibernate:
      dll-auto: validate
  datasource:
    url: jdbc:mysql://localhost:3306/QuickBuy?serverTimezone=GMT
    username: root
    password: 123456
    driver-class-name: com.mysql.jdbc.Driver

由于之前的QuickBuyDemo項(xiàng)目已經(jīng)占用了8080端口，因此本配置文件將通過(guò)第1行和第2行代碼設(shè)置工作端口為8090。隨后，本配置文件將通過(guò)第3~7行代碼設(shè)置RabbiMQ消息隊(duì)列的連接參數(shù)，具體是連接到本地5672端口，且連接所用的用戶名和密碼都是guest。

由于本項(xiàng)目是通過(guò)JPA的方式連接MySQL庫(kù)的,因此本配置文件通過(guò)第8-12行代碼配置了JPA的參數(shù)，通過(guò)第13-17行代碼配置了MySQL的連接參數(shù)。

此外，和QuickBuyDemo項(xiàng)目一樣，本項(xiàng)目依然是在RabbitMQConfg.java配置文件中設(shè)置RabbitMQ消息隊(duì)列和交換機(jī)，具體代碼如下，其中配置的消息隊(duì)列名字buyRecordQueue與交換機(jī)的名字myExchange需要和QuickBuyDemo項(xiàng)目中的定義保持一致。

package prj;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig{
    //定義含主題的消息隊(duì)列
    @Bean
    public Queue objectQueue() {
        return new Queue("buyRecordQueue");
    }
    //定義交換機(jī)
    TopicExchange myExchange() {
        return new TopicExchange("myExchange");
    Binding bindingObjectQueue(Queue objectQueue,TopicExchange exchange) {
        return BindingBuilder.bind(objectQueue).to(exchange).with("buyRecordQueue");
}

監(jiān)聽消息隊(duì)列并保存秒殺結(jié)果

在本項(xiàng)目的QuickBuySevivce.java文件中將會(huì)監(jiān)聽buyRecordQueue消息隊(duì)列，并把秒殺結(jié)果存入MySOL數(shù)據(jù)表，具體代碼如下:

package prj.service;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import prj.model.buyrecord;
import prj.repo.BuyRecordRepo;

@Component
@RabbitListener(queues = "buyRecordQueue")
public class QuickBuyService {
    @Autowired
    private AmqpTemplate amqpTemplate;
    private BuyRecordRepo buyRecordRepo;
    @RabbitHandler
    public void saveBuyRecord(buyrecord record){
        buyRecordRepo.save(record);
    }
}

在本類的第10行通過(guò)@RabbitListener注解說(shuō)明將要監(jiān)聽buyRecordQueue消息隊(duì)列，當(dāng)該消息隊(duì)列有消息時(shí)，會(huì)觸發(fā)本類第17行的saveBuyRecord方法，該方法被第16行的@RabbitHandler注解所修飾。在該方法中會(huì)調(diào)用JPA類buyRecordRepo的save方法向數(shù)據(jù)表中保存秒殺結(jié)果。

QuickBuyServce類中用到的模型類buyrecord和QuickBuyDemo項(xiàng)目中的很相似,由于該類需要通過(guò)消息隊(duì)列在網(wǎng)絡(luò)中傳輸，因此需要像第9行那樣實(shí)現(xiàn)Serializable接口。

package prj.model;

import java.io.Serializable;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.Table;
@Entity
@Table(name="buyrecord")
public class buyrecord implements Serializable {
    @Id
    @Column(name = "person")
    private String person;
    @Column(name = "item")
    private String item;
    public void setItem(String item) {
        this.item = item;
    }
    public void setPerson(String person) {
        this.person = person;
    public String getItem() {
        return item;
    public String getPerson() {
        return person;
}

全鏈路效果演示

開發(fā)好上述兩個(gè)項(xiàng)目以后，可以用對(duì)如下步驟觀察全鏈路的秒殺效果：

• 啟動(dòng)RabbitMQ、Redis服務(wù)器和客戶端，通過(guò)set Computer 10命令緩存秒殺商品的數(shù)量，同時(shí)通過(guò)運(yùn)行啟動(dòng)類
• 啟動(dòng)QuickBuyDemo項(xiàng)目。

啟動(dòng)DBHandlerPrj項(xiàng)目

在QuickBuyDemo項(xiàng)日中開發(fā)如下的QuickBuyThread.java文件，在其中用多線程的方式模擬多個(gè)秒殺情求，代碼如下：

package prj.client;

import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;
class QuickBuyThread extends Thread{
    public void run() {
        RestTemplate restTemplate = new RestTemplate();
        String user = Thread.currentThread().getName();
        ResponseEntity<String> entity = restTemplate.
                getForEntity("http://localhost:8080/quickBuy/Computer/"+user , String.class);
        System.out.println(entity.getBody());
    }
}
public class MockQuickBuy {
    public static void main(String[] args){
        for (int i = 0; i < 15; i++) {
            new QuickBuyThread().start();
        }

第4行定義的QuickBuyThread類以繼承Thread類的方式實(shí)現(xiàn)了線程的效果，在第5行線程的run方法中用restTemplate.getForEntity方法模擬發(fā)送了秒殺的請(qǐng)求，其中用當(dāng)前線程的名字作為發(fā)起秒殺的用戶。

public class MockQuickBuy {
    public static void main(String[] args){
        for (int i = 0; i < 15; i++) {
            new QuickBuyThread().start();
        }
    }

}

在第12行MockQuickBuy類的main方法中，通過(guò)第14行的for循環(huán)啟動(dòng)了15個(gè)線程發(fā)起秒殺請(qǐng)求。由于之前在Redis緩存中設(shè)置的Computer商品數(shù)量是10個(gè)，因此會(huì)有10個(gè)請(qǐng)求秒殺成功。5個(gè)請(qǐng)求不成功。如下輸出語(yǔ)句能確認(rèn)這一結(jié)果。

此外，如果再到 MySQL數(shù)據(jù)庫(kù)用select from QuickBuy.buyrecord語(yǔ)句觀察秒殺結(jié)果，能看到成功秒殺的用戶，這些用戶名和上述輸出結(jié)果中的用戶名完全一致。

本文來(lái)自于《Spring Boot+Vue.js+分布式組件全棧開發(fā)訓(xùn)練營(yíng)(視頻教學(xué)版)》第17章

到此這篇關(guān)于Spring+Redis+RabbitMQ限流和秒殺項(xiàng)目的開發(fā)的文章就介紹到這了,更多相關(guān)Spring Redis RabbitMQ限流秒殺內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論