Redis Support

Spring Integration 2.1 引入了对 Redis的支持：“an open source advanced key-value store”。此支持以 Redis 为基础的 MessageStore`以及 Redis 通过其 `PUBLISH, SUBSCRIBE, and UNSUBSCRIBE命令支持的发布/订阅消息传递适配器的形式提供。

Spring Integration 2.1 introduced support for Redis: “an open source advanced key-value store”. This support comes in the form of a Redis-based MessageStore as well as publish-subscribe messaging adapters that are supported by Redis through its PUBLISH, SUBSCRIBE, and UNSUBSCRIBE commands.

你需要将此依赖项包含在你的项目中：

You need to include this dependency into your project:

  • Maven

  • Gradle

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-redis</artifactId>
    <version>{project-version}</version>
</dependency>
compile "org.springframework.integration:spring-integration-redis:{project-version}"

你还需要包括 Redis 客户端依赖项，例如 Lettuce。

You also need to include Redis client dependency, e.g. Lettuce.

如需下载、安装和运行 Redis，请参见 Redis documentation。

To download, install, and run Redis, see the Redis documentation.

Connecting to Redis

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

To begin interacting with Redis, you first need to connect to it. Spring Integration uses support provided by another Spring project, Spring Data Redis, which provides typical Spring constructs: ConnectionFactory and Template. Those abstractions simplify integration with several Redis client Java APIs. Currently, Spring Data Redis supports Jedis and Lettuce.

Using RedisConnectionFactory

若要连接到 Redis，您可以使用 RedisConnectionFactory 接口的一种实现。以下清单显示了接口定义：

To connect to Redis, you can use one of the implementations of the RedisConnectionFactory interface. The following listing shows the interface definition:

public interface RedisConnectionFactory extends PersistenceExceptionTranslator {

    /**
     * Provides a suitable connection for interacting with Redis.
     * @return connection for interacting with Redis.
     */
    RedisConnection getConnection();
}

以下示例演示如何在 Java 中创建 LettuceConnectionFactory：

The following example shows how to create a LettuceConnectionFactory in Java:

LettuceConnectionFactory cf = new LettuceConnectionFactory();
cf.afterPropertiesSet();

以下示例演示如何在 Spring 的 XML 配置中创建 LettuceConnectionFactory：

The following example shows how to create a LettuceConnectionFactory in Spring’s XML configuration:

<bean id="redisConnectionFactory"
    class="o.s.data.redis.connection.lettuce.LettuceConnectionFactory">
    <property name="port" value="7379" />
</bean>

RedisConnectionFactory 的实现提供一组属性，例如端口和主机，可以在需要时设置这些属性。一旦有了 RedisConnectionFactory 实例，就可以创建一个 RedisTemplate 实例并用 RedisConnectionFactory 注入它。

The implementations of RedisConnectionFactory provide a set of properties, such as port and host, that you can set if needed. Once you have an instance of RedisConnectionFactory, you can create an instance of RedisTemplate and inject it with the RedisConnectionFactory.

Using RedisTemplate

与Spring中的其他模板类（例如`JdbcTemplate`和`JmsTemplate`）一样，RedisTemplate`是一个助手类，简化了Redis数据访问代码。更多有关`RedisTemplate`及其变体（例如`StringRedisTemplate）的信息，请参见 Spring Data Redis documentation。

As with other template classes in Spring (such as JdbcTemplate and JmsTemplate) RedisTemplate is a helper class that simplifies Redis data access code. For more information about RedisTemplate and its variations (such as StringRedisTemplate) see the Spring Data Redis documentation.

以下示例演示如何在 Java 中创建 RedisTemplate 实例：

The following example shows how to create an instance of RedisTemplate in Java:

RedisTemplate rt = new RedisTemplate<String, Object>();
rt.setConnectionFactory(redisConnectionFactory);

以下示例演示如何在 Spring 的 XML 配置中创建 RedisTemplate 实例：

The following example shows how to create an instance of RedisTemplate in Spring’s XML configuration:

<bean id="redisTemplate"
         class="org.springframework.data.redis.core.RedisTemplate">
    <property name="connectionFactory" ref="redisConnectionFactory"/>
</bean>

Messaging with Redis

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

As mentioned in the introduction, Redis provides support for publish-subscribe messaging through its PUBLISH, SUBSCRIBE, and UNSUBSCRIBE commands. As with JMS and AMQP, Spring Integration provides message channels and adapters for sending and receiving messages through Redis.

Redis Publish/Subscribe channel

与 JMS 类似，在某些情况下，生产者和消费者都应成为同一应用程序的一部分，在同一进程中运行。您可以通过使用一对入站和出站通道适配器来实现此目的。但是，与 Spring Integration 的 JMS 支持一样，有一种更简单的方法来解决此用例。您可以创建一个发布-订阅通道，如下例所示：

Similarly to JMS, there are cases where both the producer and consumer are intended to be part of the same application, running within the same process. You can accomplish this by using a pair of inbound and outbound channel adapters. However, as with Spring Integration’s JMS support, there is a simpler way to address this use case. You can create a publish-subscribe channel, as the following example shows:

<int-redis:publish-subscribe-channel id="redisChannel" topic-name="si.test.topic"/>

publish-subscribe-channel 的行为与 main Spring Integration 命名空间中的普通 <publish-subscribe-channel/> 元素非常相似。任何端点的 input-channel 和 output-channel 属性都可以引用它。不同之处在于此通道由 Redis 主题名称支持： String 值由 topic-name 属性指定。但是，与 JMS 不同，不必预先创建此主题或由 Redis 自动创建它。在 Redis 中，主题是作为地址发挥作用的简单 String 值。生产者和消费者可以使用相同的 String 值作为其主题名称进行通信。简单订阅此通道意味着在生产端点和消费端点之间可以进行异步发布-订阅消息传递。但是，与通过在简单 Spring Integration <channel/> 元素内添加 <queue/> 元素创建的异步消息通道不同，这些消息不会存储在内存队列中。相反，这些消息通过 Redis 传递，这使您可以依赖它对持久性和集群的支持以及它与其他非 Java 平台的互操作性。

