Redis 發布訂閱模式實戰：Jedis 客戶端完整集成指南

核心概念

Redis Pub/Sub 是輕量級消息通信模式：

• Publisher：向 channel 發送消息
• Subscriber：訂閱 channel 接收消息
• 類比：廣播電台（頻道）與收音機（訂閱者）

Redis 原生命令速查

命令行演示

# 客戶端1 訂閱
127.0.0.1:6379> SUBSCRIBE study
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "study"
3) (integer) 1

# 客戶端2 發布
127.0.0.1:6379> PUBLISH study "message1-go go go"
(integer) 1  # 返回訂閱者數量

# 客戶端1 收到
1) "message"
2) "study"
3) "message1-go go go"

# 模式訂閱驗證
127.0.0.1:6379> PSUBSCRIBE study*
127.0.0.1:6379> PUBLISH study:java "I hate java forever"
# 同樣收到

Jedis 整合 Spring 完整實現

依賴配置

<dependency>
  <groupId>redis.clients</groupId>
  <artifactId>jedis</artifactId>
  <version>3.7.0</version>
</dependency>
<!-- 集群模式需 jedis-cluster -->

發布者組件

@Component
public class PublishMessage {
    private static final Logger log = LoggerFactory.getLogger(PublishMessage.class);
    @Resource private JedisCluster jedisCluster;

    public void sendMessage(final String channel, final String message) {
        Thread thread = new Thread(() -> {
            Long count = jedisCluster.publish(channel, message);
            log.info("發布 [{}] -> {} (訂閱者: {})", channel, message, count);
        });
        thread.setName("redis-publish-" + channel);
        thread.start();
        log.info("發布線程已啟動");
    }
}

訂閱監聽器

@Component
public class ChatSubscribe extends JedisPubSub {
    private static final Logger log = LoggerFactory.getLogger(ChatSubscribe.class);

    @Override public void onMessage(String channel, String message) {
        log.info("收到消息 [{}]: {}", channel, message);
        RedisString.message = message; // 業務處理
    }
    @Override public void onSubscribe(String channel, int subscribedChannels) {
        log.info("訂閱成功 [{}], 當前訂閱數: {}", channel, subscribedChannels);
    }
    @Override public void onUnsubscribe(String channel, int subscribedChannels) {
        log.info("取消訂閱 [{}]", channel);
    }
    // 模式訂閱回調略...
}

訂閱管理服務

@Component
public class SubScribeMessage {
    private final ExecutorService pool = Executors.newCachedThreadPool();
    @Resource private JedisCluster jedisCluster;
    @Resource private ChatSubscribe listener;

    public void subscribeChannel(final String channel) {
        pool.execute(() -> {
            try { jedisCluster.subscribe(listener, channel); }
            catch (Exception e) { log.error("訂閱異常", e); }
        });
    }

    public void unsubscribeChannel(final String channel) {
        // 需在監聽器線程中調用 listener.unsubscribe(channel)
    }
}

生產環境注意事項

1. 非持久化：斷線期間消息丟失，關鍵業務請用 Redis Stream 或 Kafka
2. 連接隔離：訂閱連接長佔用，建議獨立連接池，避免阻塞常規命令
3. 異常重連：JedisPubSub 異常會斷開，需外層循環重試邏輯
4. 集群模式限制：Pub/Sub 廣播到所有節點，jedisCluster.subscribe 實際隨機選一節點，功能正常但需注意

REST 控制器演示

@RestController
@RequestMapping("/redis/pubsub")
public class PubSubController {
    @Resource private PublishMessage publisher;
    @Resource private SubScribeMessage subscriber;

    @GetMapping("/publish")
    public String publish(@RequestParam String channel, @RequestParam String msg) {
        publisher.sendMessage(channel, msg);
        return "發布已提交";
    }

    @GetMapping("/subscribe")
    public String subscribe(@RequestParam String channel) {
        subscriber.subscribeChannel(channel);
        return "訂閱已啟動";
    }
}