基于SpringBoot+Redis實(shí)現(xiàn)一個(gè)完整的發(fā)布訂閱系統(tǒng)

更新時(shí)間：2026年02月28日 08:35:41 作者：J_liaty

在分布式系統(tǒng)中,消息隊(duì)列和發(fā)布訂閱模式是實(shí)現(xiàn)服務(wù)間解耦、異步通信的重要手段,本文將詳細(xì)介紹如何在Spring Boot項(xiàng)目中集成Redis,實(shí)現(xiàn)一個(gè)簡(jiǎn)單但功能完整的發(fā)布訂閱系統(tǒng),需要的朋友可以參考下

前言

在分布式系統(tǒng)中，消息隊(duì)列和發(fā)布訂閱模式是實(shí)現(xiàn)服務(wù)間解耦、異步通信的重要手段。Redis不僅是一個(gè)高性能的鍵值存儲(chǔ)系統(tǒng)，還提供了發(fā)布訂閱（Publish/Subscribe）功能，可以輕松實(shí)現(xiàn)消息的發(fā)布與訂閱。

本文將詳細(xì)介紹如何在Spring Boot項(xiàng)目中集成Redis，實(shí)現(xiàn)一個(gè)簡(jiǎn)單但功能完整的發(fā)布訂閱系統(tǒng)。

一、環(huán)境準(zhǔn)備

1.1 Redis發(fā)布訂閱簡(jiǎn)介

Redis的發(fā)布訂閱是一種消息通信模式：

發(fā)布者（Publisher）：向指定頻道發(fā)送消息
訂閱者（Subscriber）：訂閱感興趣的頻道并接收消息
頻道（Channel）：消息的邏輯分類，用于連接發(fā)布者和訂閱者

特點(diǎn)：

解耦：發(fā)布者和訂閱者無需知道彼此的存在
一對(duì)多：一個(gè)消息可以同時(shí)發(fā)送給多個(gè)訂閱者
實(shí)時(shí)性：消息實(shí)時(shí)傳遞
無持久化：消息不持久化，訂閱者離線會(huì)丟失消息

二、項(xiàng)目搭建

2.1 創(chuàng)建Spring Boot項(xiàng)目

使用Spring Initializr創(chuàng)建項(xiàng)目，或者手動(dòng)創(chuàng)建Maven項(xiàng)目并添加以下依賴：

<?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（可選，簡(jiǎn)化代碼） -->
        <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

三、核心實(shí)現(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進(jìn)行對(duì)象序列化，key使用String序列化
     *
     * @param factory Redis連接工廠，Spring自動(dòng)注入
     * @return 配置好的RedisTemplate實(shí)例
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        // 創(chuàng)建RedisTemplate實(shí)例
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        // 設(shè)置Redis連接工廠
        template.setConnectionFactory(factory);

        // ========================================
        // 配置Key的序列化器（使用String序列化）
        // ========================================
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        template.setKeySerializer(stringRedisSerializer);          // 設(shè)置key的序列化器
        template.setHashKeySerializer(stringRedisSerializer);      // 設(shè)置Hash中key的序列化器

        // ========================================
        // 配置Value的序列化器（使用Fastjson序列化）
        // ========================================
        GenericFastJsonRedisSerializer fastJsonRedisSerializer = new GenericFastJsonRedisSerializer();
        template.setValueSerializer(fastJsonRedisSerializer);      // 設(shè)置value的序列化器
        template.setHashValueSerializer(fastJsonRedisSerializer);  // 設(shè)置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); // 最大等待時(shí)間

        // 創(chuàng)建Lettuce客戶端配置
        LettuceClientConfiguration clientConfig = LettuceClientConfiguration.builder()
                .commandTimeout(Duration.ofSeconds(3))  // 命令超時(shí)時(shí)間
                .shutdownTimeout(Duration.ZERO)          // 關(guān)閉超時(shí)時(shí)間
                .poolConfig(poolConfig)                 // 連接池配置
                .build();

        // 創(chuàng)建連接工廠并返回
        return new LettuceConnectionFactory(new RedisStandaloneConfiguration("localhost", 6379), clientConfig);
    }
}

