使用 Spring Data Redis 发布订阅消息

1. 概述

在 Redis 中，发布者并没有将消息发送给特定的订阅者。是将发布的消息被划分为通道，并不知道会有哪些订阅者(如果有的话)。

类似地，订阅者表示对一个或多个主题感兴趣，并且只接收感兴趣的消息，而不知道有哪些发布者(如果有的话)。

发布者和订阅者的这种解耦可以实现更大的可伸缩性和更动态的网络拓扑。

2. Redis 配置

让我们开始添加消息队列所需的配置。

首先，我们将定义一个 MessageListenerAdapter，其中包含名为 RedisMessageSubscriber 的 MessageListener 接口的自定义实现。这个 bean 充当发布-订阅消息模型中的订阅者:

@Bean
MessageListenerAdapter messageListener() { 
    return new MessageListenerAdapter(new RedisMessageSubscriber());
}

RedisMessageListenerContainer 是 Spring Data Redis 提供的一个类。这是内部调用的，根据 Spring Data Redis 文档 的说法 —— “处理监听、转换和消息调度的底层细节。”

@Bean
RedisMessageListenerContainer redisContainer() {
    RedisMessageListenerContainer container 
      = new RedisMessageListenerContainer(); 
    container.setConnectionFactory(jedisConnectionFactory()); 
    container.addMessageListener(messageListener(), topic()); 
    return container; 
}

我们还将使用定制的 MessagePublisher 接口和 RedisMessagePublisher 实现创建 bean。这样，我们可以有一个通用的消息发布 API，并让 Redis 实现采用 redisTemplate 和 topic 作为构造函数参数：

@Bean
MessagePublisher redisPublisher() { 
    return new RedisMessagePublisher(redisTemplate(), topic());
}

最后，我们将设置一个主题，发布者将向其发送消息，订阅者将接收消息:

@Bean
ChannelTopic topic() {
    return new ChannelTopic("messageQueue");
}

3. 发布消息

3.1. 定义 MessagePublisher 接口

Spring Data Redis 没有提供用于消息分发的 MessagePublisher 接口。我们可以定义一个自定义接口，它将在实现中使用 redisTemplate：

public interface MessagePublisher {
    void publish(String message);
}

3.2. RedisMessagePublisher 实现

我们接下来提供 MessagePublisher 接口的实现，添加消息发布的细节并使用 redisTemplate 中的函数。

该模板包含了一组非常丰富的函数，用于广泛的操作—— 其中 convertAndSend 能够通过主题向队列发送消息：

public class RedisMessagePublisher implements MessagePublisher {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private ChannelTopic topic;

    public RedisMessagePublisher() {
    }

    public RedisMessagePublisher(
      RedisTemplate<String, Object> redisTemplate, ChannelTopic topic) {
      this.redisTemplate = redisTemplate;
      this.topic = topic;
    }

    public void publish(String message) {
        redisTemplate.convertAndSend(topic.getTopic(), message);
    }
}

如您所见，发布者实现非常简单。它使用 redisTemplate 的 convertAndSend() 方法格式化给定的消息并将其发布到配置的主题。

主题实现了发布和订阅语义：当消息发布时，它将发送给所有注册侦听该主题的订阅者。

4. 订阅消息

RedisMessageSubscriber 实现了 Spring Data Redis 提供的 MessageListener 接口：

@Service
public class RedisMessageSubscriber implements MessageListener {

    public static List<String> messageList = new ArrayList<String>();

    public void onMessage(Message message, byte[] pattern) {
        messageList.add(message.toString());
        System.out.println("Message received: " + message.toString());
    }
}

注意，还有第二个参数 pattern，在本例中我们没有使用它。Spring Data Redis 文档指出，该参数表示“匹配通道的模式(如果指定)”，但它可以为 null。

5. 发送与接收消息

现在我们把它们结合起来。我们创建一个消息，然后使用 RedisMessagePublisher 发布它：

String message = "Message " + UUID.randomUUID();
redisMessagePublisher.publish(message);

当我们调用 publish(message) 时，内容被发送到 Redis，在那里它被路由到我们的发布者中定义的消息队列主题。然后将它分发给该主题的订阅者。

您可能已经注意到 RedisMessageSubscriber 是一个侦听器，它将自己注册到队列以检索消息。

消息到达时，订阅者定义的 onMessage() 方法被触发。

在我们的例子中，我们可以通过检查 RedisMessageSubscriber 中的 messageList 来验证我们已经收到了已经发布的消息：

RedisMessageSubscriber.messageList.get(0).contains(message)

6. 结论

在本文中，我们研究了使用Spring Data Redis 实现的发布/订阅消息队列。

上述示例的实现可以在 GitHub project 项目中找到。

【注】本文译自：PubSub Messaging with Spring Data Redis | Baeldung