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 策略也为 存根（Claim Check）模式提供了基础，该模式也在 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（参见 Metadata Store）实现。您可以使用 RedisMetadataStore 在应用程序重启后维护 MetadataStore 的状态。您可以将此新的 MetadataStore 实现与以下适配器一起使用：

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

RedisMetadataStore 由 RedisProperties 提供支持。与之交互使用 BoundHashOperations，这反过来要求整个 Properties 存储需要一个 key。对于 MetadataStore，这个 key 扮演着区域（region）的角色，这在分布式环境中，当多个应用程序使用同一个 Redis 服务器时非常有用。默认情况下，这个 key 的值为 MetaData。

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

Redis 存储入站通道适配器是一个轮询消费者（polling consumer），它从 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 表达式，并将其执行与事务协调（同步）。支持 before-commit、after-commit 和 after-rollback 的表达式，以及发送评估结果（如果有）的通道（每种事件一个）。对于每个子元素，您可以指定 expression 和 channel 属性。如果只存在 channel 属性，接收到的消息将作为特定同步场景的一部分发送到该通道。如果只存在 expression 属性并且表达式结果为非空值，则会生成一个以结果作为载荷的消息，并发送到默认通道（NullChannel）并在日志中显示（在 DEBUG 级别）。如果您希望评估结果发送到特定通道，请添加 channel 属性。如果表达式结果为 null 或 void，则不生成消息。

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

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

如果 autoAck 设置为 false，Redis Stream 中的 Record 不会被 Redis 驱动程序自动确认，而是在要生成的消息中添加一个 IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK 消息头，其值为 SimpleAcknowledgment 实例。当基于该记录的消息的业务逻辑完成后，目标集成流负责调用其 acknowledge() 回调。即使在反序列化期间发生异常并且配置了 errorChannel 时，也需要类似的逻辑。因此，目标错误处理器必须决定是 ack 还是 nack 该失败消息。除了 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 在多个应用程序实例之间提供此功能，这样一次只有一个实例可以操作该组。

当本地线程释放锁时，另一个本地线程通常可以立即获取锁。如果使用不同注册表实例的线程释放锁，则可能需要长达 100ms 才能获取锁。

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

从版本 5.0 开始，RedisLockRegistry 实现了 ExpirableLockRegistry，它会移除距离上次获取时间超过 age 且当前未被锁定的锁。

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

从版本 5.5.13 开始，RedisLockRegistry 暴露了一个 setRedisLockType(RedisLockType) 选项，用于确定应以哪种模式进行 Redis 锁获取：

发布/订阅是首选模式 - 客户端与 Redis 服务器之间的网络通信更少，性能更高 - 当订阅在其他进程中收到解锁通知时，锁会立即获取。但是，Redis 在主/副本连接（例如在 AWS ElastiCache 环境中）中不支持发布/订阅，因此默认选择忙等待模式，以使注册表在任何环境中都能工作。

从版本 6.4 开始，RedisLockRegistry.RedisLock.unlock() 方法不再抛出 IllegalStateException，而是抛出 ConcurrentModificationException，如果锁的所有权已过期。

从版本 6.4 开始，添加了 RedisLockRegistry.setRenewalTaskScheduler() 以配置用于定期续订锁的调度器。设置后，锁成功获取后，将每隔过期时间的 1/3 自动续订，直到解锁或 Redis 键被移除。