3.2 消息實(shí)體類

定義消息實(shí)體，用于傳輸消息內(nèi)容：

package com.example.redispubsub.model;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.io.Serializable;
import java.time.LocalDateTime;

/**
 * 消息實(shí)體類
 *
 * @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ā)送時(shí)間
     */
    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ā)布者服務(wù)，用于向指定頻道發(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ā)布者服務(wù) - 負(fù)責(zé)向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. 該方法會(huì)自動(dòng)將消息對(duì)象序列化為字節(jié)數(shù)組
     * 3. 然后將消息發(fā)送到Redis服務(wù)器的指定頻道
     * 4. 所有訂閱該頻道的訂閱者都會(huì)收到該消息
     *
     * @param channel 頻道名稱（字符串）
     * @param message 要發(fā)布的消息對(duì)象（會(huì)被序列化為JSON）
     */
    public void publish(String channel, Message message) {
        try {
            // 記錄日志：準(zhǔn)備發(fā)布消息
            log.info("準(zhǔn)備發(fā)布消息到頻道 [{}]: {}", channel, JSON.toJSONString(message));
            
            // ========================================
            // 核心發(fā)布邏輯
            // ========================================
            // convertAndSend是RedisTemplate提供的方法，用于發(fā)布消息
            // 它會(huì)自動(dòng)進(jìn)行對(duì)象的序列化和反序列化
            redisTemplate.convertAndSend(channel, message);
            
            // 記錄日志：消息發(fā)布成功
            log.info("消息發(fā)布成功到頻道 [{}]", channel);
        } catch (Exception e) {
            // 記錄錯(cuò)誤日志
            log.error("消息發(fā)布失敗: {}", e.getMessage(), e);
            // 實(shí)際項(xiàng)目中，這里可以添加重試邏輯或告警通知
        }
    }

    /**
     * 發(fā)布文本消息（便捷方法）
     * 自動(dòng)創(chuàng)建Message對(duì)象并發(fā)布
     *
     * @param channel 頻道名稱
     * @param content 消息內(nèi)容
     */
    public void publishText(String channel, String content) {
        // 創(chuàng)建消息對(duì)象：類型為TEXT，發(fā)送者為SYSTEM
        Message message = Message.create("TEXT", content, "SYSTEM");
        // 調(diào)用核心發(fā)布方法
        publish(channel, message);
    }
}

3.4 消息訂閱者