A publish-subscribe-channel behaves much like a normal <publish-subscribe-channel/> element from the main Spring Integration namespace. It can be referenced by both the input-channel and the output-channel attributes of any endpoint. The difference is that this channel is backed by a Redis topic name: a String value specified by the topic-name attribute. However, unlike JMS, this topic does not have to be created in advance or even auto-created by Redis. In Redis, topics are simple String values that play the role of an address. The producer and consumer can communicate by using the same String value as their topic name. A simple subscription to this channel means that asynchronous publish-subscribe messaging is possible between the producing and consuming endpoints. However, unlike the asynchronous message channels created by adding a <queue/> element within a simple Spring Integration <channel/> element, the messages are not stored in an in-memory queue. Instead, those messages are passed through Redis, which lets you rely on its support for persistence and clustering as well as its interoperability with other non-Java platforms.

Redis Inbound Channel Adapter

Redis 入站通道适配器 (RedisInboundChannelAdapter) 以与其他入站适配器相同的方式将传入的 Redis 消息调整为 Spring 消息。它接收特定于平台的消息（在这种情况下为 Redis），并使用 MessageConverter 策略将它们转换为 Spring 消息。以下示例演示如何配置 Redis 入站通道适配器：

The Redis inbound channel adapter (RedisInboundChannelAdapter) adapts incoming Redis messages into Spring messages in the same way as other inbound adapters. It receives platform-specific messages (Redis in this case) and converts them to Spring messages by using a MessageConverter strategy. The following example shows how to configure a Redis inbound channel adapter:

<int-redis:inbound-channel-adapter id="redisAdapter"
       topics="thing1, thing2"
       channel="receiveChannel"
       error-channel="testErrorChannel"
       message-converter="testConverter" />

<bean id="redisConnectionFactory"
    class="o.s.data.redis.connection.lettuce.LettuceConnectionFactory">
    <property name="port" value="7379" />
</bean>

<bean id="testConverter" class="things.something.SampleMessageConverter" />

前面的示例显示了 Redis 入站通道适配器的简单但完整的配置。请注意，前面的配置依赖于自动发现某些 bean 的熟悉的 Spring 范例。在这种情况下，redisConnectionFactory 会隐式注入到适配器中。您可以改为使用 connection-factory 属性显式指定它。

The preceding example shows a simple but complete configuration of a Redis inbound channel adapter. Note that the preceding configuration relies on the familiar Spring paradigm of auto-discovering certain beans. In this case, the redisConnectionFactory is implicitly injected into the adapter. You can specify it explicitly by using the connection-factory attribute instead.

此外，还要注意，前面的配置会用自定义 MessageConverter 注入适配器。该方法类似于 JMS，在其中 MessageConverter 实例用于在 Redis 消息和 Spring Integration 消息有效负载之间进行转换。默认值为 SimpleMessageConverter。

Also, note that the preceding configuration injects the adapter with a custom MessageConverter. The approach is similar to JMS, where MessageConverter instances are used to convert between Redis messages and the Spring Integration message payloads. The default is a SimpleMessageConverter.

入站适配器可以订阅多个主题名称，因此 topics 属性中会出现逗号分隔的值集。

Inbound adapters can subscribe to multiple topic names, hence the comma-separated set of values in the topics attribute.

自3.0版本开始，入站适配器现在除了现有的`topics`属性之外，还具有`topic-patterns`属性。此属性包含一个以逗号分隔的Redis主题模式集。更多有关Redis发布-订阅的信息，请参见 Redis Pub/Sub。

Since version 3.0, the inbound adapter, in addition to the existing topics attribute, now has the topic-patterns attribute. This attribute contains a comma-separated set of Redis topic patterns. For more information regarding Redis publish-subscribe, see Redis Pub/Sub.

入站适配器可以使用 RedisSerializer`反序列化 Redis 消息的主体。<int-redis:inbound-channel-adapter>`的 `serializer`属性可以设置为一个空字符串，这会生成 `RedisSerializer`属性的 `null`值。在这种情况下，Redis 消息的原始 `byte[]`主体将作为消息有效负载提供。

Inbound adapters can use a RedisSerializer to deserialize the body of Redis messages. The serializer attribute of the <int-redis:inbound-channel-adapter> can be set to an empty string, which results in a null value for the RedisSerializer property. In this case, the raw byte[] bodies of Redis messages are provided as the message payloads.

从 5.0 版开始，您可以通过使用 <int-redis:inbound-channel-adapter> 的 task-executor 属性向入站适配器提供 Executor 实例。此外，接收的 Spring Integration 消息现在具有 RedisHeaders.MESSAGE_SOURCE 头，以指示已发布消息的来源：主题或模式。您可以将其用于下游路由逻辑。

Since version 5.0, you can provide an Executor instance to the inbound adapter by using the task-executor attribute of the <int-redis:inbound-channel-adapter>. Also, the received Spring Integration messages now have the RedisHeaders.MESSAGE_SOURCE header to indicate the source of the published message: topic or pattern. You can use this downstream for routing logic.

Redis Outbound Channel Adapter

Redis 出站通道适配器以与其他出站适配器相同的方式将传出的 Spring Integration 消息调整为 Redis 消息。它接收 Spring Integration 消息并使用 MessageConverter 策略将它们转换为特定于平台的消息（在这种情况下为 Redis）。以下示例演示如何配置 Redis 出站通道适配器：

The Redis outbound channel adapter adapts outgoing Spring Integration messages into Redis messages in the same way as other outbound adapters. It receives Spring Integration messages and converts them to platform-specific messages (Redis in this case) by using a MessageConverter strategy. The following example shows how to configure a Redis outbound channel adapter:

<int-redis:outbound-channel-adapter id="outboundAdapter"
    channel="sendChannel"
    topic="thing1"
    message-converter="testConverter"/>

<bean id="redisConnectionFactory"
    class="o.s.data.redis.connection.lettuce.LettuceConnectionFactory">
    <property name="port" value="7379"/>
</bean>

<bean id="testConverter" class="things.something.SampleMessageConverter" />

此配置与 Redis 入站通道适配器类似。适配器隐式注入 RedisConnectionFactory，其通过 redisConnectionFactory 作为 Bean 名称进行定义。此示例还包括可选的（和自定义的）MessageConverter（testConverter Bean）。

The configuration parallels the Redis inbound channel adapter. The adapter is implicitly injected with a RedisConnectionFactory, which is defined with redisConnectionFactory as its bean name. This example also includes the optional (and custom) MessageConverter (the testConverter bean).

