Spring Boot Messaging Chapter 6 Messaging with Redis

本文主要是介绍Spring Boot Messaging Chapter 6 Messaging with Redis，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

摘要：本章向您展示了如何使用Redis集成Spring Boot作为消息中间件。Redis是一个内存数据库，它被用作数据库、缓存和消息中间件。它不仅可以存储键值对，还可以用来存储复杂的数据类型，如散列、列表、集合、排序集、位图、超日志和地理空间索引。

Spring Boot使用Spring Data模块，特别是Redis one。换句话说，为了在您的项目中使用Redis，您必须将spring-boot-starer-redis依赖于您的pom.xml文件或Gradle。然后，您将拥有连接到Redis服务器的所有必要依赖项。

一：Redis as a Message Broker(Redis作为一个消息中间件)

Redis不仅提供了存储数据结构的方法，而且还实现了发布/订阅消息传递范型。前面的章节用JMS解释了这种范例。

这里的重要部分向您展示了如何与Redis交互并启用消息中间件。Redis拥有一个通道的标识，其中一个消息将由生成者发送。它将被那些对一个或多个频道感兴趣的订阅者使用。正如您所看到的，通道关键字在Redis中使用。Redis中的通道是JMS世界中的主题。

Redis有几个命令允许你与发布/订阅功能进行交互：

• SUBSCRIBE: 告诉Redis订阅一个特定的频道或频道。例如：

127.0.0.1:6379> SUBSCRIBE spring-boot-chat

您可以一次订阅多个通道，将它们按空格分隔开。

• UNSUBSCRIBE: 取消订阅的频道。这个命令不需要参数。

• PUBLISH: 通过指定通道（作为第一个参数）和实际消息来发布一条消息。例如：

127.0.0.1:6379> PUBLISH spring-boot-chat "Hi there"

• PSUBSCRIBE: 这个命令与SUBSCRIBE相同，但接受多个通道的模式。例如：

127.0.0.1:6379> PSUBSCRIBE currency.*

这个例子将订阅任何以货币开头的频道，比如currency.us,currency.asia.jp,currency.eu.gb,等等。

•PUNSUBSCRIBE: 取消订阅使用模式匹配。例如:

127.0.0.1:6379> PUNSUBSCRIBE currency.asia.*

•PING: 如果没有提供任何参数，则返回一个PONG。通常情况下，您可以使用这个来测试连接是否仍然存在。

为了试一试，确保你已经安装了Redis，它已经启动并运行了。（您可以从https://redis.io/download下载它。）打开一个新的终端窗口，并使用redis-cli命令与Redis进行交互。如图6 - 1所示。

Figure 6-1. Redis interaction with the publish/subscribe commands

图6-1向您展示了一种与Redis交互的简单方法，并确定订阅和发布一条消息是多么容易。

本章不讨论Redis的集群，也不讨论哨兵，分片，等等。只讨论消息传递。许多客户使用Redis作为消息传递代理，并作为实时数据分析的web会话管理工具。本章的开始是使用Spring Boot和Redis来进行消息传递。

二：Publish/Subscribe Messaging with Redis(使用Redis发布/订阅消息)

正如前面提到的，Spring Boot将使用Spring Data Redis的力量，它与您所熟悉的Spring JMS非常相似。Spring Boot将配置必要的组件，例如连接、使用的数据库（默认情况下是0索引DB）、集群节点、池、哨兵、超时等等。记住简单地添加 spring-boot-starter-redis可以启用Redis自动配置。

Spring Data Redis模块有两个主要的领域，用于消息的生产或发布，以及消息的消费或订阅。对于消息发布，它使用RedisTemplate<K,V>类（它使用模板设计模式），对于消息订阅，它有一个专用的异步消息侦听器容器（一个MDP消息驱动pojo）。对于同步消息，它使用一个RedisConnection接口约定。下面的部分只讨论异步订阅。

在这一章中，我们将使用两个项目：redis-demo和rest-api-redis。redis-demo项目拥有完成和补充货币项目的所有必要代码。