創(chuàng)建消息訂閱者，實(shí)現(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;

/**
 * 聊天消息訂閱者 - 實(shí)現(xiàn)MessageListener接口，監(jiān)聽chat頻道的消息
 *
 * 工作原理：
 * 1. 實(shí)現(xiàn)Spring Data Redis的MessageListener接口
 * 2. 當(dāng)Redis服務(wù)器有消息發(fā)布到訂閱的頻道時(shí)，會(huì)自動(dòng)調(diào)用onMessage方法
 * 3. onMessage方法在獨(dú)立的線程中執(zhí)行，不會(huì)阻塞主線程
 *
 * @author example
 * @date 2024-02-27
 */
@Slf4j
@Component
public class ChatMessageSubscriber implements MessageListener {

    /**
     * 消息接收回調(diào)方法（由Redis消息監(jiān)聽容器自動(dòng)調(diào)用）
     *
     * @param message Redis消息對(duì)象，包含頻道名和消息體
     * @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對(duì)象
            // 注意：需要確保發(fā)布的對(duì)象和接收的對(duì)象類型一致
            Message msg = JSON.parseObject(body, Message.class);

            // ========================================
            // 步驟3：記錄消息詳情
            // ========================================
            log.info("收到消息 - 頻道: {}, 消息ID: {}, 類型: {}, 內(nèi)容: {}, 發(fā)送者: {}, 時(shí)間: {}",
                    channel,
                    msg.getMessageId(),
                    msg.getMessageType(),
                    msg.getContent(),
                    msg.getSender(),
                    msg.getSendTime());

            // ========================================
            // 步驟4：處理業(yè)務(wù)邏輯
            // ========================================
            handleMessage(msg);

        } catch (Exception e) {
            // 異常處理：記錄錯(cuò)誤日志，避免因單個(gè)消息處理失敗影響后續(xù)消息接收
            log.error("消息處理失敗: {}", e.getMessage(), e);
            // 實(shí)際項(xiàng)目中，可以考慮將失敗的消息記錄到死信隊(duì)列或數(shù)據(jù)庫，以便后續(xù)重試
        }
    }

    /**
     * 處理消息業(yè)務(wù)邏輯
     * 根據(jù)不同的消息類型執(zhí)行不同的業(yè)務(wù)處理
     *
     * @param message 消息對(duì)象
     */
    private void handleMessage(Message message) {
        // 根據(jù)消息類型進(jìn)行分支處理
        switch (message.getMessageType()) {
            case "TEXT":
                // 處理文本消息：記錄日志、保存到數(shù)據(jù)庫等
                log.info("處理文本消息: {}", message.getContent());
                // TODO: 添加具體的業(yè)務(wù)邏輯，例如保存聊天記錄
                break;

            case "ORDER":
                // 處理訂單消息：更新訂單狀態(tài)、發(fā)送通知等
                log.info("處理訂單消息: {}", message.getContent());
                // TODO: 添加訂單相關(guān)的業(yè)務(wù)邏輯
                break;

            case "NOTIFICATION":
                // 處理通知消息：推送到前端、發(fā)送郵件等
                log.info("處理通知消息: {}", message.getContent());
                // TODO: 添加通知相關(guān)的業(yè)務(wù)邏輯
                break;

            default:
                // 未知消息類型：記錄警告日志
                log.warn("未知消息類型: {}, 內(nèi)容: {}", message.getMessageType(), message.getContent());
        }
    }
}

3.5 消息監(jiān)聽器配置

創(chuàng)建監(jiān)聽器配置類，將訂閱者注冊(cè)到監(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)聽器配置類 - 負(fù)責(zé)注冊(cè)訂閱者和頻道的綁定關(guān)系
 *
 * @author example
 * @date 2024-02-27
 */
@Configuration
public class MessageListenerConfig {

    // ========================================
    // 注入訂閱者實(shí)例（由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. 它會(huì)自動(dòng)管理與Redis的連接，并啟動(dòng)獨(dú)立的線程來接收消息
     * 3. 在容器啟動(dòng)時(shí)，會(huì)自動(dòng)注冊(cè)所有訂閱者到對(duì)應(yīng)的頻道
     * 4. 當(dāng)Redis服務(wù)器有消息發(fā)布到已訂閱的頻道時(shí)，容器會(huì)立即接收并調(diào)用對(duì)應(yīng)的訂閱者
     *
     * @param connectionFactory Redis連接工廠，由Spring自動(dòng)注入
     * @return 配置好的消息監(jiān)聽容器
     */
    @Bean
    public RedisMessageListenerContainer messageListenerContainer(
            RedisConnectionFactory connectionFactory) {

        // ========================================
        // 創(chuàng)建消息監(jiān)聽容器
        // ========================================
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);

        // ========================================
        // 注冊(cè)訂閱者到對(duì)應(yīng)的頻道
        // ========================================
        
        // 注冊(cè)聊天消息訂閱者 - 監(jiān)聽"chat"頻道
        // PatternTopic: 支持通配符模式的頻道訂閱，例如 "chat.*" 會(huì)匹配所有以chat.開頭的頻道
        container.addMessageListener(chatMessageSubscriber, new PatternTopic("chat"));
        System.out.println("[Redis Pub/Sub] 已注冊(cè)訂閱者: ChatMessageSubscriber -> 頻道: chat");

        // 注冊(cè)訂單消息訂閱者 - 監(jiān)聽"order"頻道（如果存在）
        if (orderMessageSubscriber != null) {
            container.addMessageListener(orderMessageSubscriber, new PatternTopic("order"));
            System.out.println("[Redis Pub/Sub] 已注冊(cè)訂閱者: OrderMessageSubscriber -> 頻道: order");
        }

        // 注冊(cè)新聞消息訂閱者 - 使用模式訂閱，監(jiān)聽所有"news.*"開頭的頻道（如果存在）
        if (newsMessageSubscriber != null) {
            // 模式訂閱示例：news.sports、news.tech、news.weather 都會(huì)被監(jiān)聽到
            container.addMessageListener(newsMessageSubscriber, new PatternTopic("news.*"));
            System.out.println("[Redis Pub/Sub] 已注冊(cè)訂閱者: NewsMessageSubscriber -> 模式: news.*");
        }

        // ========================================
        // 可選：設(shè)置線程池（提升性能）
        // ========================================
        // 如果需要自定義線程池，可以取消下面注釋
        /*
        container.setTaskExecutor(Executors.newFixedThreadPool(10));
        */

        return container;
    }
}