自 Spring Integration 3.0 起，<int-redis:outbound-channel-adapter> 针对 topic 属性提供了一种替代方法：您可以在运行时使用 topic-expression 属性来确定消息的 Redis 主题。这些属性是互斥的。

Since Spring Integration 3.0, the <int-redis:outbound-channel-adapter> offers an alternative to the topic attribute: You can use the topic-expression attribute to determine the Redis topic for the message at runtime. These attributes are mutually exclusive.

Redis Queue Inbound Channel Adapter

Spring Integration 3.0 引入了一个队列入站通道适配器来从 Redis 列表中“弹出”消息。默认情况下，它使用“右侧弹出”，但您可以将其配置为使用“左侧弹出”。适配器是消息驱动的。它使用一个内部侦听器线程，而不使用轮询器。

Spring Integration 3.0 introduced a queue inbound channel adapter to “pop” messages from a Redis list. By default, it uses “right pop”, but you can configure it to use “left pop” instead. The adapter is message-driven. It uses an internal listener thread and does not use a poller.

以下清单显示了 queue-inbound-channel-adapter 的所有可用属性：

The following listing shows all the available attributes for queue-inbound-channel-adapter:

<int-redis:queue-inbound-channel-adapter id=""  1
                    channel=""  2
                    auto-startup=""  3
                    phase=""  4
                    connection-factory=""  5
                    queue=""  6
                    error-channel=""  7
                    serializer=""  8
                    receive-timeout=""  9
                    recovery-interval=""  10
                    expect-message=""  11
                    task-executor=""  12
                    right-pop=""/>  13
1 The component bean name. If you do not provide the channel attribute, a DirectChannel is created and registered in the application context with this id attribute as the bean name. In this case, the endpoint itself is registered with the bean name id plus .adapter. (If the bean name were thing1, the endpoint is registered as thing1.adapter.)
2 The MessageChannel to which to send Message instances from this Endpoint.
3 A SmartLifecycle attribute to specify whether this endpoint should start automatically after the application context start or not. It defaults to true.
4 A SmartLifecycle attribute to specify the phase in which this endpoint is started. It defaults to 0.
5 A reference to a RedisConnectionFactory bean. It defaults to redisConnectionFactory.
6 The name of the Redis list on which the queue-based 'pop' operation is performed to get Redis messages.
7 The MessageChannel to which to send ErrorMessage instances when exceptions are received from the listening task of the endpoint. By default, the underlying MessagePublishingErrorHandler uses the default errorChannel from the application context.
8 The RedisSerializer bean reference. It can be an empty string, which means 'no serializer'. In this case, the raw byte[] from the inbound Redis message is sent to the channel as the Message payload. By default, it is a JdkSerializationRedisSerializer.
9 The timeout in milliseconds for 'pop' operation to wait for a Redis message from the queue. The default is 1 second.
10 The time in milliseconds for which the listener task should sleep after exceptions on the 'pop' operation, before restarting the listener task.
11 Specifies whether this endpoint expects data from the Redis queue to contain entire Message instances. If this attribute is set to true, the serializer cannot be an empty string, because messages require some form of deserialization (JDK serialization by default). Its default is false.
12 A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor) bean. It is used for the underlying listening task. It defaults to a SimpleAsyncTaskExecutor.
13 Specifies whether this endpoint should use “right pop” (when true) or “left pop” (when false) to read messages from the Redis list. If true, the Redis List acts as a FIFO queue when used with a default Redis queue outbound channel adapter. Set it to false to use with software that writes to the list with “right push” or to achieve a stack-like message order. Its default is true. Since version 4.3.

必须使用多个线程配置 task-executor 来进行处理；否则，当 RedisQueueMessageDrivenEndpoint 尝试在出现错误后重新启动侦听器任务时，可能会产生死锁。errorChannel 可用于处理这些错误，以避免重新启动，但最好不要将您的应用程序暴露于可能的死锁情况。请参阅 Spring Framework Reference Manual 以了解可能的 TaskExecutor 实现。

The task-executor has to be configured with more than one thread for processing; otherwise there is a possible deadlock when the RedisQueueMessageDrivenEndpoint tries to restart the listener task after an error. The errorChannel can be used to process those errors, to avoid restarts, but it is preferable to not expose your application to the possible deadlock situation. See Spring Framework Reference Manual for possible TaskExecutor implementations.

Redis Queue Outbound Channel Adapter

Spring Integration 3.0 引入了一个队列出站通道适配器来从 Spring Integration 消息中“推送”到 Redis 列表。默认情况下，它使用“左侧推送”，但您可以将其配置为使用“右侧推送”。以下清单显示了 Redis queue-outbound-channel-adapter 的所有可用属性：

Spring Integration 3.0 introduced a queue outbound channel adapter to “push” to a Redis list from Spring Integration messages. By default, it uses “left push”, but you can configure it to use “right push” instead. The following listing shows all the available attributes for a Redis queue-outbound-channel-adapter:

<int-redis:queue-outbound-channel-adapter id=""  1
                    channel=""  2
                    connection-factory=""  3
                    queue=""  4
                    queue-expression=""  5
                    serializer=""  6
                    extract-payload=""  7
                    left-push=""/>  8
1 The component bean name. If you do not provide the channel attribute, a DirectChannel is created and registered in the application context with this id attribute as the bean name. In this case, the endpoint is registered with a bean name of id plus .adapter. (If the bean name were thing1, the endpoint is registered as thing1.adapter.)
2 The MessageChannel from which this endpoint receives Message instances.
3 A reference to a RedisConnectionFactory bean. It defaults to redisConnectionFactory.
4 The name of the Redis list on which the queue-based 'push' operation is performed to send Redis messages. This attribute is mutually exclusive with queue-expression.
5 A SpEL Expression to determine the name of the Redis list. It uses the incoming Message at runtime as the #root variable. This attribute is mutually exclusive with queue.
6 A RedisSerializer bean reference. It defaults to a JdkSerializationRedisSerializer. However, for String payloads, a StringRedisSerializer is used, if a serializer reference is not provided.
7 Specifies whether this endpoint should send only the payload or the entire Message to the Redis queue. It defaults to true.
8 Specifies whether this endpoint should use “left push” (when true) or “right push” (when false) to write messages to the Redis list. If true, the Redis list acts as a FIFO queue when used with a default Redis queue inbound channel adapter. Set it to false to use with software that reads from the list with “left pop” or to achieve a stack-like message order. It defaults to true. Since version 4.3.

