Redis --- 秒殺優化方案（阻塞隊列+基于Stream流的消息隊列）

下面是我們的秒殺流程：

對于正常的秒殺處理，我們需要多次查詢數據庫，會給數據庫造成相當大的壓力，這個時候我們需要加入緩存，進而緩解數據庫壓力。

在上面的圖示中，我們可以將一條流水線的任務拆成兩條流水線來做，如果我們直接將判斷秒殺庫存與校驗一人一單放在流水線A上，剩下的放在另一條流水線B，那么如果流水線A就可以相當于服務員直接判斷是否符合資格，如果符合資格那么直接生成信息給另一條流水線B去處理業務，這里的流水線就是咱們的線程，而流水線A也是基于數據庫進行查詢，也會壓力數據庫，那么這種情況我們就可以將待查詢信息保存在Redis緩存中。

但是我們不能再流水線A判斷完成后去直接調用流水線B，這樣的效率是大打折扣的，這種情況我們需要開啟獨立線程去執行流水線B的操作，如何知道給哪個用戶創建訂單呢？這個時候就要流水線A在判斷成功后去生成信息給獨立線程。

最后的業務就變成，用戶直接訪問流水線A，通過流水線A去判斷，如果通過則生成信息給流水線B去創建訂單，過程如下圖：

那么什么樣的數據結構滿足下面條件：① 一個key能夠保存很多值? ?②唯一性：一人一單需要保證用戶id不能重復。

所以我們需要使用set：

那么如何判斷校驗用戶的購買資格呢？

?而上述判斷需要保證原子性，所以我們需要使用Lua腳本進行編寫：

local voucherId = ARGV[1]; -- 優惠劵id
local userId = ARGV[2]; -- 用戶id-- 庫存key
local stockKey = 'seckill:stock' .. voucherId; -- 拼接
-- 訂單key
local stockKey = 'seckill:stock' .. voucherId; -- 拼接
-- 判斷庫存是否充足
if(tonumber(redis.call('get',stockKey) <= 0)) then-- 庫存不足，返回1return 1;
end;
-- 判斷用戶是否下單
if(redis.call('sismember',orderKey,userId)) then-- 存在，說明重復下單，返回2return 2;
end
-- 扣減庫存 incrby stockKey -1
redis.call('incrby',stockKey,-1);
-- 下單（保存用戶） sadd orderKey userId
redis.call('sadd',orderKey,userId);
return 0;

之后我們按照下面步驟來實現代碼：

在方法體內執行Lua腳本來原子性判斷，然后判斷是否能夠處理并傳入阻塞隊列：

@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Autowiredprivate ISeckillVoucherService seckillVoucherService;@Autowiredprivate RedisIdWorker redisIdWorker;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate RedissonClient redissonClient;private static final DefaultRedisScript<Long> SECKILL_SCRIPT; // 泛型內填入返回值類型static { // 靜態屬性要使用靜態代碼塊進行初始化SECKILL_SCRIPT = new DefaultRedisScript<>();SECKILL_SCRIPT.setResultType(Long.class);SECKILL_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));}public Result seckillVoucherMax(Long voucherId) {// 獲取用戶信息Long userId = UserHolder.getUser().getId();// 1.執行Lua腳本來判斷用戶資格Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(), // Lua無需接受keyvoucherId.toString(),userId.toString());// 2.判斷結果是否為0int r = result.intValue();if(r != 0) {// 不為0代表無資格購買return Result.fail(r == 1 ? "庫存不足" : "不能重復下單");}// 3.有購買資格則將下單信息保存到阻塞隊列中// ... return Result.ok();}}

?接下來我們創建阻塞隊列，線程池以及線程方法，隨后使用Springboot提供的注解在@PostConstruct去給線程池傳入線程方法：