四、創(chuàng)建測(cè)試接口

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ā)布結(jié)果
     */
    @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 消息對(duì)象
     * @return 發(fā)布結(jié)果
     */
    @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ā)布結(jié)果
     */
    @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;
    }
}

五、啟動(dòng)類

創(chuàng)建Spring Boot啟動(dòng)類：

package com.example.redispubsub;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * Redis Pub/Sub Demo 啟動(dòng)類
 *
 * @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 啟動(dòng)成功！");
        System.out.println("訪問地址: http://localhost:8080");
        System.out.println("========================================");
    }
}

六、測(cè)試驗(yàn)證

6.1 啟動(dòng)Redis服務(wù)

確保Redis服務(wù)已啟動(dòng)，可以使用以下命令檢查：

# 檢查Redis服務(wù)狀態(tài)
redis-cli ping

# 如果返回 PONG，說明Redis服務(wù)正常

6.2 啟動(dòng)應(yīng)用

運(yùn)行RedisPubSubApplication的main方法，啟動(dòng)Spring Boot應(yīng)用。

6.3 測(cè)試發(fā)布消息

方式1：使用curl命令

# 發(fā)布簡(jiǎn)單文本消息
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

請(qǐng)求1：發(fā)布簡(jiǎn)單文本消息

URL: POST http://localhost:8080/api/message/publish
參數(shù): content=測(cè)試消息

請(qǐng)求2：發(fā)布自定義消息

URL: POST http://localhost:8080/api/message/publish/custom
Body (JSON):

{
  "messageId": "1001",
  "messageType": "ORDER",
  "content": "訂單號(hào): ORDER-20240227-001",
  "sender": "ORDER_SERVICE",
  "sendTime": "2024-02-27T10:30:00"
}

6.4 查看日志輸出

應(yīng)用啟動(dòng)后，當(dāng)有消息發(fā)布到訂閱的頻道時(shí)，控制臺(tái)會(huì)輸出類似以下日志：

收到消息 - 頻道: chat, 消息ID: 1709025600000, 類型: TEXT, 內(nèi)容: 測(cè)試消息, 發(fā)送者: SYSTEM, 時(shí)間: 2024-02-27T10:30:00
處理文本消息: 測(cè)試消息

七、進(jìn)階功能

7.1 多訂閱者示例