2.1.Subscriber(订阅者)

作为Redis订阅者，您可以通过使用固定名称或使用模式匹配来订阅一个或多个通道（或主题）。Spring Data Redis模块提供了一种进行低级操作的方法通过RedisConnection订阅。这包括subscribe和pSubscribe方法。

低级订阅需要一种方法来处理简单侦听器的连接和线程管理。现在想象有多个侦听器。您可能认为实现此功能将是一件麻烦的事情，但是Spring Data Redis包括RedisMessageListenerContainer类，它负责所有的繁重工作，并支持消息驱动pojo（MDPs）。换句话说，你可以创建自己的类和方法来接收消息并处理它。这意味着您需要使用MessageListenerAdapter类来使用这个特性。不要太担心，这是我接下来要给你们看的。

打开redis-demo项目，并检查com.micai.spring.messaging.config.RedisConfig类，如清单6-1所示。

Listing 6-1. com.micai.spring.messaging.config.RedisConfig.java

@Configuration
@EnableConfigurationProperties(SimpleRedisProperties.class)
public class RedisConfig {// Simple Message Listener@Beanpublic RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,MessageListenerAdapter listenerAdapter,@Value("${micai.redis.topic}") String topic) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.addMessageListener(listenerAdapter, new PatternTopic(topic));return container;}@BeanMessageListenerAdapter listenerAdapter(Subscriber subscriber) {return new MessageListenerAdapter(subscriber);}
}

清单6-1向您展示了我们将要使用的配置，以便订阅一个频道。让我们更详细地了解代码：

•RedisMessageListenerContainer: 这是一个完成所有繁重工作的类，并充当一个消息侦听器容器，它将接收来自Redis通道（topic）的消息。您需要设置连接工厂和将要处理接收到的消息的消息侦听器。请记住，这个侦听器容器负责所有线程和消息分派。

•RedisConnectionFactory: 这个接口对于RedisMessageListenerContainer是必需的，它包含关于Redis连接的所有信息。因为它是这个方法的一部分，Spring会自动将它连接起来，所以您不需要手工创建它。在幕后，Spring Boot会为您处理这种配置。

•MessageListenerAdapter: 该类是一个适配器，它将收到的消息委托给一个已声明的类，该类是通过MessageListener签名进行投诉的。您可以看到，这是通过调用容器来设置的。addMessageListener方法并作为参数传递订阅者（listenerAdapter-MessageListenerAdapter）和它将订阅的主题（PatternTopic）。

•PatternTopic: 这是一个类，也是消息侦听器需要的参数之一。它通常保存主题的名称或用于订阅正确频道的名称模式。

接下来，让我们看一下订阅者类。打开com.micai.spring.messaging.redis.Subscriber类，如清单6-2所示。

Listing 6-2. com.micai.spring.messaging.redis.Subscriber.java

@Component
public class Subscriber {private static final Logger LOGGER = LoggerFactory.getLogger(Subscriber.class);// If only one method defined, it must be named: handleMessagepublic void handleMessage(String message) {// Process message here ...LOGGER.info("消费Redis的消息内容为：{}", message);}
}

清单6-2显示了只有一个必需方法（handleMessage）的订阅者类。该订阅者类是消息委托的适配器。换句话说，MessageListenerAdapter符合以下签名:

void handleMessage(String message);
void handleMessage(Map message);
void handleMessage(byte[] message);
void handleMessage(Serializable message);//You can get the channel or the pattern used
void handleMessage(Serializable message, String channel);
void handleMessage(byte[] bytes, String pattern);//You can have your own object
void handleMessage(MyOwnDomainObject obj);

有时，您的适配器类会有更多的方法进行额外的处理，或者从外部调用。在这些情况下，您可以让MessageListenerAdapter类通过在其构造函数中添加额外的参数来知道您想要使用什么方法。例如:

