Spring認證中國教育管理中心-Spring Data Redis架構教程三

原标題：Spring認證中國教育管理中心-Spring Data Redis架構教程三

10.15.支援類

Packageorg.springframework.data.redis.support提供了各種可重用的元件，這些元件依賴于 Redis 作為後備存儲。目前，該包包含基于 Redis 的各種基于 JDK 的接口實作，例如原子計數器和 JDK集合。

原子計數器可以輕松包裝 Redis 密鑰增量，而集合可以輕松管理 Redis 密鑰，同時将存儲暴露或 API 洩​漏降至最低。特别是，RedisSet和RedisZSet接口提供了對 Redis 支援的集合操作的輕松通路，例如intersection和union。RedisList實作List，Queue以及Deque合同（和它們的等效阻擋兄弟姐妹）上的Redis的頂部，露出存儲作為FIFO（先入先出），LIFO（後進先出）或封端的集合以最小的配置. 以下示例顯示了使用 的 bean 的配置RedisList：

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:p="http://www.springframework.org/schema/p" xsi:schemaLocation="

http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd">

<constructor-arg ref="redisTemplate"/>
<constructor-arg value="queue-key"/>           

以下示例顯示了一個 Java 配置示例Deque：

public class AnotherExample {

// injected

private Deque queue;

public void addTag(String tag) {

queue.push(tag);           

}

Spring認證中國教育管理中心-Spring Data Redis架構教程三

如前面的示例所示，消費代碼與實際存儲實作分離。事實上，沒有任何迹象表明在下面使用了Redis。這使得從開發環境到生産環境的轉變變得透明，并大大提高了可測試性（Redis 實作可以用記憶體中的實作代替）。

1.響應式Redis支援

本節介紹反應式 Redis 支援以及如何開始。響應式 Redis 支援自然與指令式 Redis 支援有一定的重疊。

11.1.Redis 要求

Spring Data Redis 目前與Lettuce內建，作為唯一的反應式 Java 連接配接器。Project Reactor用作反應式組合庫。

11.2.使用響應式驅動程式連接配接到 Redis

使用 Redis 和 Spring 時的首要任務之一是通過 IoC 容器連接配接到存儲。為此，需要一個 Java 連接配接器（或綁定）。無論你選擇的庫中，你必須使用

org.springframework.data.redis.connection包及其ReactiveRedisConnection和ReactiveRedisConnectionFactory接口的工作和取得積極connections對Redis的。

11.2.1.Redis 操作模式

Redis 可以作為獨立伺服器運作，使用Redis Sentinel或以Redis 叢集模式運作。 Lettuce支援所有前面提到的連接配接類型。

11.2.2.ReactiveRedisConnection和ReactiveRedisConnectionFactory

ReactiveRedisConnection是 Redis 通信的核心，因為它處理與 Redis 後端的通信。它還自動将底層驅動程式異常轉換為 Spring 一緻的 DAO 異常層次結構，是以您可以在不更改任何代碼的情況下切換連接配接器，因為操作語義保持不變。

ReactiveRedisConnectionFactory建立活動ReactiveRedisConnection執行個體。此外，工廠充當PersistenceExceptionTranslator執行個體，這意味着一旦聲明，它們就可以讓您進行透明的異常轉換——例如，通過使用@Repository注釋和 AOP 進行異常轉換。有關更多資訊，請參閱Spring Framework 文檔中的專用部分。

根據底層配置，工廠可以傳回新連接配接或現有連接配接（如果使用池或共享本機連接配接）。

使用 a 的最簡單方法

ReactiveRedisConnectionFactory是通過 IoC 容器配置适當的連接配接器并将其注入 using 類。

11.2.3.配置生菜連接配接器

Spring Data Redis 通過

org.springframework.data.redis.connection.lettuce包支援Lettuce。

您可以

ReactiveRedisConnectionFactory按如下方式設定生菜：

@Bean

public ReactiveRedisConnectionFactory connectionFactory() {

return new LettuceConnectionFactory("localhost", 6379);

以下示例顯示了一個更複雜的配置，包括 SSL 和逾時，它使用

LettuceClientConfigurationBuilder：

public ReactiveRedisConnectionFactory lettuceConnectionFactory() {

LettuceClientConfiguration clientConfig = LettuceClientConfiguration.builder()

.useSsl().and()
.commandTimeout(Duration.ofSeconds(2))
.shutdownTimeout(Duration.ZERO)
.build();
           

return new LettuceConnectionFactory(new RedisStandaloneConfiguration("localhost", 6379), clientConfig);

有關更詳細的用戶端配置調整，請參閱

LettuceClientConfiguration。

11.3.通過 ReactiveRedisTemplate 處理對象

大多數使用者可能會使用ReactiveRedisTemplate及其對應的包，

org.springframework.data.redis.core. 由于其豐富的功能集，模闆實際上是 Redis 子產品的中心類。該模闆為 Redis 互動提供了進階抽象。雖然ReactiveRedisConnection提供了接受和傳回二進制值 ( ByteBuffer)的低級方法，但模闆負責序列化和連接配接管理，使您無需處理此類細節。

此外，該模闆提供操作視圖（遵循 Redis 指令參考中的分組），提供豐富的通用接口，用于針對特定類型工作，如下表所述：

配置後，模闆是線程安全的，可以跨多個執行個體重複使用。

ReactiveRedisTemplate大多數操作使用基于 Java 的序列化程式。這意味着模闆寫入或讀取的任何對象都通過RedisElementWriter或 進行序列化或反序列化RedisElementReader。序列化上下文在構造時傳遞給模闆，Redis 子產品在

org.springframework.data.redis.serializer包中提供了幾種可用的實作。有關更多資訊，請參閱序列化程式。

以下示例顯示了ReactiveRedisTemplate用于傳回 a 的 a Mono：

@Configuration

class RedisConfiguration {

ReactiveRedisTemplate<String, String> reactiveRedisTemplate(ReactiveRedisConnectionFactory factory) {

return new ReactiveRedisTemplate<>(factory, RedisSerializationContext.string());           

public class Example {

@Autowired

private ReactiveRedisTemplate<String, String> template;

public Mono addLink(String userId, URL url) {

return template.opsForList().leftPush(userId, url.toExternalForm());           

11.4.以字元串為中心的便利課程

因為它是存儲在Redis的是一個鍵和值相當普遍java.lang.String，Redis的子產品提供了一個基于字元串的擴充ReactiveRedisTemplate：

ReactiveStringRedisTemplate。是密集String作業的便捷一站式解決方案。除了綁定到String鍵之外，模闆還使用 String-based RedisSerializationContext，這意味着存儲的鍵和值是人類可讀的（假設在 Redis 和您的代碼中使用相同的編碼）。以下示例顯示ReactiveStringRedisTemplate在使用中：

ReactiveStringRedisTemplate reactiveRedisTemplate(ReactiveRedisConnectionFactory factory) {

return new ReactiveStringRedisTemplate<>(factory);           

private ReactiveStringRedisTemplate redisTemplate;

return redisTemplate.opsForList().leftPush(userId, url.toExternalForm());           

11.5.Redis 消息傳遞/釋出訂閱

Spring Data 為 Redis 提供了專門的消息傳遞內建，在功能和命名上與 Spring Framework 中的 JMS 內建非常相似；事實上，熟悉 Spring 中 JMS 支援的使用者應該會有賓至如歸的感覺。

Redis 消息傳遞大緻可以分為兩個功能區，即消息的生産或釋出和消費或訂閱，是以有快捷方式pubsub（釋出/訂閱）。所述ReactiveRedisTemplate類用于消息生成。對于異步接收，Spring Data 提供了一個專用的消息偵聽器容器，用于消費消息流。僅出于訂閱目的，ReactiveRedisTemplate提供了使用偵聽器容器的精簡替代方案。

包

org.springframework.data.redis.connection并org.springframework.data.redis.listener提供使用 Redis 消息傳遞的核心功能。

11.5.1.發送/釋出消息

要釋出消息，與其他操作一樣，可以使用低級ReactiveRedisConnection或進階ReactiveRedisTemplate. 這兩個實體都提供了一個釋出方法，該方法接受需要發送的消息以及目标通道作為參數。雖然ReactiveRedisConnection需要原始資料，但ReactiveRedisTemplate允許将任意對象作為消息傳入：

// send message through ReactiveRedisConnection

ByteBuffer msg = …

ByteBuffer channel = …

Mono publish = con.publish(msg, channel);

// send message through ReactiveRedisTemplate

ReactiveRedisTemplate template = …

Mono publish = template.convertAndSend("channel", "message");

11.5.2.接收/訂閱消息

在接收端，可以通過直接命名或使用模式比對來訂閱一個或多個頻道。後一種方法非常有用，因為它不僅允許使用一個指令建立多個訂閱，而且還可以偵聽訂閱時尚未建立的頻道（隻要它們比對模式）。

在底層，ReactiveRedisConnection提供subscribe和pSubscribe方法映射Redis指令以分别按模式按頻道訂閱。請注意，可以使用多個通道或模式作為參數。要更改訂閱，隻需查詢 的頻道和模式ReactiveSubscription。

Spring Data Redis 中的響應式訂閱指令是非阻塞的，并且可能會在不發出元素的情況下結束。

如上所述，一旦訂閱，連接配接就會開始等待消息。除了添加新訂閱或修改/取消現有訂閱之外，不能對其調用其他指令。subscribe、pSubscribe、unsubscribe、 或以外的指令pUnsubscribe是非法的，會導緻異常。

為了接收消息，需要擷取消息流。請注意，訂閱僅釋出在該特定訂閱中注冊的頻道和模式的消息。消息流本身是一個熱序列，它在不考慮需求的情況下生成元素。確定注冊足夠的需求以免耗盡消息緩沖區。

消息偵聽器容器

Spring Data 提供

ReactiveRedisMessageListenerContainer它代表使用者完成所有繁重的轉換和訂閱狀态管理。

ReactiveRedisMessageListenerContainer充當消息偵聽器容器。它用于從 Redis 通道接收消息并公開一個消息流，該消息流通過應用反序列化發出通道消息。負責注冊接收消息、資源擷取與釋放、異常轉換等。這允許您作為應用程式開發人員編寫與接收消息（并對其作出反應）相關的（可能很複雜）業務邏輯，并将樣闆 Redis 基礎設施問題委托給架構。消息流在釋出者訂閱時在 Redis 中注冊訂閱，如果訂閱被取消則取消注冊。

此外，為了最小化應用程式占用空間，

ReactiveRedisMessageListenerContainer允許多個偵聽器共享一個連接配接和一個線程，即使它們不共享訂閱。是以，無論應用程式跟蹤多少個偵聽器或通道，運作時成本在其整個生命周期内都将保持不變。此外，容器允許運作時配置更改，是以可以在應用程式運作時添加或删除偵聽器，而無需重新啟動。此外，容器使用惰性訂閱方法，ReactiveRedisConnection僅在需要時使用 - 如果所有偵聽器都取消訂閱，則會自動執行清理。

消息偵聽器容器本身不需要外部線程資源。它使用驅動程式線程來釋出消息。

ReactiveRedisConnectionFactory factory = …

ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(factory);

Flux<ChannelMessage<String, String>> stream = container.receive(ChannelTopic.of("my-channel"));

要等待并確定正确訂閱，您可以使用receiveLater傳回Mono<Flux>. 将得到的Mono完成具有内部出版商作為完成訂閱給定主題的結果。通過攔截onNext信号，您可以同步伺服器端訂閱。

Mono<Flux<ChannelMessage<String, String>>> stream = container.receiveLater(ChannelTopic.of("my-channel"));

stream.doOnNext(inner -> // notification hook when Redis subscriptions are synchronized with the server)

.flatMapMany(Function.identity())
.…;           

通過模闆API訂閱

如上所述，您可以直接使用ReactiveRedisTemplate訂閱頻道/模式。這種方法提供了一種直接但有限的解決方案，因為您無法在初始訂閱之後添加訂閱。盡管如此，您仍然可以通過傳回的Flux使用例如控制消息流。take(Duration). 完成讀取、出錯或取消時，所有綁定資源将再次釋放。

redisTemplate.listenToChannel("channel1", "channel2").doOnNext(msg -> {

// message processing ...           

}).subscribe();

11.6.反應式腳本

您可以使用響應式基礎架構運作 Redis 腳本，ReactiveScriptExecutor最好通過ReactiveRedisTemplate.

public Flux theAnswerToLife() {

DefaultRedisScript<Long> script = new DefaultRedisScript<>();
script.setLocation(new ClassPathResource("META-INF/scripts/42.lua"));
script.setResultType(Long.class);

return reactiveTemplate.execute(script);           

有關腳本指令的更多詳細資訊，請參閱腳本部分。

12.Redis叢集

使用Redis 叢集需要 Redis Server 3.0+ 版本。有關更多資訊，請參閱叢集教程。

12.1.啟用Redis叢集

叢集支援基于與非叢集通信相同的建構塊。RedisClusterConnection是 的擴充RedisConnection，處理與 Redis 叢集的通信并将錯誤轉換為 Spring DAO 異常層次結構。 RedisClusterConnection執行個體是使用 來建立的RedisConnectionFactory，必須使用關聯的 進行設定RedisClusterConfiguration，如以下示例所示：

示例 5. Redis 叢集的示例 RedisConnectionFactory 配置

@Component

@ConfigurationProperties(prefix = "spring.redis.cluster")

public class ClusterConfigurationProperties {

/*
 * spring.redis.cluster.nodes[0] = 127.0.0.1:7379
 * spring.redis.cluster.nodes[1] = 127.0.0.1:7380
 * ...
 */
List<String> nodes;

/**
 * Get initial collection of known cluster nodes in format {@code host:port}.
 *
 * @return
 */
public List<String> getNodes() {
    return nodes;
}

public void setNodes(List<String> nodes) {
    this.nodes = nodes;
}           

public class AppConfig {

/**
 * Type safe representation of application.properties
 */
@Autowired ClusterConfigurationProperties clusterProperties;

public @Bean RedisConnectionFactory connectionFactory() {

    return new JedisConnectionFactory(
        new RedisClusterConfiguration(clusterProperties.getNodes()));
}           

RedisClusterConfiguration也可以通過定義PropertySource并具有以下屬性：

配置屬性

spring.redis.cluster.nodes：逗号分隔的主機：端口對清單。

spring.redis.cluster.max-redirects：允許的叢集重定向數。

初始配置将驅動程式庫指向一組初始叢集節點。實時叢集重新配置導緻的更改僅保留在本機驅動程式中，不會寫回到配置中。

12.2.使用 Redis 叢集連接配接

如前所述，Redis Cluster 的行為與單節點​​ Redis 甚至 Sentinel 監控的主副本環境不同。這是因為自動分片将密鑰映射到 16384 個插槽之一，這些插槽分布在節點上。是以，涉及多個鍵的指令必須斷言所有鍵映射到完全相同的槽以避免跨槽錯誤。單個叢集節點僅提供一組專用密鑰。針對一個特定伺服器發出的指令僅傳回該伺服器提供的那些密鑰的結果。作為一個簡單的例子，考慮KEYS指令。當發送到叢集環境中的伺服器時，它隻傳回請求發送到的節點所服務的密鑰，而不一定傳回叢集内的所有密鑰。是以，要擷取叢集環境中的所有密鑰，您必須從所有已知的主節點讀取密鑰。

雖然将特定鍵重定向到相應的插槽服務節點由驅動程式庫處理，但更進階别的功能，例如跨節點收集資訊或向叢集中的所有節點發送指令，由RedisClusterConnection. 以前面的鍵示例為例，這意味着該keys(pattern)方法會擷取叢集中的每個主節點，并同時KEYS在每個主節點上運作指令，同時擷取結果并傳回累積的鍵集。僅請求單個節點的鍵RedisClusterConnection為這些方法提供了重載（例如，keys(node, pattern)）。

ARedisClusterNode可以從

RedisClusterConnection.clusterGetNodes主機和端口或節點 Id 中擷取或建構。

以下示例顯示了在叢集中運作的一組指令：

示例 6. 跨叢集運作指令的示例

[email protected]:7379 > cluster nodes

6b38bb... 127.0.0.1:7379 master - 0 0 25 connected 0-5460

7bb78c... 127.0.0.1:7380 master - 0 1449730618304 2 connected 5461-10922

164888... 127.0.0.1:7381 master - 0 1449730618304 3 connected 10923-16383

b8b5ee... 127.0.0.1:7382 slave 6b38bb... 0 1449730618304 25 connected

RedisClusterConnection connection = connectionFactory.getClusterConnnection();

connection.set("thing1", value);

connection.set("thing2", value);

connection.keys("*");

connection.keys(NODE_7379, "*");

connection.keys(NODE_7380, "*");

connection.keys(NODE_7381, "*");

connection.keys(NODE_7382, "*");

主節點服務插槽 0 到 5460 複制到副本 7382

主節點服務時隙 5461 到 10922

主節點服務時隙 10923 到 16383

副本節點在 7379 處持有主節點的副本

請求路由到 7381 服務時隙 12182 處的節點

請求路由到 7379 服務時隙 5061 處的節點

請求路由到節點 7379, 7380, 7381 → [thing1, thing2]

請求路由到節點 7379 → [thing2]

請求路由到節點 7380 → []

請求路由到節點 7381 → [thing1]

請求路由到節點 7382 → [thing2]

當所有鍵都映射到同一個插槽時，本機驅動程式庫會自動提供跨插槽請求，例如MGET. 但是，如果情況并非如此，RedisClusterConnection則GET針對時隙服務節點運作多個并行指令并再次傳回累積結果。這比單槽方法性能低，是以應謹慎使用。如果有疑問，請考慮通過在大括号中提供字首（例如{my-prefix}.thing1和 ）将密鑰固定到同一插槽{my-prefix}.thing2，這将映射到相同的插槽編号。以下示例顯示了跨槽請求處理：

示例 7. 跨時隙請求處理示例

7bb...

connection.set("thing1", value); // slot: 12182

connection.set("{thing1}.thing2", value); // slot: 12182

connection.set("thing2", value); // slot: 5461

connection.mGet("thing1", "{thing1}.thing2");

connection.mGet("thing1", "thing2");

與之前示例中的配置相同。

鍵映射到同一插槽 → 127.0.0.1:7381 MGET thing1 {thing1}.thing2

鍵映射到不同的插槽并被拆分為路由到相應節點的單個插槽

→ 127.0.0.1:7379 GET thing2

→ 127.0.0.1:7381 GET thing1

前面的示例示範了 Spring Data Redis 遵循的一般政策。請注意，某些操作可能需要将大量資料加載到記憶體中才能計算所需的指令。此外，并非所有跨時隙請求都可以安全地移植到多個單時隙請求中，如果誤用（例如，PFCOUNT）會出錯。

12.3.與RedisTemplate和ClusterOperations

請參閱通過 RedisTemplate 處理對象部分，了解有關RedisTemplate.

RedisTemplate#keySerializer使用任何 JSON 進行 設定時要小心RedisSerializers，因為更改 JSON 結構會對哈希槽計算産生直接影響。

RedisTemplate通過ClusterOperations接口提供對叢集特定操作的通路，該接口可以從

RedisTemplate.opsForCluster(). 這使您可以在叢集内的單個節點上顯式運作指令，同時保留為模闆配置的序列化和反序列化功能。它還提供管理指令（例如CLUSTER MEET）或更進階的操作（例如重新分片）。

下面的例子示範如何通路RedisClusterConnection使用RedisTemplate：

實施例8.通路RedisClusterConnection與RedisTemplate

ClusterOperations clusterOps = redisTemplate.opsForCluster();

clusterOps.shutdown(NODE_7379);

在 7379 關閉節點，交叉手指有一個可以接管的副本。