創(chuàng)建多個(gè)訂閱者，訂閱不同的頻道：

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頻道的訂單相關(guān)消息
 * 
 * 實(shí)際應(yīng)用場(chǎng)景：
 * - 當(dāng)訂單創(chuàng)建、支付、發(fā)貨時(shí)，發(fā)布消息到order頻道
 * - 本訂閱者接收到訂單消息后，進(jìn)行相應(yīng)的業(yè)務(wù)處理
 * 
 * @author example
 * @date 2024-02-27
 */
@Slf4j
@Component
public class OrderMessageSubscriber implements MessageListener {

    /**
     * 訂單消息接收回調(diào)方法
     * 
     * @param message Redis消息對(duì)象
     * @param pattern 頻道匹配模式
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        try {
            // 獲取頻道名稱
            String channel = new String(message.getChannel());
            
            // 反序列化消息對(duì)象
            Message msg = JSON.parseObject(message.getBody(), Message.class);

            log.info("[訂單訂閱者] 收到訂單消息 - 頻道: {}, 消息ID: {}, 內(nèi)容: {}",
                    channel, msg.getMessageId(), msg.getContent());

            // ========================================
            // 訂單業(yè)務(wù)處理邏輯
            // ========================================
            handleOrderMessage(msg);

        } catch (Exception e) {
            log.error("[訂單訂閱者] 消息處理失敗: {}", e.getMessage(), e);
        }
    }

    /**
     * 處理訂單業(yè)務(wù)邏輯
     * 
     * @param message 訂單消息
     */
    private void handleOrderMessage(Message message) {
        String content = message.getContent();
        
        // 根據(jù)訂單內(nèi)容判斷訂單類型（實(shí)際項(xiàng)目中可以定義更復(fù)雜的消息結(jié)構(gòu)）
        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ā)送物流信息給客戶
        }
    }
}

在配置類中注冊(cè)：

@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 模式訂閱

使用模式訂閱，訂閱匹配的多個(gè)頻道：

/**
 * 新聞消息訂閱者 - 使用模式訂閱
 */
@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("[新聞?dòng)嗛喺遌 頻道: {}, 新聞: {}", 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;
}

八、注意事項(xiàng)

8.1 Redis發(fā)布訂閱的局限性

消息不持久化

Redis發(fā)布訂閱是實(shí)時(shí)消息傳遞
如果訂閱者離線，期間發(fā)布的消息會(huì)永久丟失
不適用于需要消息可靠性的場(chǎng)景

無消息確認(rèn)機(jī)制

發(fā)布者無法知道消息是否被訂閱者接收
沒有重試機(jī)制

擴(kuò)展性限制

在高并發(fā)場(chǎng)景下可能成為性能瓶頸
Redis集群模式下，Pub/Sub可能存在限制

8.2 生產(chǎn)環(huán)境建議

如果需要消息可靠性和持久化，建議使用以下方案：

Redis Streams（Redis 5.0+）

支持消息持久化
支持消費(fèi)者組
支持消息確認(rèn)和重試

專業(yè)消息隊(duì)列

RabbitMQ
Apache Kafka
Apache RocketMQ

8.3 性能優(yōu)化

合理設(shè)置連接池

根據(jù)并發(fā)量調(diào)整max-active、max-idle等參數(shù)

消息序列化優(yōu)化

選擇高效的序列化方式（如Fastjson、Protobuf）

批量處理

對(duì)于大量消息，考慮批量發(fā)布和接收

九、完整項(xiàng)目結(jié)構(gòu)

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

十、總結(jié)

本文詳細(xì)介紹了在Spring Boot中集成Redis實(shí)現(xiàn)發(fā)布訂閱功能的完整流程，包括：

項(xiàng)目搭建：Maven依賴配置、Redis連接配置
核心實(shí)現(xiàn)：Redis配置類、消息實(shí)體、發(fā)布者、訂閱者
測(cè)試驗(yàn)證：?jiǎn)?dòng)應(yīng)用、發(fā)布消息、查看日志
進(jìn)階功能：多訂閱者、模式訂閱
注意事項(xiàng)：局限性和生產(chǎn)環(huán)境建議

