Redis學(xué)習(xí)

• 一:認(rèn)識消息隊列
• 二:基于List實現(xiàn)消息隊列
• • 1.基于List結(jié)構(gòu)模擬消息隊列
  • 2.基于List的消息隊列有哪些優(yōu)缺點(diǎn)
• 三:基于PubSub的消息隊列
• • 1.說明
  • 2.基于PubSub的消息隊列的優(yōu)缺點(diǎn)
• 四:Redis消息隊列-基于Stream的消息隊列
• • 1.發(fā)送消息的命令
  • 2.讀取消息的方式之一：XREAD
  • • 2.1.使用示例
    • 2.2.問題
• 五:基于Stream的消息隊列-消費(fèi)者組
• • 1.常用命令
  • 2.消費(fèi)者監(jiān)聽消息的基本思路
  • 3.STREAM類型消息隊列的XREADGROUP命令特點(diǎn)
  • 4.對比
• 六: 基于Redis的Stream結(jié)構(gòu)作為消息隊列，實現(xiàn)異步秒殺下單
• • 1.需求
  • 2.代碼
• 七:總結(jié)

10年積累的成都做網(wǎng)站、網(wǎng)站制作經(jīng)驗，可以快速應(yīng)對客戶對網(wǎng)站的新想法和需求。提供各種問題對應(yīng)的解決方案。讓選擇我們的客戶得到更好、更有力的網(wǎng)絡(luò)服務(wù)。我雖然不認(rèn)識你，你也不認(rèn)識我。但先網(wǎng)站設(shè)計制作后付款的網(wǎng)站建設(shè)流程，更有介休免費(fèi)網(wǎng)站建設(shè)讓你可以放心的選擇與我們合作。一:認(rèn)識消息隊列

什么是消息隊列：字面意思就是存放消息的隊列。最簡單的消息隊列模型包括3個角色：

• 消息隊列：存儲和管理消息，也被稱為消息代理（Message Broker）
• 生產(chǎn)者：發(fā)送消息到消息隊列
• 消費(fèi)者：從消息隊列獲取消息并處理消息

使用隊列的好處在于 解耦， 所謂解耦，舉一個生活中的例子就是：快遞員(生產(chǎn)者)把快遞放到快遞柜里邊(Message Queue)去，我們(消費(fèi)者)從快遞柜里邊去拿東西，這就是一個異步，如果耦合，那么這個快遞員相當(dāng)于直接把快遞交給你，這事固然好，但是萬一你不在家，那么快遞員就會一直等你，這就浪費(fèi)了快遞員的時間，所以這種思想在我們?nèi)粘ｉ_發(fā)中，是非常有必要的。

這種場景在我們秒殺中就變成了：我們下單之后，利用redis去進(jìn)行校驗下單條件，再通過隊列把消息發(fā)送出去，然后再啟動一個線程去消費(fèi)這個消息，完成解耦，同時也加快我們的響應(yīng)速度。

這里我們可以使用一些現(xiàn)成的mq，比如kafka，rabbitmq等等,我們也可以直接使用redis提供的mq方案，降低我們的部署和學(xué)習(xí)成本。

二:基于List實現(xiàn)消息隊列 1.基于List結(jié)構(gòu)模擬消息隊列

消息隊列（Message Queue），字面意思就是存放消息的隊列。而Redis的list數(shù)據(jù)結(jié)構(gòu)是一個雙向鏈表，很容易模擬出隊列效果。

隊列是入口和出口不在一邊，因此我們可以利用：LPUSH 結(jié)合 RPOP、或者 RPUSH 結(jié)合 LPOP來實現(xiàn)。
不過要注意的是，當(dāng)隊列中沒有消息時RPOP或LPOP操作會返回null，并不像JVM的阻塞隊列那樣會阻塞并等待消息。因此這里應(yīng)該使用BRPOP或者BLPOP來實現(xiàn)阻塞效果。

2.基于List的消息隊列有哪些優(yōu)缺點(diǎn)

優(yōu)點(diǎn)：

• 利用Redis存儲，不受限于JVM內(nèi)存上限
• 基于Redis的持久化機(jī)制，數(shù)據(jù)安全性有保證
• 可以滿足消息有序性

缺點(diǎn)：

• 無法避免消息丟失
• 只支持單消費(fèi)者

三:基于PubSub的消息隊列 1.說明

PubSub（發(fā)布訂閱）是Redis2.0版本引入的消息傳遞模型。顧名思義，消費(fèi)者可以訂閱一個或多個channel，生產(chǎn)者向?qū)?yīng)channel發(fā)送消息后，所有訂閱者都能收到相關(guān)消息。

