Redis 支持

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

如 引言 中所述，Redis 通过其 PUBLISH、SUBSCRIBE 和 UNSUBSCRIBE 命令提供对发布/订阅消息传递的支持。与 JMS 和 AMQP 一样，Spring Integration 提供了消息通道和适配器，用于通过 Redis 发送和接收消息。

如企业集成模式（EIP）书中所述，消息存储允许您持久化消息。当处理具有缓冲消息功能的组件（聚合器、重新排序器等）时，这在可靠性至关重要的情况下非常有用。在 Spring Integration 中，MessageStore策略也为凭证检查模式提供了基础，该模式也在 EIP 中进行了描述。

Spring Integration 的 Redis 模块提供了RedisMessageStore。以下示例展示了如何在聚合器中使用它

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

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

从 4.3.10 版本开始，框架为 Message 实例和 MessageHeaders 实例提供了 Jackson 序列化器和反序列化器实现——分别为 MessageJacksonDeserializer 和 MessageHeadersJacksonSerializer。它们必须使用 SimpleModule 选项为 ObjectMapper 配置。此外，您应该在 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（参见 元数据存储）实现。您可以使用 RedisMetadataStore 来维护 MetadataStore 的状态，使其在应用程序重启后仍然存在。您可以将此新的 MetadataStore 实现与以下适配器一起使用：

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

RedisMetadataStore 由 RedisProperties 支持。与它的交互使用 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属性，该属性绑定到事务IntegrationResourceHolder的RedisStore实例，可以从TransactionSynchronizationProcessor实现中访问。

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

RedisStore 出站通道适配器允许您将消息有效负载写入 Redis 集合，如下面的示例所示

前面的配置使用store-inbound-channel-adapter元素创建一个 Redis 存储出站通道适配器。它为各种属性提供值，例如

由于它对 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 规范。

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 引入了 Reactive Redis 流出站通道适配器，用于将 Message 负载写入 Redis 流。出站通道适配器使用 ReactiveStreamOperations.add(…​) 将 Record 添加到流中。以下示例展示了如何使用 Java 配置和服务类来使用 Redis 流出站通道适配器。

Spring Integration 5.4 引入了 Reactive 流入站通道适配器，用于从 Redis 流中读取消息。入站通道适配器使用 StreamReceiver.receive(…​) 或 StreamReceiver.receiveAutoAck()（基于自动确认标志）从 Redis 流中读取记录。以下示例展示了如何使用 Java 配置来使用 Redis 流入站通道适配器。

如果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 流消费者在发生反序列化错误时应继续轮询，则需要此函数。默认函数将消息发送到错误通道（如果提供），并可能对失败的消息进行确认，如上所述。所有这些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.setCacheCapacity()` 自动清理 `RedisLockRegistry.locks` 中 redisLocks 的缓存。有关更多信息，请参阅其 JavaDocs。

从 5.5.13 版本开始，`RedisLockRegistry` 公开了一个 `setRedisLockType(RedisLockType)` 选项，用于确定以哪种模式获取 Redis 锁。

发布-订阅模式是首选模式 - 减少了客户端 Redis 服务器之间的网络通信，并且性能更高 - 当订阅通知到另一个进程解锁时，锁会立即获取。但是，Redis 在主/从连接中不支持发布-订阅模式（例如在 AWS ElastiCache 环境中），因此选择繁忙循环模式作为默认模式，以使注册表在任何环境中都能正常工作。