@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Autowiredprivate ISeckillVoucherService seckillVoucherService;@Autowiredprivate RedisIdWorker redisIdWorker;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate RedissonClient redissonClient;private static final DefaultRedisScript<Long> SECKILL_SCRIPT; // 泛型內填入返回值類型static { // 靜態屬性要使用靜態代碼塊進行初始化SECKILL_SCRIPT = new DefaultRedisScript<>();SECKILL_SCRIPT.setResultType(Long.class);SECKILL_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));}private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024); // 創建阻塞隊列private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();  // 創建線程池// 讓大類在開始初始化時就能夠執行線程任務@PostConstructprivate void init() {SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderTask());}// 創建線程任務private class VoucherOrderTask implements Runnable {@Overridepublic void run() {while(true){try {// 獲取隊列中的訂單信息VoucherOrder voucherOrder = orderTasks.take();// 取出頭部信息// 創建訂單handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error("處理訂單異常",e);}}}}// 創建訂單private void handleVoucherOrder(VoucherOrder voucherOrder) {RLock lock = redissonClient.getLock("lock:order:" + voucherOrder.getUserId().toString());boolean isLock = lock.tryLock();// 判斷是否獲取鎖成功if (!isLock) {// 獲取鎖失敗，返回錯誤或重試log.error("不允許重復下單");return ;}try {proxy.createVoucherOrderMax(voucherOrder);} finally {lock.unlock();}}@Overridepublic void createVoucherOrderMax(VoucherOrder voucherOrder) {// 一人一單Long userId = voucherOrder.getUserId();// 查詢訂單int count = query().eq("user_id",userId).eq("voucher_id", voucherOrder.getVoucherId()).count();// 判斷是否存在if(count > 0){// 用戶已經購買過log.error("用戶已經購買過");return ;}// CAS改進：將庫存判斷改成stock > 0以此來提高性能boolean success = seckillVoucherService.update().setSql("stock= stock -1") // set stock = stock - 1.eq("voucher_id", voucherOrder.getVoucherId()).eq("stock",0) // where id = ? and stock > 0.update();if (!success) {//扣減庫存log.error("庫存不足！");return ;}//6.創建訂單save(voucherOrder);}private IVoucherOrderService proxy; // 代理對象public Result seckillVoucherMax(Long voucherId) {// 獲取用戶信息Long userId = UserHolder.getUser().getId();// 1.執行Lua腳本來判斷用戶資格Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(), // Lua無需接受keyvoucherId.toString(),userId.toString());// 2.判斷結果是否為0int r = result.intValue();if(r != 0) {// 不為0代表無資格購買return Result.fail(r == 1 ? "庫存不足" : "不能重復下單");}// 3.有購買資格則將下單信息保存到阻塞隊列中Long orderId = redisIdWorker.nextId("order");// 創建訂單VoucherOrder voucherOrder = new VoucherOrder();voucherOrder.setId(orderId);voucherOrder.setUserId(userId);voucherOrder.setVoucherId(voucherId);// 放入阻塞隊列orderTasks.add(voucherOrder);// 4.獲取代理對象(線程異步執行，需要手動在方法內獲取)proxy = (IVoucherOrderService)AopContext.currentProxy(); // 獲取當前類的代理對象  (需要引入aspectjweaver依賴，并且在實現類加入@EnableAspectJAutoProxy(exposeProxy = true)以此來暴露代理對象)return Result.ok();}}

在上面代碼中，我們使用下面代碼創建了一個單線程的線程池。它保證所有提交的任務都按照提交的順序執行，每次只有一個線程在工作。

private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

下面代碼是一個常見的阻塞隊列實現，具有固定大小（在這里是 1024 * 1024），它的作用是緩沖和排隊任務。ArrayBlockingQueue 是一個線程安全的隊列，它會自動處理線程之間的同步問題。當隊列滿時，調用 put() 方法的線程會被阻塞，直到隊列有空間；當隊列為空時，調用 take() 方法的線程會被阻塞，直到隊列中有數據。

private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);

在下面代碼中，orderTasks 阻塞隊列用于存放需要處理的訂單對象，每個訂單的處理邏輯都由 VoucherOrderTask 線程池中的線程異步執行：

VoucherOrder voucherOrder = orderTasks.take();
handleVoucherOrder(voucherOrder);

之后我們需要調用 Runnable 接口去實現VoucherOrderTask類以此來創建線程方法：

