Redis Support

要开始与Redis交互，首先需要连接到它。Spring Integration使用了由另一个Spring项目 Spring Data Redis提供的支持，该项目提供了典型的Spring构造：ConnectionFactory`和`Template。这些抽象简化了与多个Redis客户端Java API的集成。目前，Spring Data Redis支持 Jedis和 Lettuce。

如“@34”中所述，Redis 通过其“@31”、“@32”和“@33”命令为发布-订阅消息传递提供支持。与 JMS 和 AMQP 一样，Spring Integration 提供消息通道和适配器，用于通过 Redis 发送和接收消息。

如_Enterprise Integration Patterns_（EIP）书中所述， message store可让你持久化消息。在处理能够缓冲消息（聚合器、重新排序器和其它）的组件时，当可靠性是一个问题时，这一点可能很有用。在Spring Integration中，`MessageStore`策略还为 claim check模式提供了基础，EIP中也介绍了此模式。

Spring Integration 的 Redis 模块提供了 RedisMessageStore。以下示例说明如何将其与聚合器一起使用：

前面的示例是一个 Bean 配置，它期望一个 RedisConnectionFactory 作为构造函数参数。

默认情况下，RedisMessageStore 使用 Java 序列化来序列化消息。但是，如果您想要使用其他序列化技术（例如 JSON），您可以通过设置 RedisMessageStore 的 valueSerializer 属性来提供自己的序列化程序。

从 4.3.10 版本开始，该框架为 Message 实例和 MessageHeaders 实例提供了 Jackson 序列化程序和反序列化程序实现——分别为 MessageJacksonDeserializer 和 MessageHeadersJacksonSerializer。它们必须针对 ObjectMapper 配置有 SimpleModule 选项。此外，您应该在 ObjectMapper 上设置 enableDefaultTyping 以添加每个序列化的复杂对象（如果您信任来源）的类型信息。此类型信息然后在反序列化过程中使用。该框架提供了一个名为 JacksonJsonUtils.messagingAwareMapper() 的实用方法，它已经提供了所有前面提到的属性和序列化程序。此实用方法带有 trustedPackages 参数，可限制反序列化的 Java 包以避免安全漏洞。默认可信包：java.util、java.lang、org.springframework.messaging.support、org.springframework.integration.support、org.springframework.integration.message、org.springframework.integration.store。要管理 RedisMessageStore 中的 JSON 序列化，您必须以类似于以下示例的方式对其进行配置：

从 4.3.12 版本开始，RedisMessageStore 支持 prefix 选项，以允许区分同一 Redis 服务器上的存储实例。

Spring Integration 3.0引入了基于Redis的新 MetadataStore（请参见Metadata Store）实现。你可以使用`RedisMetadataStore`跨应用程序重启维护`MetadataStore`的状态。你可以将此新`MetadataStore`实现与以下适配器配合使用：

为了指示这些适配器使用新的 RedisMetadataStore，请声明一个名为 metadataStore 的 Spring Bean。Feed 入站通道适配器和 Feed 入站通道适配器都会自动获取并使用声明的 RedisMetadataStore。以下示例显示如何声明此类 Bean：

RedisMetadataStore 由 link:https://docs.spring.io/spring-data/data-redis/docs/current/api/org/springframework/data/redis/support/collections/RedisProperties.html[RedisProperties 支持。与其交互使用 link:https://docs.spring.io/spring-data/data-redis/docs/current/api/org/springframework/data/redis/core/BoundHashOperations.html[BoundHashOperations，而这反过来又要求整个 Properties 存储使用 key。对于 MetadataStore，此 key 扮演了区域角色，当多个应用程序使用同一个 Redis 服务器时，这在分布式环境中很有用。默认情况下，此 key 的值为 MetaData。

从版本 4.0 开始，此存储库实现了 ConcurrentMetadataStore，使其能够在多个应用程序实例之间可靠地共享，其中只有一个实例被允许存储或修改密钥值。

Redis 存储库入站通道适配器是一个轮询消费器，它从 Redis 集合中读取数据，并将其作为 Message 有效负载发送。以下示例显示了如何配置 Redis 存储库入站通道适配器：

