Redis发布订阅模式详解 – 编程技术之美-IT之美

发表于2024年5月16日2024年3月29日作者 IT之美

Redis的发布/订阅（Publish/Subscribe）模式是一种消息通信机制，允许客户端之间通过频道（channels）进行异步通信。在这一模式中，不同的客户端可以扮演两种角色：发布者（publisher）和订阅者（subscriber）。具体工作流程如下：

发布者：
发布者将消息发送至指定的频道。它使用 PUBLISH 命令来完成这个操作。例如：

   PUBLISH channel message

这个命令会将 message 消息推送到名为 channel 的频道。

订阅者：
订阅者通过调用 SUBSCRIBE 或 PSUBSCRIBE 命令订阅一个或多个频道以接收消息。

使用 SUBSCRIBE channel [channel ...] 可以订阅一个或多个具体的频道。
使用 PSUBSCRIBE pattern [pattern ...] 则可以订阅符合给定模式的频道，这是一种通配符匹配。

消息分发：
当有消息被发布到已订阅的频道时，Redis服务器会将消息广播给所有订阅该频道的客户端。每个订阅者都会接收到消息，并且Redis客户端库通常会提供回调函数来处理这些接收到的消息。
阻塞性质：
Redis客户端在执行 SUBSCRIBE 或 PSUBSCRIBE 后，会进入一个特殊的订阅状态，在此状态下，客户端只能接收来自订阅频道的消息，而不能执行其他Redis命令，直到执行 UNSUBSCRIBE 或 PUNSUBSCRIBE 退出订阅状态。
模式匹配：
PSUBSCRIBE 提供了模式匹配功能，允许订阅者监听特定模式的所有频道。例如，订阅 news.* 将会接收到所有以 news. 开头的频道发布的消息。
多对多关系：
任意数量的客户端可以订阅任意数量的频道，同样，一个客户端也可以向任意数量的频道发布消息。这样就形成了一种灵活的多对多消息传递模型。
实时性：
Redis发布订阅模式适用于实时通信场景，如消息通知、事件驱动系统等，因为它能够即时地将消息从发布者传递给订阅者。

Redis发布订阅模式为应用程序提供了一种高效、轻量级且易于使用的消息队列服务，但它并不保证消息的顺序传递或者消息的持久化存储，也不支持复杂的事务处理或消息确认机制，而是专注于实时消息传递与事件通知。

在Java中使用Redis的发布/订阅功能，可以通过Jedis（Redis官方推荐的一个Java客户端）或其他第三方库如Lettuce、JRedisClient等来实现。以下是一个使用Jedis进行发布/订阅的基本示例：

发布者（Publisher）

import redis.clients.jedis.Jedis;

public class RedisPublisher {

    private final Jedis jedis;

    public RedisPublisher(String host, int port) {
        // 连接到Redis服务器
        jedis = new Jedis(host, port);
    }

    public void sendMessageToChannel(String channel, String message) {
        // 向指定频道发布消息
        jedis.publish(channel, message);
    }

    // 在程序结束时关闭连接
    public void close() {
        jedis.close();
    }

    public static void main(String[] args) {
        RedisPublisher publisher = new RedisPublisher("localhost", 6379);
        // 发送一个消息到名为"mychannel"的频道
        publisher.sendMessageToChannel("mychannel", "Hello from the publisher!");
        publisher.close();
    }
}

订阅者（Subscriber）

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

public class RedisSubscriber extends JedisPubSub {

    private final Jedis jedis;

    public RedisSubscriber(String host, int port) {
        jedis = new Jedis(host, port);
    }

    public void subscribeToChannels(String... channels) {
        // 开始订阅多个频道
        jedis.subscribe(this, channels);
    }

    @Override
    public void onMessage(String channel, String message) {
        // 当从订阅的频道接收到消息时触发此方法
        System.out.println("Received message in channel '" + channel + "': " + message);
    }

    // 其他回调方法...
    @Override
    public void onSubscribe(String channel, int subscribedChannels) {
        System.out.println("Subscribed to channel '" + channel + "' with " + subscribedChannels + " total subscriptions.");
    }

    // ...其他类似onUnsubscribe、onPSubscribe和onPUnsubscribe的方法

    // 在程序结束时关闭连接
    public void close() {
        jedis.close();
    }

    public static void main(String[] args) {
        RedisSubscriber subscriber = new RedisSubscriber("localhost", 6379);

        // 订阅名为"mychannel"的频道
        subscriber.subscribeToChannels("mychannel");

        // 由于`jedis.subscribe()`是阻塞操作，因此此处不会继续执行主逻辑
        // 若要退出订阅，需要在另一个线程或信号处理机制下调用subscriber.unsubscribe()

        // 注意：在实际应用中，通常会将订阅操作放在后台线程中运行以避免阻塞主线程
    }
}

以上代码展示了如何通过Java使用Jedis与Redis进行发布/订阅通信的基础结构。订阅者通过继承JedisPubSub类并重写其回调方法来接收来自频道的消息和其他事件。发布者则直接使用jedis.publish()方法发送消息。

注意，在实际生产环境中，为了保持长时间监听，并确保在接收到消息时能够正确处理业务逻辑，通常会将订阅过程放入单独的守护线程中执行。同时，还需要考虑异常处理和资源释放等问题。