適用場(chǎng)景：

實(shí)時(shí)通知系統(tǒng)
聊天應(yīng)用
實(shí)時(shí)數(shù)據(jù)推送
微服務(wù)間事件通知

Redis發(fā)布訂閱是一個(gè)輕量級(jí)的消息傳遞方案，適合對(duì)消息可靠性要求不高、但需要實(shí)時(shí)性的場(chǎng)景。如果需要更強(qiáng)的消息可靠性，建議使用Redis Streams或?qū)I(yè)消息隊(duì)列。

以上就是基于SpringBoot+Redis實(shí)現(xiàn)一個(gè)完整的發(fā)布訂閱系統(tǒng)的詳細(xì)內(nèi)容，更多關(guān)于SpringBoot Redis發(fā)布訂閱功能的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章:

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

基于SpringBoot+Redis實(shí)現(xiàn)一個(gè)完整的發(fā)布訂閱系統(tǒng)

目錄

前言

一、環(huán)境準(zhǔn)備

1.1 Redis發(fā)布訂閱簡(jiǎn)介

二、項(xiàng)目搭建

2.1 創(chuàng)建Spring Boot項(xiàng)目

2.2 配置Redis連接

三、核心實(shí)現(xiàn)

3.1 Redis配置類

3.2 消息實(shí)體類

3.3 消息發(fā)布者

3.4 消息訂閱者

3.5 消息監(jiān)聽器配置

四、創(chuàng)建測(cè)試接口

4.1 發(fā)布消息接口

五、啟動(dòng)類

六、測(cè)試驗(yàn)證

6.1 啟動(dòng)Redis服務(wù)

6.2 啟動(dòng)應(yīng)用

6.3 測(cè)試發(fā)布消息

方式1：使用curl命令

方式2：使用Postman

6.4 查看日志輸出

七、進(jìn)階功能

7.1 多訂閱者示例

7.2 模式訂閱

八、注意事項(xiàng)

8.1 Redis發(fā)布訂閱的局限性

8.2 生產(chǎn)環(huán)境建議

8.3 性能優(yōu)化

九、完整項(xiàng)目結(jié)構(gòu)

十、總結(jié)

相關(guān)文章

最新評(píng)論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

基于SpringBoot+Redis實(shí)現(xiàn)一個(gè)完整的發(fā)布訂閱系統(tǒng)

目錄

前言

一、環(huán)境準(zhǔn)備

1.1 Redis發(fā)布訂閱簡(jiǎn)介

二、項(xiàng)目搭建

2.1 創(chuàng)建Spring Boot項(xiàng)目

2.2 配置Redis連接

三、核心實(shí)現(xiàn)

3.1 Redis配置類

3.2 消息實(shí)體類

3.3 消息發(fā)布者

3.4 消息訂閱者

3.5 消息監(jiān)聽器配置

四、創(chuàng)建測(cè)試接口

4.1 發(fā)布消息接口

五、啟動(dòng)類

六、測(cè)試驗(yàn)證

6.1 啟動(dòng)Redis服務(wù)

6.2 啟動(dòng)應(yīng)用

6.3 測(cè)試發(fā)布消息

方式1：使用curl命令

方式2：使用Postman

6.4 查看日志輸出

七、進(jìn)階功能

7.1 多訂閱者示例

7.2 模式訂閱

八、注意事項(xiàng)

8.1 Redis發(fā)布訂閱的局限性

8.2 生產(chǎn)環(huán)境建議

8.3 性能優(yōu)化

九、完整項(xiàng)目結(jié)構(gòu)

十、總結(jié)

相關(guān)文章

最新評(píng)論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

一、環(huán)境準(zhǔn)備

二、項(xiàng)目搭建

三、核心實(shí)現(xiàn)

五、啟動(dòng)類

六、測(cè)試驗(yàn)證

八、注意事項(xiàng)

十、總結(jié)