上一个示例通过使用 `store-inbound-channel-adapter`元素，显示了如何配置 Redis 存储库入站通道适配器，为各个属性提供数值，例如：

由于它具有 key 的文字值，上一个示例相对简单且静态。有时，你可能需要根据某些条件在运行时更改密钥值。要实现此目的，请改用 key-expression，提供的表达式可以是任何有效的 SpEL 表达式。

此外，你可能需要对从 Redis 集合中读取成功处理的数据执行一些后期处理。例如，你可能需要在处理后移动或移除值。你可以使用 Spring Integration 2.2 添加的事务同步功能来实现此目的。以下示例使用 key-expression 和事务同步：

你可以通过使用 transactional 元素声明你的轮询器作为事务性的。此元素可以引用一个真正的交易管理器（例如，如果你的流程的某个其他部分调用 JDBC）。如果没有一个“真正的”事务，你能够使用一个 o.s.i.transaction.PseudoTransactionManager，这是 Spring 的 PlatformTransactionManager 的实现，且当没有实际事务时，能够使用 Redis 适配器的事务同步功能。

一旦你的轮询器是事务性的，你能够在 transactional 元素上设置 o.s.i.transaction.TransactionSynchronizationFactory 的实例。TransactionSynchronizationFactory 创建 TransactionSynchronization 的实例。为了你的方便，我们展示了一个基于 SpEL 的默认 TransactionSynchronizationFactory，它允许你配置 SpEL 表达式，其执行与事务协调（同步）。支持事前提交、事后提交和事后回滚的表达式，以及发送评估结果（如果存在）的通道（每个事件类型一个）。对于每个子元素，你能指定 expression 和 channel 属性。如果仅存在 channel 属性，则接收到的消息将作为特定同步方案的一部分发送到那里。如果仅存在 expression 属性且表达式的结果为非空值，则生成一条消息，其中结果作为有效负载，并将其发送到默认通道 (NullChannel) 且显示在日志中（以 DEBUG 级别）。如果你想让评估结果转到特定的通道，请添加 channel 属性。如果表达式的结果为空或无效，则不生成消息。

RedisStoreMessageSource 添加一个 store 属性，其中有一个 RedisStore 实例绑定至事务 IntegrationResourceHolder，它能够从 TransactionSynchronizationProcessor 实现中进行访问。

有关事务同步的更多信息，请参阅 Transaction Synchronization。

Redis 存储库出站通道适配器允许你将消息有效负载写入 Redis 集合，如以下示例所示：

上述配置了一个 Redis 存储库出站通道适配器，使用 store-inbound-channel-adapter 元素。它为各个属性提供数值，例如：

由于它具有 key 和其他属性的文字值，上一个示例相对简单且静态。有时，你可能需要在运行时根据某些条件动态更改值。要实现此目的，请使用它们的 -expression 等效项（key-expression、map-key-expression 等），提供的表达式可以是任何有效的 SpEL 表达式。

Spring Integration 4.0 引入了 Redis 命令网关，允许你通过使用通用 RedisConnection#execute 方法执行任何标准 Redis 命令。下表显示了 Redis 出站网关的可用属性：

你能使用 <int-redis:outbound-gateway> 作为通用组件来执行任何所需的 Redis 操作。以下示例显示了如何从 Redis 原子计数器获取递增值：

Message 有效负载应具有 redisCounter 的名称，此名称可能由 org.springframework.data.redis.support.atomic.RedisAtomicInteger bean 定义提供。

RedisConnection#execute`方法的返回类型具有通用的`Object。实际结果取决于命令类型。例如，MGET`返回`List<byte[]>。有关命令、其参数和结果类型，更多信息请参见 Redis Specification。

Spring Integration 引入了 Redis 队列出站网关来执行请求和答复场景。它将对话 UUID 推送到提供的 queue，将带有该 UUID 作为其密钥的值推送到 Redis 列表中，并等待来自具有 UUID 加 .reply 密钥的 Redis 列表的答复。每个交互使用不同的 UUID。下表显示了 Redis 出站网关的可用属性：