SUBSCRIBE channel [channel] ：訂閱一個或多個頻道
PUBLISH channel msg ：向一個頻道發(fā)送消息
PSUBSCRIBE pattern[pattern] ：訂閱與pattern格式匹配的所有頻道

2.基于PubSub的消息隊列的優(yōu)缺點(diǎn)

優(yōu)點(diǎn)：

• 采用發(fā)布訂閱模型，支持多生產(chǎn)、多消費(fèi)

缺點(diǎn)：

• 不支持?jǐn)?shù)據(jù)持久化
• 無法避免消息丟失
• 消息堆積有上限，超出時數(shù)據(jù)丟失.數(shù)據(jù)不會持久化,但是會緩存在客戶端

四:Redis消息隊列-基于Stream的消息隊列

Stream 是 Redis 5.0 引入的一種新數(shù)據(jù)類型，可以實現(xiàn)一個功能非常完善的消息隊列。

1.發(fā)送消息的命令

例如：

2.讀取消息的方式之一：XREAD 2.1.使用示例

例如，使用XREAD讀取第一個消息：

XREAD阻塞方式，讀取最新的消息：

在業(yè)務(wù)開發(fā)中，我們可以循環(huán)的調(diào)用XREAD阻塞方式來查詢最新消息，從而實現(xiàn)持續(xù)監(jiān)聽隊列的效果，偽代碼如下

2.2.問題

注意：當(dāng)我們指定起始ID為$時，代表讀取最新的消息，如果我們處理一條消息的過程中，又有超過1條以上的消息到達(dá)隊列，則下次獲取時也只能獲取到最新的一條，會出現(xiàn)漏讀消息的問題

STREAM類型消息隊列的XREAD命令特點(diǎn)：

• 消息可回溯
• 一個消息可以被多個消費(fèi)者讀取
• 可以阻塞讀取
• 有消息漏讀的風(fēng)險

五:基于Stream的消息隊列-消費(fèi)者組

消費(fèi)者組（Consumer Group）：將多個消費(fèi)者劃分到一個組中，監(jiān)聽同一個隊列。具備下列特點(diǎn)：

1.常用命令

創(chuàng)建消費(fèi)者組：

XGROUP CREATE KEY GROUPNAME ID [MKSTREAM] 

key：隊列名稱
groupName：消費(fèi)者組名稱
ID：起始ID標(biāo)示，$代表隊列中最后一個消息，0則代表隊列中第一個消息
MKSTREAM：隊列不存在時自動創(chuàng)建隊列
其它常見命令：

刪除指定的消費(fèi)者組

XGROUP DESTORY key groupName

給指定的消費(fèi)者組添加消費(fèi)者

XGROUP CREATECONSUMER key groupname consumername

刪除消費(fèi)者組中的指定消費(fèi)者

XGROUP DELCONSUMER key groupname consumername

從消費(fèi)者組讀取消息：

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]

• group：消費(fèi)組名稱
• consumer：消費(fèi)者名稱，如果消費(fèi)者不存在，會自動創(chuàng)建一個消費(fèi)者
• count：本次查詢的大數(shù)量
• BLOCK milliseconds：當(dāng)沒有消息時最長等待時間
• NOACK：無需手動ACK，獲取到消息后自動確認(rèn)
• STREAMS key：指定隊列名稱
• ID：獲取消息的起始ID：

“>”：從下一個未消費(fèi)的消息開始
其它：根據(jù)指定id從pending-list中獲取已消費(fèi)但未確認(rèn)的消息，例如0，是從pending-list中的第一個消息開始

2.消費(fèi)者監(jiān)聽消息的基本思路

3.STREAM類型消息隊列的XREADGROUP命令特點(diǎn)

• 消息可回溯
• 可以多消費(fèi)者爭搶消息，加快消費(fèi)速度
• 可以阻塞讀取
• 沒有消息漏讀的風(fēng)險
• 有消息確認(rèn)機(jī)制，保證消息至少被消費(fèi)一次

4.對比

六: 基于Redis的Stream結(jié)構(gòu)作為消息隊列，實現(xiàn)異步秒殺下單 1.需求

• 創(chuàng)建一個Stream類型的消息隊列，名為stream.orders
• 修改之前的秒殺下單Lua腳本，在認(rèn)定有搶購資格后，直接向stream.orders中添加消息，內(nèi)容包含voucherId、userId、orderId
• 項目啟動時，開啟一個線程任務(wù)，嘗試獲取stream.orders中的消息，完成下單