Redis Application Events

自 Spring Integration 3.0 起，Redis 模块提供了 IntegrationEvent 的一个实现，后者又是一个 org.springframework.context.ApplicationEvent。RedisExceptionEvent 封装了 Redis 操作中的异常（其中端点是事件的“来源”）。例如，<int-redis:queue-inbound-channel-adapter/> 在从 BoundListOperations.rightPop 操作捕获异常后发出这些事件。异常可能是任何泛 org.springframework.data.redis.RedisSystemException 或 org.springframework.data.redis.RedisConnectionFailureException。使用 <int-event:inbound-channel-adapter/> 处理这些事件对于确定后台 Redis 任务的问题和采取管理操作很有用。

Since Spring Integration 3.0, the Redis module provides an implementation of IntegrationEvent, which, in turn, is a org.springframework.context.ApplicationEvent. The RedisExceptionEvent encapsulates exceptions from Redis operations (with the endpoint being the “source” of the event). For example, the <int-redis:queue-inbound-channel-adapter/> emits those events after catching exceptions from the BoundListOperations.rightPop operation. The exception may be any generic org.springframework.data.redis.RedisSystemException or a org.springframework.data.redis.RedisConnectionFailureException. Handling these events with an <int-event:inbound-channel-adapter/> can be useful to determine problems with background Redis tasks and to take administrative actions.

Redis Message Store

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

As described in the Enterprise Integration Patterns (EIP) book, a message store lets you persist messages. This can be useful when dealing with components that have a capability to buffer messages (aggregator, resequencer, and others) when reliability is a concern. In Spring Integration, the MessageStore strategy also provides the foundation for the claim check pattern, which is described in EIP as well.

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

Spring Integration’s Redis module provides the RedisMessageStore. The following example shows how to use it with a aggregator:

<bean id="redisMessageStore" class="o.s.i.redis.store.RedisMessageStore">
    <constructor-arg ref="redisConnectionFactory"/>
</bean>

<int:aggregator input-channel="inputChannel" output-channel="outputChannel"
         message-store="redisMessageStore"/>

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

The preceding example is a bean configuration, and it expects a RedisConnectionFactory as a constructor argument.

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

By default, the RedisMessageStore uses Java serialization to serialize the message. However, if you want to use a different serialization technique (such as JSON), you can provide your own serializer by setting the valueSerializer property of the RedisMessageStore.

从 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 序列化，您必须以类似于以下示例的方式对其进行配置：

Starting with version 4.3.10, the Framework provides Jackson serializer and deserializer implementations for Message instances and MessageHeaders instances — MessageJacksonDeserializer and MessageHeadersJacksonSerializer, respectively. They have to be configured with the SimpleModule options for the ObjectMapper. In addition, you should set enableDefaultTyping on the ObjectMapper to add type information for each serialized complex object (if you trust the source). That type information is then used during deserialization. The framework provides a utility method called JacksonJsonUtils.messagingAwareMapper(), which is already supplied with all the previously mentioned properties and serializers. This utility method comes with the trustedPackages argument to limit Java packages for deserialization to avoid security vulnerabilities. The default trusted packages: java.util, java.lang, org.springframework.messaging.support, org.springframework.integration.support, org.springframework.integration.message, org.springframework.integration.store. To manage JSON serialization in the RedisMessageStore, you must configure it in a fashion similar to the following example:

RedisMessageStore store = new RedisMessageStore(redisConnectionFactory);
ObjectMapper mapper = JacksonJsonUtils.messagingAwareMapper();
RedisSerializer<Object> serializer = new GenericJackson2JsonRedisSerializer(mapper);
store.setValueSerializer(serializer);

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

Starting with version 4.3.12, RedisMessageStore supports the prefix option to allow distinguishing between instances of the store on the same Redis server.

Redis Channel Message Stores

RedisMessageStore`shown earlier将每个组作为单个键（组ID）下的值进行维护。虽然你可以使用它为`QueueChannel`提供持久性支持，但出于此目的已提供了专门的`RedisChannelMessageStore（自4.0版本开始）。此存储针对每个通道使用`LIST`，在发送消息时使用`LPUSH`，在接收消息时使用`RPOP`。默认情况下，此存储还使用JDK序列化，但你可以修改值序列化器，就像described earlier一样。

The RedisMessageStore shown earlier maintains each group as a value under a single key (the group ID). While you can use this to back a QueueChannel for persistence, a specialized RedisChannelMessageStore is provided for that purpose (since version 4.0). This store uses a LIST for each channel, LPUSH when sending messages, and RPOP when receiving messages. By default, this store also uses JDK serialization, but you can modify the value serializer, as described earlier.

我们建议使用此支持通道的存储库，而不是使用通用 RedisMessageStore。以下示例定义了一个 Redis 消息存储库，并将其用于带有一个队列的通道：

We recommend using this store backing channels, instead of using the general RedisMessageStore. The following example defines a Redis message store and uses it in a channel with a queue:

<bean id="redisMessageStore" class="o.s.i.redis.store.RedisChannelMessageStore">
	<constructor-arg ref="redisConnectionFactory"/>
</bean>

<int:channel id="somePersistentQueueChannel">
    <int:queue message-store="redisMessageStore"/>
<int:channel>

用于存储数据的键具有以下形式：<storeBeanName>:<channelId>（在前一个示例中，为 redisMessageStore:somePersistentQueueChannel）。

The keys used to store the data have the form: <storeBeanName>:<channelId> (in the preceding example, redisMessageStore:somePersistentQueueChannel).

此外，还提供了 RedisChannelPriorityMessageStore 子类。当您将其与 QueueChannel 一起使用时，消息以（FIFO）优先级顺序收到。它使用标准 IntegrationMessageHeaderAccessor.PRIORITY 头文件，并支持优先级值（“0 - 9”）。具有其他优先级（和没有优先级）的消息在任何具有优先级的消息之后以 FIFO 顺序检索。

In addition, a subclass RedisChannelPriorityMessageStore is also provided. When you use this with a QueueChannel, the messages are received in (FIFO) priority order. It uses the standard IntegrationMessageHeaderAccessor.PRIORITY header and supports priority values (0 - 9). Messages with other priorities (and messages with no priority) are retrieved in FIFO order after any messages with priority.

