基于SpringBoot+Redis實現(xiàn)一個完整的發(fā)布訂閱系統(tǒng)
更新時間：2026年02月28日 08:35:41 作者：J_liaty
在分布式系統(tǒng)中,消息隊列和發(fā)布訂閱模式是實現(xiàn)服務間解耦、異步通信的重要手段,本文將詳細介紹如何在Spring Boot項目中集成Redis,實現(xiàn)一個簡單但功能完整的發(fā)布訂閱系統(tǒng),需要的朋友可以參考下
前言
在分布式系統(tǒng)中，消息隊列和發(fā)布訂閱模式是實現(xiàn)服務間解耦、異步通信的重要手段。Redis不僅是一個高性能的鍵值存儲系統(tǒng)，還提供了發(fā)布訂閱（Publish/Subscribe）功能，可以輕松實現(xiàn)消息的發(fā)布與訂閱。
本文將詳細介紹如何在Spring Boot項目中集成Redis，實現(xiàn)一個簡單但功能完整的發(fā)布訂閱系統(tǒng)。
一、環(huán)境準備
1.1 Redis發(fā)布訂閱簡介
Redis的發(fā)布訂閱是一種消息通信模式：
- 發(fā)布者（Publisher）：向指定頻道發(fā)送消息
- 訂閱者（Subscriber）：訂閱感興趣的頻道并接收消息
- 頻道（Channel）：消息的邏輯分類，用于連接發(fā)布者和訂閱者
特點：
- 解耦：發(fā)布者和訂閱者無需知道彼此的存在
- 一對多：一個消息可以同時發(fā)送給多個訂閱者
- 實時性：消息實時傳遞
- 無持久化：消息不持久化，訂閱者離線會丟失消息
二、項目搭建
2.1 創(chuàng)建Spring Boot項目
使用Spring Initializr創(chuàng)建項目，或者手動創(chuàng)建Maven項目并添加以下依賴：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.10</version>
<relativePath/>
</parent>
<groupId>com.example</groupId>
<artifactId>redis-pubsub-demo</artifactId>
<version>1.0.0</version>
<name>Redis Pub/Sub Demo</name>
<description>Spring Boot + Redis 發(fā)布訂閱示例</description>
<properties>
<java.version>1.8</java.version>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<!-- Spring Boot Web -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- Spring Boot Data Redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- Lombok（可選，簡化代碼） -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<!-- Fastjson（JSON序列化） -->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.83</version>
</dependency>
<!-- Spring Boot Test -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
2.2 配置Redis連接
在src/main/resources/application.yml中配置Redis連接信息：
server:
port: 8080
spring:
application:
name: redis-pubsub-demo
# Redis配置
redis:
host: localhost
port: 6379
password: # 如果有密碼則填寫
database: 0
timeout: 3000
# Lettuce連接池配置
lettuce:
pool:
max-active: 20
max-idle: 10
min-idle: 5
max-wait: 3000ms
# 日志配置
logging:
level:
com.example: DEBUG
org.springframework.data.redis: DEBUG
三、核心實現(xiàn)
3.1 Redis配置類
創(chuàng)建Redis配置類，配置序列化器和消息監(jiān)聽容器：
package com.example.redispubsub.config;
import com.alibaba.fastjson.support.spring.GenericFastJsonRedisSerializer;
import io.lettuce.core.LettuceConnection;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.support.ConnectionPoolSupport;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import java.time.Duration;
/**
* Redis配置類 - 配置Redis連接和序列化
*
* @author example
* @date 2024-02-27
*/
@Configuration
public class RedisConfig {
/**
* 配置RedisTemplate
* 使用Fastjson進行對象序列化，key使用String序列化
*
* @param factory Redis連接工廠，Spring自動注入
* @return 配置好的RedisTemplate實例
*/
@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
// 創(chuàng)建RedisTemplate實例
RedisTemplate<String, Object> template = new RedisTemplate<>();
// 設置Redis連接工廠
template.setConnectionFactory(factory);
// ========================================
// 配置Key的序列化器（使用String序列化）
// ========================================
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
template.setKeySerializer(stringRedisSerializer); // 設置key的序列化器
template.setHashKeySerializer(stringRedisSerializer); // 設置Hash中key的序列化器
// ========================================
// 配置Value的序列化器（使用Fastjson序列化）
// ========================================
GenericFastJsonRedisSerializer fastJsonRedisSerializer = new GenericFastJsonRedisSerializer();
template.setValueSerializer(fastJsonRedisSerializer); // 設置value的序列化器
template.setHashValueSerializer(fastJsonRedisSerializer); // 設置Hash中value的序列化器
// 執(zhí)行初始化操作（必須調(diào)用）
template.afterPropertiesSet();
return template;
}
/**
* 配置Redis連接池（Lettuce）
* 優(yōu)化連接池參數(shù)以提高性能
*
* @return Lettuce客戶端配置
*/
@Bean
public LettuceConnectionFactory redisConnectionFactory() {
// 配置連接池
GenericObjectPoolConfig<LettuceConnection> poolConfig = new GenericObjectPoolConfig<>();
poolConfig.setMaxTotal(20); // 最大連接數(shù)
poolConfig.setMaxIdle(10); // 最大空閑連接數(shù)
poolConfig.setMinIdle(5); // 最小空閑連接數(shù)
poolConfig.setMaxWaitMillis(3000); // 最大等待時間
// 創(chuàng)建Lettuce客戶端配置
LettuceClientConfiguration clientConfig = LettuceClientConfiguration.builder()
.commandTimeout(Duration.ofSeconds(3)) // 命令超時時間
.shutdownTimeout(Duration.ZERO) // 關閉超時時間
.poolConfig(poolConfig) // 連接池配置
.build();
// 創(chuàng)建連接工廠并返回
return new LettuceConnectionFactory(new RedisStandaloneConfiguration("localhost", 6379), clientConfig);
}
}
3.2 消息實體類
定義消息實體，用于傳輸消息內(nèi)容：
package com.example.redispubsub.model;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
import java.time.LocalDateTime;
/**
* 消息實體類
*
* @author example
* @date 2024-02-27
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Message implements Serializable {
private static final long serialVersionUID = 1L;
/**
* 消息ID
*/
private String messageId;
/**
* 消息類型
*/
private String messageType;
/**
* 消息內(nèi)容
*/
private String content;
/**
* 發(fā)送時間
*/
private LocalDateTime sendTime;
/**
* 發(fā)送者
*/
private String sender;
/**
* 創(chuàng)建消息
*/
public static Message create(String messageType, String content, String sender) {
Message message = new Message();
message.setMessageId(System.currentTimeMillis() + "");
message.setMessageType(messageType);
message.setContent(content);
message.setSendTime(LocalDateTime.now());
message.setSender(sender);
return message;
}
}
3.3 消息發(fā)布者
創(chuàng)建消息發(fā)布者服務，用于向指定頻道發(fā)布消息：
package com.example.redispubsub.publisher;
import com.alibaba.fastjson.JSON;
import com.example.redispubsub.model.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
/**
* 消息發(fā)布者服務 - 負責向Redis頻道發(fā)布消息
*
* @author example
* @date 2024-02-27
*/
@Slf4j
@Service
public class MessagePublisher {
// ========================================
// 注入RedisTemplate，用于操作Redis
// 注意：RedisTemplate是線程安全的，可以在多線程環(huán)境中共享使用
// ========================================
@Autowired
private RedisTemplate<String, Object> redisTemplate;
/**
* 發(fā)布消息到指定頻道
*
* 工作原理：
* 1. 使用RedisTemplate的convertAndSend方法發(fā)布消息
* 2. 該方法會自動將消息對象序列化為字節(jié)數(shù)組
* 3. 然后將消息發(fā)送到Redis服務器的指定頻道
* 4. 所有訂閱該頻道的訂閱者都會收到該消息
*
* @param channel 頻道名稱（字符串）
* @param message 要發(fā)布的消息對象（會被序列化為JSON）
*/
public void publish(String channel, Message message) {
try {
// 記錄日志：準備發(fā)布消息
log.info("準備發(fā)布消息到頻道 [{}]: {}", channel, JSON.toJSONString(message));
// ========================================
// 核心發(fā)布邏輯
// ========================================
// convertAndSend是RedisTemplate提供的方法，用于發(fā)布消息
// 它會自動進行對象的序列化和反序列化
redisTemplate.convertAndSend(channel, message);
// 記錄日志：消息發(fā)布成功
log.info("消息發(fā)布成功到頻道 [{}]", channel);
} catch (Exception e) {
// 記錄錯誤日志
log.error("消息發(fā)布失敗: {}", e.getMessage(), e);
// 實際項目中，這里可以添加重試邏輯或告警通知
}
}
/**
* 發(fā)布文本消息（便捷方法）
* 自動創(chuàng)建Message對象并發(fā)布
*
* @param channel 頻道名稱
* @param content 消息內(nèi)容
*/
public void publishText(String channel, String content) {
// 創(chuàng)建消息對象：類型為TEXT，發(fā)送者為SYSTEM
Message message = Message.create("TEXT", content, "SYSTEM");
// 調(diào)用核心發(fā)布方法
publish(channel, message);
}
}
3.4 消息訂閱者
創(chuàng)建消息訂閱者，實現(xiàn)消息監(jiān)聽接口：
package com.example.redispubsub.subscriber;
import com.alibaba.fastjson.JSON;
import com.example.redispubsub.model.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;
/**
* 聊天消息訂閱者 - 實現(xiàn)MessageListener接口，監(jiān)聽chat頻道的消息
*
* 工作原理：
* 1. 實現(xiàn)Spring Data Redis的MessageListener接口
* 2. 當Redis服務器有消息發(fā)布到訂閱的頻道時，會自動調(diào)用onMessage方法
* 3. onMessage方法在獨立的線程中執(zhí)行，不會阻塞主線程
*
* @author example
* @date 2024-02-27
*/
@Slf4j
@Component
public class ChatMessageSubscriber implements MessageListener {
/**
* 消息接收回調(diào)方法（由Redis消息監(jiān)聽容器自動調(diào)用）
*
* @param message Redis消息對象，包含頻道名和消息體
* @param pattern 頻道匹配模式（如果是模式訂閱，則為匹配的模式；否則為null）
*/
@Override
public void onMessage(Message message, byte[] pattern) {
try {
// ========================================
// 步驟1：解析頻道名稱
// ========================================
String channel = new String(message.getChannel());
log.debug("收到消息，來自頻道: {}", channel);
// ========================================
// 步驟2：解析消息體
// ========================================
byte[] body = message.getBody();
// 使用Fastjson反序列化為Message對象
// 注意：需要確保發(fā)布的對象和接收的對象類型一致
Message msg = JSON.parseObject(body, Message.class);
// ========================================
// 步驟3：記錄消息詳情
// ========================================
log.info("收到消息 - 頻道: {}, 消息ID: {}, 類型: {}, 內(nèi)容: {}, 發(fā)送者: {}, 時間: {}",
channel,
msg.getMessageId(),
msg.getMessageType(),
msg.getContent(),
msg.getSender(),
msg.getSendTime());
// ========================================
// 步驟4：處理業(yè)務邏輯
// ========================================
handleMessage(msg);
} catch (Exception e) {
// 異常處理：記錄錯誤日志，避免因單個消息處理失敗影響后續(xù)消息接收
log.error("消息處理失敗: {}", e.getMessage(), e);
// 實際項目中，可以考慮將失敗的消息記錄到死信隊列或數(shù)據(jù)庫，以便后續(xù)重試
}
}
/**
* 處理消息業(yè)務邏輯
* 根據(jù)不同的消息類型執(zhí)行不同的業(yè)務處理
*
* @param message 消息對象
*/
private void handleMessage(Message message) {
// 根據(jù)消息類型進行分支處理
switch (message.getMessageType()) {
case "TEXT":
// 處理文本消息：記錄日志、保存到數(shù)據(jù)庫等
log.info("處理文本消息: {}", message.getContent());
// TODO: 添加具體的業(yè)務邏輯，例如保存聊天記錄
break;
case "ORDER":
// 處理訂單消息：更新訂單狀態(tài)、發(fā)送通知等
log.info("處理訂單消息: {}", message.getContent());
// TODO: 添加訂單相關的業(yè)務邏輯
break;
case "NOTIFICATION":
// 處理通知消息：推送到前端、發(fā)送郵件等
log.info("處理通知消息: {}", message.getContent());
// TODO: 添加通知相關的業(yè)務邏輯
break;
default:
// 未知消息類型：記錄警告日志
log.warn("未知消息類型: {}, 內(nèi)容: {}", message.getMessageType(), message.getContent());
}
}
}
3.5 消息監(jiān)聽器配置
創(chuàng)建監(jiān)聽器配置類，將訂閱者注冊到監(jiān)聽容器中：
package com.example.redispubsub.config;
import com.example.redispubsub.subscriber.ChatMessageSubscriber;
import com.example.redispubsub.subscriber.OrderMessageSubscriber;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
/**
* 消息監(jiān)聽器配置類 - 負責注冊訂閱者和頻道的綁定關系
*
* @author example
* @date 2024-02-27
*/
@Configuration
public class MessageListenerConfig {
// ========================================
// 注入訂閱者實例（由Spring容器管理）
// ========================================
@Autowired
private ChatMessageSubscriber chatMessageSubscriber; // 聊天消息訂閱者
@Autowired(required = false)
private OrderMessageSubscriber orderMessageSubscriber; // 訂單消息訂閱者（可選）
@Autowired(required = false)
private NewsMessageSubscriber newsMessageSubscriber; // 新聞消息訂閱者（可選）
/**
* 配置Redis消息監(jiān)聽容器
*
* 功能說明：
* 1. RedisMessageListenerContainer是Spring Data Redis提供的消息監(jiān)聽容器
* 2. 它會自動管理與Redis的連接，并啟動獨立的線程來接收消息
* 3. 在容器啟動時，會自動注冊所有訂閱者到對應的頻道
* 4. 當Redis服務器有消息發(fā)布到已訂閱的頻道時，容器會立即接收并調(diào)用對應的訂閱者
*
* @param connectionFactory Redis連接工廠，由Spring自動注入
* @return 配置好的消息監(jiān)聽容器
*/
@Bean
public RedisMessageListenerContainer messageListenerContainer(
RedisConnectionFactory connectionFactory) {
// ========================================
// 創(chuàng)建消息監(jiān)聽容器
// ========================================
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
// ========================================
// 注冊訂閱者到對應的頻道
// ========================================
// 注冊聊天消息訂閱者 - 監(jiān)聽"chat"頻道
// PatternTopic: 支持通配符模式的頻道訂閱，例如 "chat.*" 會匹配所有以chat.開頭的頻道
container.addMessageListener(chatMessageSubscriber, new PatternTopic("chat"));
System.out.println("[Redis Pub/Sub] 已注冊訂閱者: ChatMessageSubscriber -> 頻道: chat");
// 注冊訂單消息訂閱者 - 監(jiān)聽"order"頻道（如果存在）
if (orderMessageSubscriber != null) {
container.addMessageListener(orderMessageSubscriber, new PatternTopic("order"));
System.out.println("[Redis Pub/Sub] 已注冊訂閱者: OrderMessageSubscriber -> 頻道: order");
}
// 注冊新聞消息訂閱者 - 使用模式訂閱，監(jiān)聽所有"news.*"開頭的頻道（如果存在）
if (newsMessageSubscriber != null) {
// 模式訂閱示例：news.sports、news.tech、news.weather 都會被監(jiān)聽到
container.addMessageListener(newsMessageSubscriber, new PatternTopic("news.*"));
System.out.println("[Redis Pub/Sub] 已注冊訂閱者: NewsMessageSubscriber -> 模式: news.*");
}
// ========================================
// 可選：設置線程池（提升性能）
// ========================================
// 如果需要自定義線程池，可以取消下面注釋
/*
container.setTaskExecutor(Executors.newFixedThreadPool(10));
*/
return container;
}
}
四、創(chuàng)建測試接口
4.1 發(fā)布消息接口
創(chuàng)建Controller，提供HTTP接口用于發(fā)布消息：
package com.example.redispubsub.controller;
import com.example.redispubsub.model.Message;
import com.example.redispubsub.publisher.MessagePublisher;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import java.util.HashMap;
import java.util.Map;
/**
* 消息發(fā)布控制器
*
* @author example
* @date 2024-02-27
*/
@Slf4j
@RestController
@RequestMapping("/api/message")
public class MessageController {
@Autowired
private MessagePublisher messagePublisher;
/**
* 發(fā)布文本消息到chat頻道
*
* @param content 消息內(nèi)容
* @return 發(fā)布結果
*/
@PostMapping("/publish")
public Map<String, Object> publishMessage(@RequestParam String content) {
Map<String, Object> result = new HashMap<>();
try {
messagePublisher.publishText("chat", content);
result.put("code", 200);
result.put("message", "消息發(fā)布成功");
result.put("data", content);
} catch (Exception e) {
log.error("消息發(fā)布失敗", e);
result.put("code", 500);
result.put("message", "消息發(fā)布失敗: " + e.getMessage());
}
return result;
}
/**
* 發(fā)布自定義消息
*
* @param message 消息對象
* @return 發(fā)布結果
*/
@PostMapping("/publish/custom")
public Map<String, Object> publishCustomMessage(@RequestBody Message message) {
Map<String, Object> result = new HashMap<>();
try {
messagePublisher.publish("chat", message);
result.put("code", 200);
result.put("message", "消息發(fā)布成功");
result.put("data", message);
} catch (Exception e) {
log.error("消息發(fā)布失敗", e);
result.put("code", 500);
result.put("message", "消息發(fā)布失敗: " + e.getMessage());
}
return result;
}
/**
* 向指定頻道發(fā)布消息
*
* @param channel 頻道名稱
* @param content 消息內(nèi)容
* @return 發(fā)布結果
*/
@PostMapping("/publish/{channel}")
public Map<String, Object> publishToChannel(
@PathVariable String channel,
@RequestParam String content) {
Map<String, Object> result = new HashMap<>();
try {
messagePublisher.publishText(channel, content);
result.put("code", 200);
result.put("message", "消息發(fā)布到頻道 [" + channel + "] 成功");
result.put("data", content);
} catch (Exception e) {
log.error("消息發(fā)布失敗", e);
result.put("code", 500);
result.put("message", "消息發(fā)布失敗: " + e.getMessage());
}
return result;
}
}
五、啟動類
創(chuàng)建Spring Boot啟動類：
package com.example.redispubsub;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* Redis Pub/Sub Demo 啟動類
*
* @author example
* @date 2024-02-27
*/
@SpringBootApplication
public class RedisPubSubApplication {
public static void main(String[] args) {
SpringApplication.run(RedisPubSubApplication.class, args);
System.out.println("========================================");
System.out.println("Redis Pub/Sub Demo 啟動成功！");
System.out.println("訪問地址: http://localhost:8080");
System.out.println("========================================");
}
}
六、測試驗證
6.1 啟動Redis服務
確保Redis服務已啟動，可以使用以下命令檢查：
# 檢查Redis服務狀態(tài) redis-cli ping # 如果返回 PONG，說明Redis服務正常
6.2 啟動應用
運行RedisPubSubApplication的main方法，啟動Spring Boot應用。
6.3 測試發(fā)布消息
方式1：使用curl命令
# 發(fā)布簡單文本消息 curl -X POST "http://localhost:8080/api/message/publish?content=Hello%20Redis%20Pub/Sub" # 發(fā)布到指定頻道 curl -X POST "http://localhost:8080/api/message/publish/order?content=訂單創(chuàng)建成功"
方式2：使用Postman
請求1：發(fā)布簡單文本消息
- URL:
POST http://localhost:8080/api/message/publish - 參數(shù):
content=測試消息
請求2：發(fā)布自定義消息
- URL:
POST http://localhost:8080/api/message/publish/custom - Body (JSON):
{
"messageId": "1001",
"messageType": "ORDER",
"content": "訂單號: ORDER-20240227-001",
"sender": "ORDER_SERVICE",
"sendTime": "2024-02-27T10:30:00"
}
6.4 查看日志輸出
應用啟動后，當有消息發(fā)布到訂閱的頻道時，控制臺會輸出類似以下日志：
收到消息 - 頻道: chat, 消息ID: 1709025600000, 類型: TEXT, 內(nèi)容: 測試消息, 發(fā)送者: SYSTEM, 時間: 2024-02-27T10:30:00 處理文本消息: 測試消息
七、進階功能
7.1 多訂閱者示例
創(chuàng)建多個訂閱者，訂閱不同的頻道：
package com.example.redispubsub.subscriber;
import com.alibaba.fastjson.JSON;
import com.example.redispubsub.model.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;
/**
* 訂單消息訂閱者 - 監(jiān)聽order頻道的訂單相關消息
*
* 實際應用場景：
* - 當訂單創(chuàng)建、支付、發(fā)貨時，發(fā)布消息到order頻道
* - 本訂閱者接收到訂單消息后，進行相應的業(yè)務處理
*
* @author example
* @date 2024-02-27
*/
@Slf4j
@Component
public class OrderMessageSubscriber implements MessageListener {
/**
* 訂單消息接收回調(diào)方法
*
* @param message Redis消息對象
* @param pattern 頻道匹配模式
*/
@Override
public void onMessage(Message message, byte[] pattern) {
try {
// 獲取頻道名稱
String channel = new String(message.getChannel());
// 反序列化消息對象
Message msg = JSON.parseObject(message.getBody(), Message.class);
log.info("[訂單訂閱者] 收到訂單消息 - 頻道: {}, 消息ID: {}, 內(nèi)容: {}",
channel, msg.getMessageId(), msg.getContent());
// ========================================
// 訂單業(yè)務處理邏輯
// ========================================
handleOrderMessage(msg);
} catch (Exception e) {
log.error("[訂單訂閱者] 消息處理失敗: {}", e.getMessage(), e);
}
}
/**
* 處理訂單業(yè)務邏輯
*
* @param message 訂單消息
*/
private void handleOrderMessage(Message message) {
String content = message.getContent();
// 根據(jù)訂單內(nèi)容判斷訂單類型（實際項目中可以定義更復雜的消息結構）
if (content.contains("創(chuàng)建")) {
log.info("[訂單訂閱者] 處理訂單創(chuàng)建: {}", content);
// TODO: 1. 訂單數(shù)據(jù)入庫
// TODO: 2. 發(fā)送訂單創(chuàng)建通知
// TODO: 3. 觸發(fā)庫存扣減邏輯
} else if (content.contains("支付")) {
log.info("[訂單訂閱者] 處理訂單支付: {}", content);
// TODO: 1. 更新訂單狀態(tài)為已支付
// TODO: 2. 觸發(fā)發(fā)貨流程
// TODO: 3. 發(fā)送支付成功通知
} else if (content.contains("發(fā)貨")) {
log.info("[訂單訂閱者] 處理訂單發(fā)貨: {}", content);
// TODO: 1. 更新訂單狀態(tài)為已發(fā)貨
// TODO: 2. 發(fā)送物流信息給客戶
}
}
}
在配置類中注冊：
@Configuration
public class MessageListenerConfig {
@Autowired
private ChatMessageSubscriber chatMessageSubscriber;
@Autowired
private OrderMessageSubscriber orderMessageSubscriber;
@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(
RedisConnectionFactory connectionFactory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
// 訂閱chat頻道
container.addMessageListener(chatMessageSubscriber, new PatternTopic("chat"));
// 訂閱order頻道
container.addMessageListener(orderMessageSubscriber, new PatternTopic("order"));
return container;
}
}
7.2 模式訂閱
使用模式訂閱，訂閱匹配的多個頻道：
/**
* 新聞消息訂閱者 - 使用模式訂閱
*/
@Slf4j
@Component
public class NewsMessageSubscriber implements MessageListener {
@Override
public void onMessage(Message message, byte[] pattern) {
String channel = new String(message.getChannel());
Message msg = JSON.parseObject(message.getBody(), Message.class);
log.info("[新聞訂閱者] 頻道: {}, 新聞: {}", channel, msg.getContent());
}
}
配置模式訂閱：
@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(
RedisConnectionFactory connectionFactory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
// 模式訂閱：訂閱所有news.*開頭的頻道
container.addMessageListener(newsMessageSubscriber, new PatternTopic("news.*"));
return container;
}
八、注意事項
8.1 Redis發(fā)布訂閱的局限性
消息不持久化
- Redis發(fā)布訂閱是實時消息傳遞
- 如果訂閱者離線，期間發(fā)布的消息會永久丟失
- 不適用于需要消息可靠性的場景
無消息確認機制
- 發(fā)布者無法知道消息是否被訂閱者接收
- 沒有重試機制
擴展性限制
- 在高并發(fā)場景下可能成為性能瓶頸
- Redis集群模式下，Pub/Sub可能存在限制
8.2 生產(chǎn)環(huán)境建議
如果需要消息可靠性和持久化，建議使用以下方案：
Redis Streams（Redis 5.0+）
- 支持消息持久化
- 支持消費者組
- 支持消息確認和重試
專業(yè)消息隊列
- RabbitMQ
- Apache Kafka
- Apache RocketMQ
8.3 性能優(yōu)化
合理設置連接池
- 根據(jù)并發(fā)量調(diào)整
max-active、max-idle等參數(shù)
消息序列化優(yōu)化
- 選擇高效的序列化方式（如Fastjson、Protobuf）
批量處理
- 對于大量消息，考慮批量發(fā)布和接收
九、完整項目結構
redis-pubsub-demo/ ├── src/ │ ├── main/ │ │ ├── java/ │ │ │ └── com/example/redispubsub/ │ │ │ ├── config/ │ │ │ │ ├── MessageListenerConfig.java │ │ │ │ └── RedisConfig.java │ │ │ ├── controller/ │ │ │ │ └── MessageController.java │ │ │ ├── model/ │ │ │ │ └── Message.java │ │ │ ├── publisher/ │ │ │ │ └── MessagePublisher.java │ │ │ ├── subscriber/ │ │ │ │ ├── ChatMessageSubscriber.java │ │ │ │ ├── OrderMessageSubscriber.java │ │ │ │ └── NewsMessageSubscriber.java │ │ │ └── RedisPubSubApplication.java │ │ └── resources/ │ │ └── application.yml │ └── test/ │ └── java/ └── pom.xml
十、總結
本文詳細介紹了在Spring Boot中集成Redis實現(xiàn)發(fā)布訂閱功能的完整流程，包括：
- 項目搭建：Maven依賴配置、Redis連接配置
- 核心實現(xiàn)：Redis配置類、消息實體、發(fā)布者、訂閱者
- 測試驗證：啟動應用、發(fā)布消息、查看日志
- 進階功能：多訂閱者、模式訂閱
- 注意事項：局限性和生產(chǎn)環(huán)境建議
適用場景：
- 實時通知系統(tǒng)
- 聊天應用
- 實時數(shù)據(jù)推送
- 微服務間事件通知
Redis發(fā)布訂閱是一個輕量級的消息傳遞方案，適合對消息可靠性要求不高、但需要實時性的場景。如果需要更強的消息可靠性，建議使用Redis Streams或專業(yè)消息隊列。
以上就是基于SpringBoot+Redis實現(xiàn)一個完整的發(fā)布訂閱系統(tǒng)的詳細內(nèi)容，更多關于SpringBoot Redis發(fā)布訂閱功能的資料請關注腳本之家其它相關文章！
相關文章
Spring Boot 自動配置的底層實現(xiàn)原理深度解析
SpringBoot自動配置的核心在于通過“約定+條件判斷”實現(xiàn)配置的自動化加載與生效,其底層實現(xiàn)可以拆解為觸發(fā)入口、配置類加載、條件過濾、Bean注冊和配置覆蓋五個核心環(huán)節(jié),本文介紹Spring Boot自動配置的底層實現(xiàn)原理,感興趣的朋友一起看看吧2025-12-12
這篇文章主要介紹了如何在Java中獲取當天零點和明天零點的時間和時間戳,并提供了示例代碼,新手小白完全可以通過文中介紹的代碼實現(xiàn),需要的朋友可以參考下2025-03-03
這篇文章主要介紹了Spring框架中的JdbcTemplate相關知識,本文通過實例代碼給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友參考下吧2025-05-05
Springboot3利用redis生成唯一訂單號的實現(xiàn)示例
本文主要介紹了Springboot3利用redis生成唯一訂單號的實現(xiàn)示例,包括UUID、雪花算法和數(shù)據(jù)庫約束,具有一定的參考價值,感興趣的可以了解一下2025-03-03
這篇文章主要介紹了synchronized鎖的釋放機制,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2025-04-04
這篇文章主要介紹了sharding-jdbc中的事務詳細解讀,sharding-jdbc在分庫分表方面提供了很大的便利性,在使用DB的時候,通常都會涉及到事務這個概念,而在分庫分表的環(huán)境上再加上事務,就會使事情變得復雜起來,需要的朋友可以參考下2023-12-12
本篇文章主要介紹了Spring學習之依賴注入的方法（三種），小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-07-07
SpringBoot加載外部Jar實現(xiàn)功能按需擴展
這篇文章主要為大家詳細介紹了SpringBoot加載外部Jar實現(xiàn)功能按需擴展的相關知識,文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學習一下2025-06-06
這篇文章主要介紹了spring-cloud Sleuth的使用方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-02-02
SpringBoot3實現(xiàn)數(shù)據(jù)備份操作的多種方法全解析
在當今數(shù)字化時代,數(shù)據(jù)已然成為企業(yè)最為寶貴的資產(chǎn)之一,Spring Boot 3 作為一款廣受歡迎的 Java 開發(fā)框架,為我們實現(xiàn)高效、可靠的數(shù)據(jù)備份操作提供了豐富的工具與便捷的方式,下面我們就來看看具體實現(xiàn)方法吧2025-12-12