Spring Integration 4.1 引入了 Redis 队列入站网关来执行请求和答复场景。它从提供的 queue 中弹出对话 UUID，从 Redis 列表中弹出具有该 UUID 作为其密钥的值，并将答复推送到具有 UUID 加 .reply 密钥的 Redis 列表中。下表显示了 Redis 队列入站网关的可用属性：

Spring Integration 5.4 引入了反应式 Redis Stream 输出通道适配器，将消息负载写入 Redis 流中。输出通道适配器使用 ReactiveStreamOperations.add(…​) 将一个 Record 添加到流中。以下示例展示了如何使用 Java 配置和服务类来使用 Redis Stream 输出通道适配器。

Spring Integration 5.4 引入了反应式流输入通道适配器，用于从 Redis Stream 中读取消息。输入通道适配器根据自动确认标志使用 StreamReceiver.receive(…​) 或 StreamReceiver.receiveAutoAck() 从 Redis 流中读取记录。以下示例展示了如何对 Redis Stream 输入通道适配器使用 Java 配置。

如果 autoAck 设置为 false，Redis 流中的 Record 不会由 Redis 驱动程序自动确认，而是将一个 IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK 头添加到要使用一个 SimpleAcknowledgment 实例作为值的消息中。由目标集成流负责在其基于此记录的消息上完成业务逻辑时调用其 acknowledge() 回调。如果在反序列化的过程中发生异常并且配置了 errorChannel，则需要类似的逻辑。因此，目标错误处理程序必须决定确认或取消确认此类失败的消息。与 IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK 一起，ReactiveRedisStreamMessageProducer 还会将这些头填充到要生成的消息中：RedisHeaders.STREAM_KEY、RedisHeaders.STREAM_MESSAGE_ID、RedisHeaders.CONSUMER_GROUP 和 RedisHeaders.CONSUMER。

从 5.5 版开始，您可以在 ReactiveRedisStreamMessageProducer 上明确地配置 StreamReceiver.StreamReceiverOptionsBuilder 选项，包括新引入的 onErrorResume 函数，如果在反序列化错误发生时 Redis Stream 消费者应继续轮询，则需要此函数。默认函数会向错误通道（如果已提供）发送带有可能的对失败消息进行确认的消息，如上所述。所有这些 StreamReceiver.StreamReceiverOptionsBuilder 与外部提供的 StreamReceiver.StreamReceiverOptions 是互斥的。

Spring Integration 4.0 引入了 RedisLockRegistry。某些组件（例如，聚合器和重新排序器）使用从 LockRegistry 实例获得的锁来确保一次只有一个线程操作一个组。DefaultLockRegistry 在单个组件内执行此功能。您现在可以在这些组件上配置外部锁注册表。将它与一个共享的 MessageGroupStore 结合使用时，您可以使用 RedisLockRegistry 为多个应用程序实例提供这一功能，从而一次仅有一个实例可以操作该组。

当锁被本地线程释放时，另一个本地线程通常可以立即获取该锁。如果锁是由使用不同注册表实例的线程释放的，则获取该锁可能需要最多 100 毫秒。

为了避免 “挂起” 锁（当服务器发生故障时），此注册表中的锁会在默认 60 秒后过期，但您可以在注册表上配置此值。锁通常会保持更短的时间。

从 5.0 版开始，RedisLockRegistry 实现 ExpirableLockRegistry，它会移除最后获取时间早于 age 且当前未锁定的锁。

从 5.5.6 版开始，RedisLockRegistry 在 RedisLockRegistry.locks 中支持自动清理 redisLock 的缓存，通过 RedisLockRegistry.setCacheCapacity()。有关更多信息，请参见其 JavaDocs。

从 5.5.13 版开始，RedisLockRegistry 公开了一个 setRedisLockType(RedisLockType) 选项，用于确定在何种模式下应发生 Redis 锁获取：

pub-sub 是首选模式 - 客户端 Redis 服务器之间的网络流量更少，并且性能更好 - 当订阅收到另一个进程中的解锁通知后，锁将立即获取。然而，Redis 在主/从连接中不支持 pub-sub（例如在 AWS ElastiCache 环境中），因此，繁忙循环模式被选择为默认模式，以便使注册表在任何环境中都能工作。