这些存储器只实现了 BasicMessageGroupStore，而没有实现 MessageGroupStore。它们只能用于作为 QueueChannel 的后备等情况。

These stores implement only BasicMessageGroupStore and do not implement MessageGroupStore. They can be used only for situations such as backing a QueueChannel.

Redis Metadata Store

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

Spring Integration 3.0 introduced a new Redis-based MetadataStore (see Metadata Store) implementation. You can use the RedisMetadataStore to maintain the state of a MetadataStore across application restarts. You can use this new MetadataStore implementation with adapters such as:

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

To instruct these adapters to use the new RedisMetadataStore, declare a Spring bean named metadataStore. The Feed inbound channel adapter and the feed inbound channel adapter both automatically pick up and use the declared RedisMetadataStore. The following example shows how to declare such a bean:

<bean name="metadataStore" class="o.s.i.redis.store.metadata.RedisMetadataStore">
    <constructor-arg name="connectionFactory" ref="redisConnectionFactory"/>
</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。

The RedisMetadataStore is backed by RedisProperties. Interaction with it uses BoundHashOperations, which, in turn, requires a key for the entire Properties store. In the case of the MetadataStore, this key plays the role of a region, which is useful in a distributed environment, when several applications use the same Redis server. By default, this key has a value of MetaData.

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

Starting with version 4.0, this store implements ConcurrentMetadataStore, letting it be reliably shared across multiple application instances where only one instance is allowed to store or modify a key’s value.

您不能在 Redis 集群中使用 RedisMetadataStore.replace()（例如在 AbstractPersistentAcceptOnceFileListFilter 中），因为目前不支持原子性的 WATCH 命令。

You cannot use the RedisMetadataStore.replace() (for example, in the AbstractPersistentAcceptOnceFileListFilter) with a Redis cluster, since the WATCH command for atomicity is not currently supported.

Redis Store Inbound Channel Adapter

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

The Redis store inbound channel adapter is a polling consumer that reads data from a Redis collection and sends it as a Message payload. The following example shows how to configure a Redis store inbound channel adapter:

<int-redis:store-inbound-channel-adapter id="listAdapter"
    connection-factory="redisConnectionFactory"
    key="myCollection"
    channel="redisChannel"
    collection-type="LIST" >
    <int:poller fixed-rate="2000" max-messages-per-poll="10"/>
</int-redis:store-inbound-channel-adapter>

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

The preceding example shows how to configure a Redis store inbound channel adapter by using the store-inbound-channel-adapter element, providing values for various attributes, such as:

  • key or key-expression: The name of the key for the collection being used.

  • collection-type: An enumeration of the collection types supported by this adapter. The supported Collections are LIST, SET, ZSET, PROPERTIES, and MAP.

  • connection-factory: Reference to an instance of o.s.data.redis.connection.RedisConnectionFactory.

  • redis-template: Reference to an instance of o.s.data.redis.core.RedisTemplate.

  • Other attributes that are common across all other inbound adapters (such as 'channel').

无法同时设置 redis-template 和 connection-factory。

You cannot set both redis-template and connection-factory.

默认情况下，适配器使用 StringRedisTemplate。这使用 StringRedisSerializer 实例作为密钥、值、哈希密钥和哈希值。如果你的 Redis 存储库包含使用其他技术序列化的对象，你必须提供配置了适当的序列化程序的 RedisTemplate。例如，如果存储库是使用具有 extract-payload-elements 设置为 false`的 Redis 存储库出站适配器进行写入的，你必须提供如下配置的 `RedisTemplate：

By default, the adapter uses a StringRedisTemplate. This uses StringRedisSerializer instances for keys, values, hash keys, and hash values. If your Redis store contains objects that are serialized with other techniques, you must supply a RedisTemplate configured with appropriate serializers. For example, if the store is written to using a Redis store outbound adapter that has its extract-payload-elements set to false, you must provide a RedisTemplate configured as follows:

<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
    <property name="connectionFactory" ref="redisConnectionFactory"/>
    <property name="keySerializer">
        <bean class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
    </property>
    <property name="hashKeySerializer">
        <bean class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
    </property>
</bean>

RedisTemplate 将 String 序列化程序用于密钥和哈希密钥，将默认 JDK 序列化程序用于值和哈希值。

The RedisTemplate uses String serializers for keys and hash keys and the default JDK Serialization serializers for values and hash values.

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

Because it has a literal value for the key, the preceding example is relatively simple and static. Sometimes, you may need to change the value of the key at runtime based on some condition. To do so, use key-expression instead, where the provided expression can be any valid SpEL expression.

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

Also, you may wish to perform some post-processing on the successfully processed data that was read from the Redis collection. For example, you may want to move or remove the value after it has been processed. You can do so by using the transaction synchronization feature that was added with Spring Integration 2.2. The following example uses key-expression and transaction synchronization:

<int-redis:store-inbound-channel-adapter id="zsetAdapterWithSingleScoreAndSynchronization"
        connection-factory="redisConnectionFactory"
        key-expression="'presidents'"
        channel="otherRedisChannel"
        auto-startup="false"
        collection-type="ZSET">
            <int:poller fixed-rate="1000" max-messages-per-poll="2">
                <int:transactional synchronization-factory="syncFactory"/>
            </int:poller>
</int-redis:store-inbound-channel-adapter>

<int:transaction-synchronization-factory id="syncFactory">
	<int:after-commit expression="payload.removeByScore(18, 18)"/>
</int:transaction-synchronization-factory>

<bean id="transactionManager" class="o.s.i.transaction.PseudoTransactionManager"/>

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

You can declare your poller to be transactional by using a transactional element. This element can reference a real transaction manager (for example, if some other part of your flow invokes JDBC). If you do not have a “real” transaction, you can use an o.s.i.transaction.PseudoTransactionManager, which is an implementation of Spring’s PlatformTransactionManager and enables the use of the transaction synchronization features of the Redis adapter when there is no actual transaction.

这不会使 Redis 活动本身具有事务性。它允许在成功（提交）之前或之后以及失败（回滚）之后执行操作的同步。

This does not make the Redis activities themselves transactional. It lets the synchronization of actions be taken before or after success (commit) or after failure (rollback).

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