@Component
public class Subscriber {
public void shipping(Order order){
// Process order here ...
}
//This method is used as listener for Redis topics
public void processTicket(String message){
// Process message here ...
}
// ... more methods
}
// RedisConfig.java
@Bean
MessageListenerAdapter listenerAdapter(Subscriber subscriber) {
return new MessageListenerAdapter(subscriber,"processTicket");
}

添加您将用来处理从Redis主题到构造器的传入消息的方法的名称;在这个例子中，这是processTicket方法。

正如您所看到的，使用MessageListenerAdapter的好处是您在POJO类中没有依赖关系，使您的应用程序更具可扩展性。

2.2.Publisher(发布者)

本节将介绍向通道（主题）发布消息。要在Redis中发布消息，您有两个选择。您可以使用 low-level RedisConnection类或高级RedisTemplate类（记住，它非常类似于JmsTemplate和RabbitTemplate）。两个接口都提供发布方法连接。发布（msg，通道）。您还必须确定通道（主题）。

使用RedisTemplate的好处是，您有一种方法来定义序列化/反序列化策略。这种方法隐藏了调用原始方法的复杂性，并且是线程安全的。

让我们打开代码com.micai.spring.messaging.RedisDemoApplication类如清单6-3所示。

Listing 6-3. com.micai.spring.messaging.RedisDemoApplication.java

@SpringBootApplication
public class RedisDemoApplication {private static final Logger LOGGER = LoggerFactory.getLogger(RedisDemoApplication.class);public static void main(String [] args) {SpringApplication.run(RedisDemoApplication.class, args);}@BeanCommandLineRunner sendMessage(StringRedisTemplate template, @Value("${micai.redis.topic}")String topic){return args -> {String message = "Hello Redis with Spring Boot!";template.convertAndSend(topic, message);LOGGER.info("发布Redis的消息内容为: {}", message);};}
}

清单6-3显示了主应用程序类。正如您所知道的，一旦Spring Boot最终确定了自动配置，它将执行sendMessage方法。该方法将自动在应用程序中自动连接StringRedisTemplate和主题。在application.properties文件中micai.redis.topic=spring-boot-chat的key。然后，它将使用模板通过模板来发送一条消息。接受主题和消息作为其参数的convertAndSend方法。

您通常需要使用RedisTemplate，但是在本例中，我们使用的是StringRedisTemplate。我们这样做是因为RedisTemplate被定义为RedisTemplate<K，V>，当key是Redis的key时（通常是一个字符串），而V是Redis值类型（它将是消息）。然后，StringRedisTemplate是一个使用字符串的子类。换句话说，它就像创建一个RedisTemplate<String，String>对象。

这里的有趣之处在于，StringRedisTemplate为不同的操作定义了多个字符串序列化器，它们适用于不同的数据结构，比如Set和Hash键/值。

现在，如果您运行这个项目（请记住让redis-server启动并运行），您将看到如图6-2、6-3和6-4所示的订阅者日志。

Figure 6-2. Project logs

图6-2显示了日志。具有Around AOP建议的RedisAudit类会生成这些日志。正如您所看到的，它使用订阅者类（侦听器适配器）和接收字符串消息的handleMessage方法。

图6-3显示了带有Redis客户端的终端。监视器在执行代码之前显示，只是为了确定Redis是否接受这些消息。正如您所看到的，Redis展示了在redis-server中执行的命令，PSUBSCRIBE和PUBLISH，当然还有一个PING命令来检查客户端和服务器之间的连接。

Figure 6-3. The redis-cli monitor command

图6-4显示了另一个终端窗口，在那里我们订阅了spring-boot-chat channel/topic。运行这个项目之后，它将打印消息。这是另一种确保您的redis-server正在运行的方法，并且您可以有多个订阅者到一个通道/主题。

Figure 6-4. redis-cli subscribe

2.3.JSON Serialization(JSON序列化)

现在，回到序列化/反序列化，回想一下我们正在使用JSON格式。您需要做些什么才能使发布/订阅使用JSON和序列化/反序列化为自定义对象？