private class VoucherOrderTask implements Runnable {@Overridepublic void run() {while (true) {try {// 獲取隊列中的訂單信息VoucherOrder voucherOrder = orderTasks.take(); // 獲取訂單// 創建訂單handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error("處理訂單異常", e);}}}
}

隨后將線程方法通過 submit() 方法將 VoucherOrderTask 提交到線程池中，這個任務是一個無限循環的任務，它會不斷從阻塞隊列中取出訂單并處理，直到線程池關閉。這種方式使得訂單處理任務可以異步執行，而不阻塞主線程，提高了系統的響應能力：

@PostConstruct
private void init() {SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderTask());
}

但是在高并發的情況下就會產生大量訂單，就會超出JVM阻塞隊列的上線，并且每當服務重啟或者宕機的情況發生，阻塞隊列的所有訂單任務就都會丟失。

所以為了解決這種情況，我們就要使用消息隊列去解決這個問題：

什么是消息隊列？

消息隊列（Message Queue, MQ）是一種用于在應用程序之間傳遞消息的通信方式。它允許應用程序通過發送和接收消息來解耦，從而提高系統的可擴展性、可靠性和靈活性。消息隊列通常用于異步通信、任務隊列、事件驅動架構等場景。

消息隊列的核心概念?：

生產者（Producer）：發送消息到消息隊列的應用程序。
消費者（Consumer）：從消息隊列中接收并處理消息的應用程序。
隊列（Queue）：消息的存儲區域，生產者將消息發送到隊列，消費者從隊列中獲取消息。
消息（Message）：在生產者與消費者之間傳遞的數據單元。
Broker：消息隊列的服務器，負責接收、存儲和轉發消息。

消息隊列是在JVM以外的一個獨立的服務，能夠不受JVM內存的限制，并且存入MQ的信息都可以做持久化存儲。

詳細教學可以查詢下面鏈接：微服務架構 --- 使用RabbitMQ進行異步處理?

但是這樣的方式是需要額外提供服務的，所以我們可以使用Redis提供的三種不同的方式來實現消息隊列：

List 結構實現消息隊列
Pub/Sub（發布/訂閱）模式
Stream 結構（Redis 5.0 及以上版本）（推薦使用）（詳細介紹）

使用 List 結構實現消息隊列：

Redis 的 List 數據結構是一個雙向鏈表，支持從頭部或尾部插入和彈出元素。我們可以利用?LPUSH?和?BRPOP?命令實現一個簡單的消息隊列。

實現步驟：

生產者：使用?LPUSH?將消息推入隊列。

消費者：使用?BRPOP?阻塞地從隊列中獲取消息。

生產者代碼：

import redis.clients.jedis.Jedis;public class ListProducer {public static void main(String[] args) {Jedis jedis = new Jedis("localhost", 6379); // 連接 RedisString queueName = "myQueue";// 發送消息for (int i = 1; i <= 5; i++) {String message = "Message " + i;jedis.lpush(queueName, message); // 將消息推入隊列System.out.println("Sent: " + message);}jedis.close(); // 關閉連接}
}

消費者代碼：

import redis.clients.jedis.Jedis;public class ListConsumer {public static void main(String[] args) {Jedis jedis = new Jedis("localhost", 6379); // 連接 RedisString queueName = "myQueue";while (true) {// 阻塞獲取消息，超時時間為 0（無限等待）var result = jedis.brpop(0, queueName);String message = result.get(1); // 獲取消息內容System.out.println("Received: " + message);}}
}

優點：簡單易用，適合輕量級場景。

缺點：不支持消息確認機制，消息一旦被消費（從隊列內取出）就會從隊列中刪除。并且只支持單消費者（一個消息只能拿出一次）

使用 Pub/Sub 模式實現消息隊列：?

Redis 的 Pub/Sub 模式是一種發布-訂閱模型，生產者將消息發布到頻道，消費者訂閱頻道以接收消息。

實現步驟：

生產者：使用?PUBLISH?命令向頻道發布消息。

消費者：使用?SUBSCRIBE?命令訂閱頻道。

生產者代碼：

import redis.clients.jedis.Jedis;public class PubSubProducer {public static void main(String[] args) {Jedis jedis = new Jedis("localhost", 6379); // 連接 RedisString channelName = "myChannel";// 發布消息for (int i = 1; i <= 5; i++) {String message = "Message " + i;jedis.publish(channelName, message); // 發布消息到頻道System.out.println("Published: " + message);}jedis.close(); // 關閉連接}
}

?消費者代碼：

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;public class PubSubConsumer {public static void main(String[] args) {Jedis jedis = new Jedis("localhost", 6379); // 連接 RedisString channelName = "myChannel";// 創建訂閱者JedisPubSub subscriber = new JedisPubSub() {@Overridepublic void onMessage(String channel, String message) {System.out.println("Received: " + message);}};// 訂閱頻道jedis.subscribe(subscriber, channelName);}
}

優點：支持一對多的消息廣播。

缺點：消息是即時的，如果消費者不在線，消息會丟失。

但是上面兩方式都是有缺點的：

不支持消息確認機制，消息一旦被消費（從隊列內取出）就會從隊列中刪除。并且只支持單消費者（一個消息只能拿出一次）
消息是即時的，如果消費者不在線，消息會丟失。

所以根據上面的兩種方式，我們推出一款全新的方式 ->

使用 Stream 結構實現消息隊列:

Redis Stream 是一種強大的數據結構，用于管理消息流。它將消息存儲在 Redis 中，并允許消費者按順序獲取消息。Stream 具有以下特點：

有序消息：消息按插入順序排列。
消費者組：一個消費者組可以有多個消費者，每個消費者可以獨立消費不同的消息。
消息 ID：每條消息都有唯一的 ID（如：1588890470850-0），ID 按時間戳生成。
自動分配消息：多個消費者可以從 Stream 中并行消費消息，保證消息不會重復消費。

在 Redis Stream 中，一個隊列可以有多個消費者組，每個消費者組可以獨立地消費隊列中的消息。每個消費者組內有多個消費者，而消費者是基于 消費者名稱 進行識別的。?

消費者組的工作方式：

每個消費者組擁有自己的 消費進度，也就是每個消費者組會從 自己獨立的消息 ID 開始消費。
多個消費者組之間是相互獨立的，即使它們消費的是同一個隊列，它們也可以從不同的位置開始消費隊列中的消息。
每個消費者組都可以有多個 消費者（在同一個組內，多個消費者可以并行消費同一個隊列的消息，但每個消息在消費者組內只能被一個消費者處理一次）。

假設有一個隊列（Stream）mystream，可以為它創建多個消費者組：
XGROUP CREATE mystream group1 $ MKSTREAM
XGROUP CREATE mystream group2 $ MKSTREAM
這樣，mystream 隊列上就有了兩個消費者組：group1 和 group2。每個消費者組可以有自己的消費者并從該隊列中讀取消息。此時，group1 和 group2 都在消費同一個隊列 mystream，但它們的消費進度是獨立的，它們各自有自己的消息 ID 記錄。

每個消費者組可以有多個消費者，而每個消費者通過一個 唯一的消費者名稱 來標識。

每個消費者組有獨立的消費進度

每個消費者組會記錄自己的消費進度，也就是它消費到隊列中的 哪個消息 ID。即使多個消費者組在消費同一個消息隊列，它們每個組都會從 不同的消費位置（消息 ID）開始讀取消息。

例如，假設有一個隊列 mystream，同時有兩個消費者組 group1 和 group2，它們都從 mystream 隊列中讀取消息：

group1 從 mystream 隊列中的消息 id1 開始消費，group1 的進度會記錄在 Redis 中。
group2 從 mystream 隊列中的消息 id2 開始消費，group2 的進度也會記錄在 Redis 中。

消費進度互不干擾，即便 group1 和 group2 都在消費 mystream 隊列，它們的消費位置是獨立的。

消費者組內部的消息消費

一個消費者組內的消費者會共享組內的消息。即使有多個消費者，每條消息 在消費者組內部只會被 一個消費者 消費。消費者之間會并行處理消息，但每條消息只會被一個消費者處理。

舉個例子：假設 group1 中有三個消費者 consumer1、consumer2、consumer3，如果隊列 mystream 有 6 條消息，那么它們會如下消費：

consumer1 處理消息 1、2
consumer2 處理消息 3、4
consumer3 處理消息 5、6

但對于消費者組 group2，如果它有自己的消費者，group2 內的消費者也會并行消費 mystream 中的消息，而 group1 和 group2 之間沒有直接關系。

首先初始化一個消息隊列：

在項目啟動時，確保 Redis 中存在對應的 Stream 和消費者組。可以通過程序在啟動時檢查并創建（如果不存在的話）。

@Configuration
public class RedisStreamConfig {@Autowiredprivate StringRedisTemplate redisTemplate;private static final String STREAM_KEY = "mystream";private static final String GROUP_NAME = "mygroup";@PostConstructpublic void init() {// 檢查消費者組是否存在，若不存在則創建try {// 如果消費者組不存在則會拋出異常，我們捕獲異常進行創建redisTemplate.opsForStream().groups(STREAM_KEY);} catch (Exception e) {// 創建消費者組，起始位置為 $ 表示從末尾開始消費新消息redisTemplate.opsForStream().createGroup(STREAM_KEY, GROUP_NAME);}}
}

注意：

opsForStream().groups(STREAM_KEY)：查詢消費者組是否已存在。
opsForStream().createGroup(STREAM_KEY, GROUP_NAME)：如果沒有消費者組，則創建一個新的組。

隨后我們生產者發送消息示例：

@Service  
public class RedisStreamProducerService {  // 定義生產者服務類 RedisStreamProducerServiceprivate static final String STREAM_KEY = "mystream";  // 定義 Redis Stream 的名稱，這里指定隊列名為 "mystream"@Autowired  private StringRedisTemplate redisTemplate;public void sendMessage(String content) {  // 定義一個方法，發送消息到 Redis Stream，參數 content 是消息的內容Map<String, String> map = new HashMap<>();  // 創建一個 Map 用來存儲消息內容map.put("content", content);  // 將消息內容添加到 Map 中，鍵是 "content"，值是傳入的內容// 在消息隊列中添加消息，調用 StringRedisTemplate 的 opsForStream 方法RecordId recordId = redisTemplate.opsForStream()  // 獲取操作 Redis Stream 的操作對象.add(StreamRecords.objectBacked(map)  // 創建一個 Stream 記錄，將 Map 轉化為對象記錄.withStreamKey(STREAM_KEY));  // 設置該記錄屬于的 Stream（消息隊列）的名稱// 輸出記錄的 ID，表示消息已經成功發送System.out.println("消息發送成功，id: " + recordId.getValue());  // 打印消息的 ID，表明該消息已經被成功加入到 Stream 中}
}

RecordId 是 Spring Data Redis 中的一個類，用來表示 消息的唯一標識符。它對應 Redis Stream 中的 消息 ID，該 ID 是 Redis Stream 中每條消息的唯一標識。Redis 中的消息 ID 通常是由時間戳和序號組成的（如 1588890470850-0）。

主要功能：

表示消息 ID：RecordId 是一個封裝類，表示 Redis Stream 中消息的 ID。
用于識別和操作消息：在消費和確認消息時，RecordId 用來標識每條消息的唯一性，并幫助 Redis 確定消息是否已經被消費。

使用場景：

RecordId 用來標識從 Stream 中讀取到的消息，我們可以通過 RecordId 來進行消息的確認、刪除或其他操作。
RecordId recordId = redisTemplate.opsForStream().add(StreamRecords.objectBacked(map).withStreamKey("mystream"));
通過 StreamRecords.objectBacked(map) 將 map 對象作為消息內容，并用 add 方法將其寫入 Stream。

在然后編寫消費者服務：

使用 RedisTemplate 的 read 方法（底層執行的是 XREADGROUP 命令）從消費者組中拉取消息，并進行處理。消費者可以采用定時任務或后臺線程不斷輪詢。

@Slf4j  
@Service  
public class RedisStreamConsumerService { private static final String STREAM_KEY = "mystream";  // Redis Stream 的名稱，這里指定隊列名為 "mystream"private static final String GROUP_NAME = "mygroup";  // 消費者組的名稱，多個消費者可以通過組名共享消費隊列private static final String CONSUMER_NAME = "consumer-1";  // 消費者的名稱，消費者名稱在同一消費者組內必須唯一@Autowired  private StringRedisTemplate redisTemplate;@PostConstruct  // 使用該注解能讓方法在 Spring 完成依賴注入后自動調用，用于初始化任務@Async  // 將該方法標記為異步執行，允許它在單獨的線程中運行，不會阻塞主線程，@EnableAsync 需要在配置類中啟用public void start() {  // 啟動方法，在應用啟動時執行// 無限循環，不斷從 Redis Stream 中讀取消息（可以改為定時任務等方式）while (true) {try {// 設置 Stream 讀取的阻塞超時，設置最多等待 2 秒StreamReadOptions options = StreamReadOptions.empty().block(Duration.ofSeconds(2));// 從指定的消費者組中讀取消息，">" 表示只消費未被消費過的消息List<MapRecord<String, Object, Object>> messages = redisTemplate.opsForStream().read(Consumer.from(GROUP_NAME, CONSUMER_NAME),  // 指定消費者組和消費者名稱options,  // 設置讀取選項，包含阻塞時間StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed())  // 從最后消費的消息開始讀取);// 如果沒有消息，繼續循環讀取if (messages == null || messages.isEmpty()) {continue;  }// 處理每一條讀取到的消息for (MapRecord<String, Object, Object> message : messages) {String messageId = message.getId();  // 獲取消息的唯一標識符（ID）Map<Object, Object> value = message.getValue();  // 獲取消息內容（以 Map 形式存儲）log.info("接收到消息，id={}，內容={}", messageId, value);  // 打印日志，記錄消息 ID 和內容// 在這里加入業務邏輯處理// 例如處理消息并執行相應的操作// ...// 消息處理成功后，需要確認消息已經被消費（通過 XACK 命令）redisTemplate.opsForStream().acknowledge(STREAM_KEY, GROUP_NAME, messageId);  // 確認消費的消息}} catch (Exception e) {log.error("讀取 Redis Stream 消息異常", e);  // 異常捕獲，記錄錯誤日志}}}
}

MapRecord<String, Object, Object> 是 Spring Data Redis 用來表示 Redis Stream 中的 消息記錄 的類。它不僅包含了消息的 ID，還包含了消息的內容（即消息數據）。在 Redis 中，每條消息都存儲為一個 key-value 對。

主要功能：

封裝消息 ID 和消息內容：MapRecord 用來封裝消息的 ID 和消息的內容。
消息的內容：消息的內容通常是一個 鍵值對（Map<String, Object>），可以是任意對象的數據結構（例如，JSON、Map 或其他序列化對象）。

字段：

getId()：返回消息的 ID（RecordId 類型）。
getValue()：返回消息的內容，以 Map<Object, Object> 的形式。

使用場景：

MapRecord 是用來表示從 Stream 中讀取到的消息，它將消息的 ID 和內容（鍵值對）封裝在一起。你可以使用 MapRecord 來獲取消息的 ID 和內容并處理。
MapRecord<String, Object, Object> message = redisTemplate.opsForStream().read(Consumer.from("mygroup", "consumer1"), options, StreamOffset.create("mystream", ReadOffset.lastConsumed()));
在這個例子中，message 是一個 MapRecord 實例，它封裝了從 mystream 隊列中讀取到的消息。我們可以通過 message.getId() 獲取消息 ID，通過 message.getValue() 獲取消息內容。

在消費者中，我們使用 MapRecord<String, Object, Object> 來封裝消息，獲取 message.getId() 來獲取消息的 ID（RecordId），以及通過 message.getValue() 獲取消息的內容。隨后在處理完消息后，調用 acknowledge() 來確認消息已經被消費。

最后啟動異步支持：

@SpringBootApplication
@EnableAsync // 啟動異步支持
public class MyApplication {public static void main(String[] args) {SpringApplication.run(MyApplication.class, args);}
}

通過這種方式，Spring Data Redis 提供了高效且類型安全的接口來操作 Redis Stream，幫助我們在分布式系統中實現高效的消息隊列。