Once your poller is transactional, you can set an instance of the o.s.i.transaction.TransactionSynchronizationFactory on the transactional element. TransactionSynchronizationFactory creates an instance of the TransactionSynchronization. For your convenience, we have exposed a default SpEL-based TransactionSynchronizationFactory, which lets you configure SpEL expressions, with their execution being coordinated (synchronized) with a transaction. Expressions for before-commit, after-commit, and after-rollback are supported, together with channels (one for each kind of event) where the evaluation result (if any) is sent. For each child element, you can specify expression and channel attributes. If only the channel attribute is present, the received message is sent there as part of the particular synchronization scenario. If only the expression attribute is present and the result of an expression is a non-null value, a message with the result as the payload is generated and sent to a default channel (NullChannel) and appears in the logs (at the DEBUG level). If you want the evaluation result to go to a specific channel, add a channel attribute. If the result of an expression is null or void, no message is generated.

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

The RedisStoreMessageSource adds a store attribute with a RedisStore instance bound to the transaction IntegrationResourceHolder, which can be accessed from a TransactionSynchronizationProcessor implementation.

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

For more information about transaction synchronization, see Transaction Synchronization.

RedisStore Outbound Channel Adapter

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

The RedisStore outbound channel adapter lets you write a message payload to a Redis collection, as the following example shows:

<int-redis:store-outbound-channel-adapter id="redisListAdapter"
          collection-type="LIST"
          channel="requestChannel"
          key="myCollection" />

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

The preceding configuration a Redis store outbound channel adapter by using the store-inbound-channel-adapter element. It provides values for various attributes, such as:

  • key or key-expression: The name of the key for the collection being used.

  • extract-payload-elements: If set to true (the default) and the payload is an instance of a “multi-value” object (that is, a Collection or a Map), it is stored by using “addAll” and “putAll” semantics. Otherwise, if set to false, the payload is stored as a single entry regardless of its type. If the payload is not an instance of a “multi-value” object, the value of this attribute is ignored and the payload is always stored as a single entry.

  • collection-type: An enumeration of the Collection types supported by this adapter. The supported Collections are LIST, SET, ZSET, PROPERTIES, and MAP.

  • map-key-expression: SpEL expression that returns the name of the key for the entry being stored. It applies only if the collection-type is MAP or PROPERTIES and 'extract-payload-elements' is false.

  • connection-factory: Reference to an instance of o.s.data.redis.connection.RedisConnectionFactory.

  • redis-template: Reference to an instance of o.s.data.redis.core.RedisTemplate.

  • Other attributes that are common across all other inbound adapters (such as 'channel').

无法同时设置 redis-template 和 connection-factory。

You cannot set both redis-template and connection-factory.

默认情况下，适配器使用 StringRedisTemplate。这使用 StringRedisSerializer 实例作为键、值、哈希键和哈希值。但是，如果 extract-payload-elements 设置为 false，将使用具有 StringRedisSerializer 实例作为键和哈希键以及 JdkSerializationRedisSerializer 实例作为值和哈希值 RedisTemplate。使用 JDK 序列化程序时，重要的是要知道对所有值使用 Java 序列化，无论该值实际上是否为集合。如果你需要更多地控制值序列化，请考虑提供你自己的 RedisTemplate，而不依赖于这些默认值。

By default, the adapter uses a StringRedisTemplate. This uses StringRedisSerializer instances for keys, values, hash keys, and hash values. However, if extract-payload-elements is set to false, a RedisTemplate that has StringRedisSerializer instances for keys and hash keys and JdkSerializationRedisSerializer instances s for values and hash values will be used. With the JDK serializer, it is important to understand that Java serialization is used for all values, regardless of whether the value is actually a collection or not. If you need more control over the serialization of values, consider providing your own RedisTemplate rather than relying upon these defaults.

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

Because it has literal values for the key and other attributes, the preceding example is relatively simple and static. Sometimes, you may need to change the values dynamically at runtime based on some condition. To do so, use their -expression equivalents (key-expression, map-key-expression, and so on), where the provided expression can be any valid SpEL expression.

Redis Outbound Command Gateway

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

Spring Integration 4.0 introduced the Redis command gateway to let you perform any standard Redis command by using the generic RedisConnection#execute method. The following listing shows the available attributes for the Redis outbound gateway:

<int-redis:outbound-gateway
        request-channel=""  1
        reply-channel=""  2
        requires-reply=""  3
        reply-timeout=""  4
        connection-factory=""  5
        redis-template=""  6
        arguments-serializer=""  7
        command-expression=""  8
        argument-expressions=""  9
        use-command-variable=""  10
        arguments-strategy="" /> 11
1 The MessageChannel from which this endpoint receives Message instances.
2 The MessageChannel where this endpoint sends reply Message instances.
3 Specifies whether this outbound gateway must return a non-null value. It defaults to true. A ReplyRequiredException is thrown when Redis returns a null value.
4 The timeout (in milliseconds) to wait until the reply message is sent. It is typically applied for queue-based limited reply-channels.
5 A reference to a RedisConnectionFactory bean. It defaults to redisConnectionFactory. It is mutually exclusive with 'redis-template' attribute.
6 A reference to a RedisTemplate bean. It is mutually exclusive with 'connection-factory' attribute.
7 A reference to an instance of org.springframework.data.redis.serializer.RedisSerializer. It is used to serialize each command argument to byte[], if necessary.
8 The SpEL expression that returns the command key. It defaults to the redis_command message header. It must not evaluate to null.
9 Comma-separated SpEL expressions that are evaluated as command arguments. Mutually exclusive with the arguments-strategy attribute. If you provide neither attribute, the payload is used as the command arguments. The argument expressions can evaluate to 'null' to support a variable number of arguments.
10 A boolean flag to specify whether the evaluated Redis command string is made available as the #cmd variable in the expression evaluation context in the o.s.i.redis.outbound.ExpressionArgumentsStrategy when argument-expressions is configured. Otherwise, this attribute is ignored.
11 Reference to an instance of o.s.i.redis.outbound.ArgumentsStrategy. It is mutually exclusive with argument-expressions attribute. If you provide neither attribute, the payload is used as the command arguments.

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

You can use the <int-redis:outbound-gateway> as a common component to perform any desired Redis operation. The following example shows how to get incremented values from Redis atomic number:

<int-redis:outbound-gateway request-channel="requestChannel"
    reply-channel="replyChannel"
    command-expression="'INCR'"/>

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

