不想自己搭建一個mq怎么辦？Redis的Stream 來幫你，Redis Stream 是 Redis 5.0 引入的一種新的數據結構，用于處理實時的、可持久化的、基于時間序列的數據流。它非常適合處理事件流、日志、消息隊列等場景。下面是一個使用 Redis Stream 的具體應用場景：簡單的消息隊列系統。

應用場景：實時消息隊列

假設你正在構建一個實時消息通知系統，多個服務需要向某個隊列寫入消息，多個消費者服務需要從這個隊列中讀取消息執行相應操作。這個消息隊列需要有高性能和高可用性，并且能夠應對突發流量。

以下是如何使用 Redis Stream 實現完成訂單后通知會員服務加積分這個應用場景的步驟：

步驟 1: 添加必要的依賴

在你的 pom.xml 文件中添加 Lettuce 和 Spring Data Redis 依賴：

<dependencies><!-- Spring Boot Starter Web --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- Spring Data Redis --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><!-- Lettuce dependency for Redis interaction --><dependency><groupId>io.lettuce.core</groupId><artifactId>lettuce-core</artifactId><version>6.1.5</version></dependency>
</dependencies>

步驟 2: 配置 Redis 連接

在你的 application.properties 或 application.yml 文件中配置 Redis 連接：

spring:redis:host: localhostport: 6379

步驟 3: 創建訂單服務 (生產者)

訂單服務在訂單完成后將訂單信息寫入 Redis Stream。可以使用 Lettuce 庫來與 Redis 進行交互。

import io.lettuce.core.RedisClient;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import org.springframework.stereotype.Service;import java.util.HashMap;
import java.util.Map;@Service
public class OrderService {private static final String STREAM_KEY = "order_stream";private RedisClient redisClient;private StatefulRedisConnection<String, String> connection;private RedisCommands<String, String> commands;public OrderService() {this.redisClient = RedisClient.create("redis://localhost:6379");this.connection = redisClient.connect();this.commands = connection.sync();}public void completeOrder(String orderId, String userId, int points) {Map<String, String> orderData = new HashMap<>();orderData.put("orderId", orderId);orderData.put("userId", userId);orderData.put("points", String.valueOf(points));String messageId = commands.xadd(STREAM_KEY, orderData);System.out.println("Order completed with messageId: " + messageId);}public void close() {connection.close();redisClient.shutdown();}
}

步驟 4: 創建會員服務 (消費者)

會員服務從 Redis Stream 中讀取消息，并處理用戶積分的增加。

import io.lettuce.core.RedisClient;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.StreamMessage;
import org.springframework.stereotype.Service;import java.util.List;
import java.util.Map;@Service
public class MemberService {private static final String STREAM_KEY = "order_stream";private static final String CONSUMER_GROUP = "member_group";private static final String CONSUMER_NAME = "member_service";private RedisClient redisClient;private StatefulRedisConnection<String, String> connection;private RedisCommands<String, String> commands;public MemberService() {this.redisClient = RedisClient.create("redis://localhost:6379");this.connection = redisClient.connect();this.commands = connection.sync();// 創建消費組try {commands.xgroupCreate(STREAM_KEY, CONSUMER_GROUP, io.lettuce.core.StreamOffset.from("0"), true);} catch (Exception e) {System.out.println("Consumer group already exists");}}public void consumeMessages() {while (true) {List<StreamMessage<String, String>> messages = commands.xreadgroup(io.lettuce.core.Consumer.from(CONSUMER_GROUP, CONSUMER_NAME),io.lettuce.core.XReadArgs.StreamOffset.lastConsumed(STREAM_KEY));for (StreamMessage<String, String> message : messages) {Map<String, String> body = message.getBody();String orderId = body.get("orderId");String userId = body.get("userId");int points = Integer.parseInt(body.get("points"));// 處理用戶積分增加邏輯System.out.println("Processing order: " + orderId + " for user: " + userId + ", adding points: " + points);// 確認處理完成commands.xack(STREAM_KEY, CONSUMER_GROUP, message.getId());}try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();break;}}}public void close() {connection.close();redisClient.shutdown();}
}

步驟 5: 調整 Spring Boot 啟動類

在 Spring Boot 啟動類中啟動訂單服務和會員服務，演示消息的生產和消費：

import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;@SpringBootApplication
public class RedisStreamDemoApplication {public static void main(String[] args) {SpringApplication.run(RedisStreamDemoApplication.class, args);}@Beanpublic CommandLineRunner demo(OrderService orderService, MemberService memberService) {return args -> {// 模擬訂單完成orderService.completeOrder("order123", "user1", 100);// 啟動會員服務，處理消息new Thread(() -> memberService.consumeMessages()).start();// 等待一段時間，確保消息處理完成Thread.sleep(5000);orderService.close();memberService.close();};}
}

6. 優點

使用 Redis Stream 實現消息隊列有以下幾個優點：

1. 高性能：Redis Stream 提供了高性能的讀寫操作，適用于高吞吐量的場景。
2. 持久化：Redis Stream 支持數據持久化，不會因為 Redis 重啟而丟失數據。
3. 消費組：支持創建消費者組，多消費者可以協同工作，提高消費效率。
4. 自動化管理：Redis 可以自動管理消息的 ID、時間戳等，簡化開發。

7. 缺點

• 內存占用：Redis 是內存數據庫，若消息量過大，可能會占用大量內存。
• 學習曲線：Stream API 的使用相對于其他簡單數據結構較為復雜，需要一定的學習成本。

總結

通過上述示例，我們展示了如何使用 Redis Stream 實現一個簡單的消息隊列系統，包括生產者發布消息、消費者讀取消息和處理以及消費組的管理。Redis Stream 的高性能、持久化和自動管理特性使其非常適合處理實時數據流、消息隊列等場景。希望這個示例能夠幫助你更好地理解如何使用 Redis Stream 應對實際開發中的問題。