如果您遵循前面的模块（JMS和RabbitMQ）的相同想法，那么回答这个问题就更容易了，因为您可以在这里应用相同的概念。

让我们先修改RedisConfig类，如清单6-4所示。

Listing 6-4. com.micai.spring.messaging.config.RedisConfig.java

@Configuration
@EnableConfigurationProperties(SimpleRedisProperties.class)
public class RedisConfig {// Simple Message Listener@Beanpublic RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,MessageListenerAdapter listenerAdapter,@Value("${micai.redis.topic}") String topic) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.addMessageListener(listenerAdapter, new PatternTopic(topic));return container;}@BeanMessageListenerAdapter rateListenerAdapter(RateSubscriber subscriber) {MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(subscriber);messageListenerAdapter.setSerializer(new Jackson2JsonRedisSerializer<>(Rate.class));return messageListenerAdapter;}@BeanRedisTemplate<String, Rate> redisTemplate(RedisConnectionFactory connectionFactory){RedisTemplate<String,Rate> redisTemplate = new RedisTemplate<String,Rate>();redisTemplate.setConnectionFactory(connectionFactory);redisTemplate.setDefaultSerializer(new Jackson2JsonRedisSerializer<>(Rate.class));redisTemplate.afterPropertiesSet();return redisTemplate;}
}

清单6-4展示了修改后的RedisConfig类。与之前的版本相比，有什么区别呢？RedisMessageListenerContainer bean是一样的，只不过现在我们也在使用rateListenerAdapter bean。让我们来回顾一下这个清单:

• MessageListenerAdapter: 这和以前是一样的，但是这里我们设置了一个新的类适配器，在这个例子中是RateSubscriber。需要注意的是，我们通过调用setSerializer方法来设置序列化器，并且我们正在使用Rate类作为对象映射器来实例化一个Jackson2JsonRedisSerializer对象。

• RedisTemplate<String,Rate>:正如您所看到的，我们定义这个bean来返回一个RedisTemplate，其中键是一个字符串，值是利率类。还要注意的是，我们通过调用setDefaultSerializer方法来设置一个序列化器，我们使用的是与以前相同的类，Jackson2JsonRedisSerializer。

查看RateSubscriber类，如清单6-5所示。

Listing 6-5. com.micai.spring.messaging.redis.RateSubscriber.java

/*** @Auther: zhaoxinguo* @Date: 2018/8/9 19:27* @Description:*/
@Component
public class RateSubscriber {private static final Logger LOGGER = LoggerFactory.getLogger(RateSubscriber.class);// If only one method defined, it must be named: handleMessagepublic void handleMessage(Rate rate){// Process message here ...LOGGER.info("消费Redis的消息内容为Rate：{}", rate);}
}

清单6-5显示了RateSubscriber类。与前面的例子没有任何改变。handleMessage方法将接收一条利率消息。现在，让我们看一下发布者，如清单6-6所示。

Listing 6-6. com.micai.spring.messaging.RedisDemoApplication.java

@SpringBootApplication
public class RedisDemoApplication {private static final Logger LOGGER = LoggerFactory.getLogger(RedisDemoApplication.class);public static void main(String [] args) {SpringApplication.run(RedisDemoApplication.class, args);}@BeanCommandLineRunner sendMessage(RedisTemplate<String, Rate> template,@Value("${micai.redis.topic}")String topic){return args -> {Rate rate = new Rate("MX", 21.17F, new Date());template.convertAndSend(topic, rate);LOGGER.info("发布Redis的消息内容为Rate: {}", rate);};}
}

清单6-6显示了主应用程序。将这个类与前一个版本进行比较;改变了什么?我们使用RedisTemplate类并使用利率作为值。我们还通过创建一个新利率来使用Rate类作为消息。

现在您可以运行您的项目并查看日志了。参见图6-5、6-6和6-7。

图6-5显示了RateSubscriber正在处理消息的日志。请记住，在幕后，为了得到对象，序列化/反序列化正在发生。