The Message payload should have a name of redisCounter, which may be provided by org.springframework.data.redis.support.atomic.RedisAtomicInteger bean definition.

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

The RedisConnection#execute method has a generic Object as its return type. Real result depends on command type. For example, MGET returns a List<byte[]>. For more information about commands, their arguments and result type, see Redis Specification.

Redis Queue Outbound Gateway

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

Spring Integration introduced the Redis queue outbound gateway to perform request and reply scenarios. It pushes a conversation UUID to the provided queue, pushes the value with that UUID as its key to a Redis list, and waits for the reply from a Redis list with a key of UUID plus .reply. A different UUID is used for each interaction. The following listing shows the available attributes for a Redis outbound gateway:

<int-redis:queue-outbound-gateway
        request-channel=""  1
        reply-channel=""  2
        requires-reply=""  3
        reply-timeout=""  4
        connection-factory=""  5
        queue=""  6
        order=""  7
        serializer=""  8
        extract-payload=""/>  9
1 The MessageChannel from which this endpoint receives Message instances.
2 The MessageChannel where this endpoint sends reply Message instances.
3 Specifies whether this outbound gateway must return a non-null value. This value is false by default. Otherwise, a ReplyRequiredException is thrown when Redis returns a null value.
4 The timeout (in milliseconds) to wait until the reply message is sent. It is typically applied for queue-based limited reply-channels.
5 A reference to a RedisConnectionFactory bean. It defaults to redisConnectionFactory. It is mutually exclusive with the 'redis-template' attribute.
6 The name of the Redis list to which the outbound gateway sends a conversation UUID.
7 The order of this outbound gateway when multiple gateways are registered.
8 The RedisSerializer bean reference. It can be an empty string, which means “no serializer”. In this case, the raw byte[] from the inbound Redis message is sent to the channel as the Message payload. By default, it is a JdkSerializationRedisSerializer.
9 Specifies whether this endpoint expects data from the Redis queue to contain entire Message instances. If this attribute is set to true, the serializer cannot be an empty string, because messages require some form of deserialization (JDK serialization by default).

Redis Queue Inbound Gateway

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

Spring Integration 4.1 introduced the Redis queue inbound gateway to perform request and reply scenarios. It pops a conversation UUID from the provided queue, pops the value with that UUID as its key from the Redis list, and pushes the reply to the Redis list with a key of UUID plus .reply. The following listing shows the available attributes for a Redis queue inbound gateway:

<int-redis:queue-inbound-gateway
        request-channel=""  1
        reply-channel=""  2
        executor=""  3
        reply-timeout=""  4
        connection-factory=""  5
        queue=""  6
        order=""  7
        serializer=""  8
        receive-timeout=""  9
        expect-message=""  10
        recovery-interval=""/>  11
1 The MessageChannel where this endpoint sends Message instances created from the Redis data.
2 The MessageChannel from where this endpoint waits for reply Message instances. Optional - the replyChannel header is still in use.
3 A reference to a Spring TaskExecutor (or a standard JDK Executor) bean. It is used for the underlying listening task. It defaults to a SimpleAsyncTaskExecutor.
4 The timeout (in milliseconds) to wait until the reply message is sent. It is typically applied for queue-based limited reply-channels.
5 A reference to a RedisConnectionFactory bean. It defaults to redisConnectionFactory. It is mutually exclusive with 'redis-template' attribute.
6 The name of the Redis list for the conversation UUID.
7 The order of this inbound gateway when multiple gateways are registered.
8 The RedisSerializer bean reference. It can be an empty string, which means “no serializer”. In this case, the raw byte[] from the inbound Redis message is sent to the channel as the Message payload. It defaults to a JdkSerializationRedisSerializer. (Note that, in releases before version 4.3, it was a StringRedisSerializer by default. To restore that behavior, provide a reference to a StringRedisSerializer).
9 The timeout (in milliseconds) to wait until the receive message is fetched. It is typically applied for queue-based limited request-channels.
10 Specifies whether this endpoint expects data from the Redis queue to contain entire Message instances. If this attribute is set to true, the serializer cannot be an empty string, because messages require some form of deserialization (JDK serialization by default).
11 The time (in milliseconds) the listener task should sleep after exceptions on the “right pop” operation before restarting the listener task.

必须使用多个线程配置 task-executor 来进行处理；否则，当 RedisQueueMessageDrivenEndpoint 尝试在出现错误后重新启动侦听器任务时，可能会产生死锁。errorChannel 可用于处理这些错误，以避免重新启动，但最好不要将您的应用程序暴露于可能的死锁情况。请参阅 Spring Framework Reference Manual 以了解可能的 TaskExecutor 实现。

The task-executor has to be configured with more than one thread for processing; otherwise there is a possible deadlock when the RedisQueueMessageDrivenEndpoint tries to restart the listener task after an error. The errorChannel can be used to process those errors, to avoid restarts, but it is preferable to not expose your application to the possible deadlock situation. See Spring Framework Reference Manual for possible TaskExecutor implementations.

Redis Stream Outbound Channel Adapter

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

Spring Integration 5.4 introduced Reactive Redis Stream outbound channel adapter to write Message payload into Redis stream. Outbound Channel adapter uses ReactiveStreamOperations.add(…​) to add a Record to the stream. The following example shows how to use Java configuration and Service class for Redis Stream Outbound Channel Adapter.

@Bean
@ServiceActivator(inputChannel = "messageChannel")
public ReactiveRedisStreamMessageHandler reactiveValidatorMessageHandler(
        ReactiveRedisConnectionFactory reactiveRedisConnectionFactory) {
    ReactiveRedisStreamMessageHandler reactiveStreamMessageHandler =
        new ReactiveRedisStreamMessageHandler(reactiveRedisConnectionFactory, "myStreamKey"); 1
    reactiveStreamMessageHandler.setSerializationContext(serializationContext); 2
    reactiveStreamMessageHandler.setHashMapper(hashMapper); 3
    reactiveStreamMessageHandler.setExtractPayload(true); 4
    return reactiveStreamMessageHandler;
}
1 Construct an instance of ReactiveRedisStreamMessageHandler using ReactiveRedisConnectionFactory and stream name to add records. Another constructor variant is based on a SpEL expression to evaluate a stream key against a request message.
2 Set a RedisSerializationContext used to serialize a record key and value before adding to the stream.
3 Set HashMapper which provides contract between Java types and Redis hashes/maps.
4 If 'true', channel adapter will extract payload from a request message for a stream record to add. Or use the whole message as a value. It defaults to true.

