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 鍵被移除。