2.代碼

Lua腳本

-- 1.參數(shù)列表

--1.1 優(yōu)惠卷id,用戶id,訂單id
local voucherId = ARGV[1]
local userId = ARGV[2]
local orderId = ARGV[3]

-- 2.key 庫存key,訂單key
local stockKey = 'secKill:stock:' .. voucherId
local orderKey = 'secKill:order:' .. voucherId

-- 判斷庫存是否充足
if (tonumber(redis.call('get', stockKey)<= 0) then
-- 不充足,返回1
    return 1
end;

--判斷用戶是否下單 存在返回2
if(redis.call('sismember',orderKey,userId) == 1) then
    return 2
end

--扣庫存
redis.call('incrby', stockKey,-1)
--下單(保存用戶)
redis.call("sadd",orderKey,userId)
-- 發(fā)送消息隊列到隊列當(dāng)中
redis.call('xadd','stream.orders','*','userId',userId,'voucherId',voucherId,'id',orderId)
-- 成功
return 0

VoucherOrderServiceImpl

package com.***.flow.impl;

import cn.hutool.core.bean.BeanUtil;
import com.***.flow.entity.Vouchers;
import com.****.flow.service.FlowProcessService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.*;

@Slf4j
@Service
public class FlowProcessServiceImpl implements FlowProcessService {@Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    //線程池
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    //啟動時就開啟線程池執(zhí)行任務(wù)
    @PostConstruct
    private void init() {SECKILL_ORDER_EXECUTOR.submit(new VocherOrderHandler());
    }


    private class VocherOrderHandler implements Runnable {@Override
        public void run() {while (true) {try {//1.獲取消息隊列的訂單信息
                    // XREADGROUP GROUP g1 c1 count 1 block streams streams.order >List>list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("streams:order", ReadOffset.lastConsumed())

                    );

                    //2.判斷消息獲取是否成功
                    if (list == null || list.isEmpty()) {//2.1 獲取失敗,說明沒有消息,繼續(xù)下一次循環(huán)
                        continue;
                    }


                    //3.解析消息中的訂單信息
                    MapRecordvalues = list.get(0);
                    Vouchers vouchers = BeanUtil.fillBeanWithMap(values, new Vouchers(), true);

                    handleVoucherOrder(vouchers);


                    //4.ack確認(rèn)
                    stringRedisTemplate.opsForStream().acknowledge("streams:order", "g1", record.getId());
                } catch (Exception e) {//沒有被處理,消息在pendingList 里面

                    log.info("處理訂單異常:", e);
                    handlePendingList();
                }
            }
        }
    }

    private void handlePendingList() {while (true) {try {//1.獲取pending_list里面的訂單信息
                // XREADGROUP GROUP g1 c1 count 1  streams streams.order 0
                List>list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create("streams:order", ReadOffset.from("2"))

                );

                //2.判斷消息獲取是否成功
                if (list == null || list.isEmpty()) {//2.1 獲取失敗,說明沒有異常信息,結(jié)束循環(huán)
                    break;
                }


                //3.解析消息中的訂單信息
                MapRecordvalues = list.get(0);
                Vouchers vouchers = BeanUtil.fillBeanWithMap(values, new Vouchers(), true);

                handleVoucherOrder(vouchers);


                //4.ack確認(rèn)
                stringRedisTemplate.opsForStream().acknowledge("streams:order", "g1", record.getId());
            } catch (Exception e) {//沒有被處理,消息在pendingList 里面
                log.info("處理訂單異常:", e);
            }
        }
    }
}

  

}

七:總結(jié)

• 這個stream 內(nèi)容有些復(fù)雜,只需要記一下邏輯就好
• 黑馬redis 入門到入土

你是否還在尋找穩(wěn)定的海外服務(wù)器提供商？創(chuàng)新互聯(lián)www.cdcxhl.cn海外機(jī)房具備T級流量清洗系統(tǒng)配攻擊溯源，準(zhǔn)確流量調(diào)度確保服務(wù)器高可用性，企業(yè)級服務(wù)器適合批量采購，新人活動首月15元起，快前往官網(wǎng)查看詳情吧

網(wǎng)頁名稱：redis:從入門到入土:6.Redis消息隊列-創(chuàng)新互聯(lián)
本文地址：http://www.rwnh.cn/article32/dgsopc.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供面包屑導(dǎo)航、小程序開發(fā)、服務(wù)器托管、全網(wǎng)營銷推廣、品牌網(wǎng)站設(shè)計、云服務(wù)器

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)