Redis Stream Inbound Channel Adapter

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

Spring Integration 5.4 introduced the Reactive Stream inbound channel adapter for reading messages from a Redis Stream. Inbound channel adapter uses StreamReceiver.receive(…​) or StreamReceiver.receiveAutoAck() based on auto acknowledgement flag to read record from Redis stream. The following example shows how to use Java configuration for Redis Stream Inbound Channel Adapter.

@Bean
public ReactiveRedisStreamMessageProducer reactiveRedisStreamProducer(
       ReactiveRedisConnectionFactory reactiveRedisConnectionFactory) {
ReactiveRedisStreamMessageProducer messageProducer =
            new ReactiveRedisStreamMessageProducer(reactiveRedisConnectionFactory, "myStreamKey"); 1
    messageProducer.setStreamReceiverOptions( 2
                StreamReceiver.StreamReceiverOptions.builder()
                      .pollTimeout(Duration.ofMillis(100))
                      .build());
    messageProducer.setAutoStartup(true); 3
    messageProducer.setAutoAck(false); 4
    messageProducer.setCreateConsumerGroup(true); 5
    messageProducer.setConsumerGroup("my-group"); 6
    messageProducer.setConsumerName("my-consumer"); 7
    messageProducer.setOutputChannel(fromRedisStreamChannel); 8
    messageProducer.setReadOffset(ReadOffset.latest()); 9
    messageProducer.extractPayload(true); 10
    return messageProducer;
}
1 Construct an instance of ReactiveRedisStreamMessageProducer using ReactiveRedisConnectionFactory and stream key to read records.
2 A StreamReceiver.StreamReceiverOptions to consume redis stream using reactive infrastructure.
3 A SmartLifecycle attribute to specify whether this endpoint should start automatically after the application context start or not. It defaults to true. If false, RedisStreamMessageProducer should be started manually messageProducer.start().
4 If false, received messages are not auto acknowledged. The acknowledgement of the message will be deferred to the client consuming message. It defaults to true.
5 If true, a consumer group will be created. During creation of consumer group stream will be created (if not exists yet), too. Consumer group track message delivery and distinguish between consumers. It defaults to false.
6 Set Consumer Group name. It defaults to the defined bean name.
7 Set Consumer name. Reads message as my-consumer from group my-group.
8 The message channel to which to send messages from this endpoint.
9 Define the offset to read message. It defaults to ReadOffset.latest().
10 If 'true', channel adapter will extract payload value from the Record. Otherwise, the whole Record is used as a payload. It defaults to true.

如果 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。

If the autoAck is set to false, the Record in Redis Stream is not acknowledge automatically by the Redis driver, instead an IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK header is added into a message to produce with a SimpleAcknowledgment instance as a value. It is a target integration flow responsibility to call its acknowledge() callback whenever the business logic is done for the message based on such a record. Similar logic is required even when an exception happens during deserialization and errorChannel is configured. So, target error handler must decide to ack or nack such a failed message. Alongside with IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, the ReactiveRedisStreamMessageProducer also populates these headers into the message to produce: RedisHeaders.STREAM_KEY, RedisHeaders.STREAM_MESSAGE_ID, RedisHeaders.CONSUMER_GROUP and RedisHeaders.CONSUMER.

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

Starting with version 5.5, you can configure StreamReceiver.StreamReceiverOptionsBuilder options explicitly on the ReactiveRedisStreamMessageProducer, including the newly introduced onErrorResume function, which is required if the Redis Stream consumer should continue polling when deserialization errors occur. The default function sends a message to the error channel (if provided) with possible acknowledgement for the failed message as it is described above. All these StreamReceiver.StreamReceiverOptionsBuilder are mutually exclusive with an externally provided StreamReceiver.StreamReceiverOptions.

Redis Lock Registry

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

Spring Integration 4.0 introduced the RedisLockRegistry. Certain components (for example, aggregator and resequencer) use a lock obtained from a LockRegistry instance to ensure that only one thread manipulates a group at a time. The DefaultLockRegistry performs this function within a single component. You can now configure an external lock registry on these components. When you use it with a shared MessageGroupStore, you can use the RedisLockRegistry to provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time.

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

When a lock is released by a local thread, another local thread can generally acquire the lock immediately. If a lock is released by a thread using a different registry instance, it can take up to 100ms to acquire the lock.

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

To avoid “hung” locks (when a server fails), the locks in this registry are expired after a default 60 seconds, but you can configure this value on the registry. Locks are normally held for a much smaller time.

因为键可能会过期，尝试解锁已过期的锁会导致抛出异常。但是，受此类锁保护的资源可能已被破坏，因此应将此类异常视为严重异常。你应为过期设置一个足够大的值以防止这种情况发生，但也应设置一个足够低的值，以便在一段时间后服务器故障后可以恢复锁。

Because the keys can expire, an attempt to unlock an expired lock results in an exception being thrown. However, the resources protected by such a lock may have been compromised, so such exceptions should be considered to be severe. You should set the expiry at a large enough value to prevent this condition, but set it low enough that the lock can be recovered after a server failure in a reasonable amount of time.

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

Starting with version 5.0, the RedisLockRegistry implements ExpirableLockRegistry, which removes locks last acquired more than age ago and that are not currently locked.

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

Starting with version 5.5.6, the RedisLockRegistry is support automatically clean up cache for redisLocks in RedisLockRegistry.locks via RedisLockRegistry.setCacheCapacity(). See its JavaDocs for more information.

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

Starting with version 5.5.13, the RedisLockRegistry exposes a setRedisLockType(RedisLockType) option to determine in which mode a Redis lock acquisition should happen:

  • RedisLockType.SPIN_LOCK - the lock is acquired by periodic loop (100ms) checking whether the lock can be acquired. Default.

  • RedisLockType.PUB_SUB_LOCK - The lock is acquired by redis pub-sub subscription.

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

The pub-sub is preferred mode - less network chatter between client Redis server, and more performant - the lock is acquired immediately when subscription is notified about unlocking in the other process. However, the Redis does not support pub-sub in the Master/Replica connections (for example in AWS ElastiCache environment), therefore a busy-spin mode is chosen as a default to make the registry working in any environment.