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

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

10.11.Redis 流

Redis Streams 以抽象方法對日志資料結構進行模組化。通常，日志是僅附加的資料結構，從一開始就在随機位置或通過流式傳輸新消息使用。

在Redis 參考文檔 中了解有關 Redis Streams 的更多資訊。

Redis Streams 大緻可以分為兩個功能領域：

追加記錄

消費記錄

盡管此模式與Pub/Sub有相似之處，但主要差別在于消息的持久性以及它們的消費方式。

雖然 Pub/Sub 依賴于瞬時消息的廣播（即，如果您不聽，就會錯過一條消息），而 Redis Stream 使用持久的、僅附加的資料類型，該資料類型會保留消息，直到流被修剪為止。消費的另一個差別是 Pub/Sub 注冊了伺服器端訂閱。Redis 将到達的消息推送到用戶端，而 Redis Streams 需要主動輪詢。

在

org.springframework.data.redis.connection和org.springframework.data.redis.stream軟體包提供了對Redis的資料流的核心功能。

10.11.1.附加

要發送記錄，您可以像其他操作一樣使用低級RedisConnection或進階StreamOperations. 兩個實體都提供add( xAdd) 方法，該方法接受記錄和目标流作為參數。雖然RedisConnection需要原始資料（位元組數組），但StreamOperations允許将任意對象作為記錄傳入，如下例所示：

// append message through connection

RedisConnection con = …

byte[] stream = …

ByteRecord record = StreamRecords.rawBytes(…).withStreamKey(stream);

con.xAdd(record);

// append message through RedisTemplate

RedisTemplate template = …

StringRecord record = StreamRecords.string(…).withStreamKey("my-stream");

template.streamOps().add(record);

流記錄攜帶一個Map, 鍵值元組，作為它們的有效負載。将記錄附加到流會傳回RecordId可用作進一步參考的 。

10.11.2.消費

在消費方面，一個人可以消費一個或多個流。Redis Streams 提供讀取指令，允許從已知流内容内和流端之外的任意位置（随機通路）消費流以消費新的流記錄。

在底層，RedisConnection提供了xRead和xReadGroup方法，分别映射了消費組内讀取和讀取的Redis指令。請注意，多個流可以用作參數。

Redis 中的訂閱指令可能會阻塞。也就是說，調用xRead連接配接會導緻目前線程在開始等待消息時阻塞。隻有在讀取指令逾時或收到消息時才會釋放線程。

要使用流消息，可以在應用程式代碼中輪詢消息，或者通過消息偵聽器容器使用兩種異步接收之一，指令式或反應式。每次有新記錄到達時，容器都會通知應用程式代碼。

同步接收

雖然流消費通常與異步處理相關聯，但也可以同步消費消息。重載StreamOperations.read(…)方法提供此功能。在同步接收期間，調用線程可能會阻塞，直到消息可用。該屬性StreamReadOptions.block指定接收者在放棄等待消息之前應等待多長時間。

// Read message through RedisTemplate

List<MapRecord<K, HK, HV>> messages = template.streamOps().read(StreamReadOptions.empty().count(2),

StreamOffset.latest("my-stream"));
           

List<MapRecord<K, HK, HV>> messages = template.streamOps().read(Consumer.from("my-group", "my-consumer"),

StreamReadOptions.empty().count(2),
            StreamOffset.create("my-stream", ReadOffset.lastConsumed()))           

通過消息偵聽器容器異步接收

由于其阻塞性質，低級輪詢沒有吸引力，因為它需要對每個消費者進行連接配接和線程管理。為了緩解這個問題，Spring Data 提供了消息偵聽器，它可以完成所有繁重的工作。如果您熟悉 EJB 和 JMS，您應該會發現這些概念很熟悉，因為它被設計為盡可能接近 Spring Framework 及其消息驅動的 POJO (MDP) 中的支援。

Spring Data 附帶了兩種針對所使用的程式設計模型量身定制的實作：

StreamMessageListenerContainer充當指令式程式設計模型的消息偵聽器容器。它用于使用 Redis Stream 中的記錄并驅動StreamListener注入其中的執行個體。

StreamReceiver提供消息偵聽器的反應式變體。它用于将來自 Redis Stream 的消息作為潛在的無限流使用，并通過Flux.

StreamMessageListenerContainer并StreamReceiver負責消息接收和分派到偵聽器中進行處理的所有線程。消息偵聽器容器/接收器是 MDP 和消息提供者之間的中介，負責注冊接收消息、資源擷取和釋放、異常轉換等。這讓您作為應用程式開發人員可以編寫與接收消息（并對其作出反應）相關的（可能很複雜）業務邏輯，并将樣闆 Redis 基礎設施問題委托給架構。

這兩個容器都允許運作時配置更改，以便您可以在應用程式運作時添加或删除訂閱，而無需重新啟動。此外，容器使用惰性訂閱方法，RedisConnection僅在需要時使用。如果所有偵聽器都取消訂閱，它會自動執行清理，并釋放線程。

至關重要的StreamMessageListenerContainer

以類似于 EJB 世界中的消息驅動 Bean (MDB) 的方式，流驅動 POJO (SDP) 充當流消息的接收器。SDP 的一個限制是它必須實作

org.springframework.data.redis.stream.StreamListener接口。還請注意，在您的 POJO 在多個線程上接收消息的情況下，確定您的實作是線程安全的很重要。

class ExampleStreamListener implements StreamListener<String, MapRecord<String, String, String>> {

@Override
public void onMessage(MapRecord<String, String, String> message) {

    System.out.println("MessageId: " + message.getId());
    System.out.println("Stream: " + message.getStream());
    System.out.println("Body: " + message.getValue());
}           

}

StreamListener 表示一個函數式接口，是以可以使用它們的 Lambda 形式重寫實作：

message -> {

System.out.println("MessageId: " + message.getId());
System.out.println("Stream: " + message.getStream());
System.out.println("Body: " + message.getValue());           

};

一旦你實作了你的StreamListener，就可以建立一個消息監聽器容器并注冊一個訂閱：

RedisConnectionFactory connectionFactory = …

StreamListener<String, MapRecord<String, String, String>> streamListener = …

StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> containerOptions = StreamMessageListenerContainerOptions

.builder().pollTimeout(Duration.ofMillis(100)).build();
           

StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer.create(connectionFactory,

containerOptions);
           

Subscription subscription = container.receive(StreamOffset.fromStart("my-stream"), streamListener);

請參閱各種消息偵聽器容器的 Javadoc，以擷取每個實作支援的功能的完整描述。

反應式StreamReceiver

流資料源的反應性消費通常通過一系列Flux事件或消息發生。反應式接收器實作提供了StreamReceiver及其重載的receive(…)消息。與

StreamMessageListenerContainer利用驅動程式提供的線程資源相比，反應式方法需要更少的基礎設施資源，例如線程。接收流是一個需求驅動的釋出者StreamMessage：

Flux<MapRecord<String, String, String>> messages = …

return messages.doOnNext(it -> {

System.out.println("MessageId: " + message.getId());
System.out.println("Stream: " + message.getStream());
System.out.println("Body: " + message.getValue());           

});

現在我們需要建立StreamReceiver并注冊一個訂閱來消費流消息：

ReactiveRedisConnectionFactory connectionFactory = …

StreamReceiverOptions<String, MapRecord<String, String, String>> options = StreamReceiverOptions.builder().pollTimeout(Duration.ofMillis(100))

.build();           

StreamReceiver<String, MapRecord<String, String, String>> receiver = StreamReceiver.create(connectionFactory, options);

Flux<MapRecord<String, String, String>> messages = receiver.receive(StreamOffset.fromStart("my-stream"));

需求驅動的消費使用背壓信号來激活和停用輪詢。StreamReceiver如果需求得到滿足，訂閱将暫停輪詢，直到訂閱者發出進一步的需求信号。根據ReadOffset政策，這可能會導緻消息被跳過。

Acknowledge政策

當您通過 a 閱讀消息時Consumer Group，伺服器将記住給定的消息已傳遞并将其添加到待處理條目清單 (PEL)。已發送但尚未确認的消息清單。

消息必須通過确認

StreamOperations.acknowledge才能從待處理條目清單中删除，如下面的片段所示。

StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = ...

container.receive(Consumer.from("my-group", "my-consumer"),

StreamOffset.create("my-stream", ReadOffset.lastConsumed()),
msg -> {

    // ...
    redisTemplate.opsForStream().acknowledge("my-group", msg); 
});           

從組my-group 中讀取為my-consumer。接收到的消息不被确認。

處理後确認消息。

要在接收時自動确認消息，請使用receiveAutoAck而不是receive.

ReadOffset政策

流讀取操作接受讀取偏移量規範以從給定偏移量開始消費消息。ReadOffset表示讀取偏移規範。Redis 支援 3 種偏移量變體，具體取決于您是獨立使用流還是在消費者組中使用流：

ReadOffset.latest() – 閱讀最新消息。

ReadOffset.from(…) – 在特定消息 ID 之後閱讀。

ReadOffset.lastConsumed() – 在最後消費的消息 ID 之後讀取（僅限消費者組）。

在基于消息容器的消費上下文中，我們需要在消費消息時提前（或增加）讀取偏移量。推進取決于請求ReadOffset和消費模式（有/沒有消費者組）。以下矩陣解釋了容器如何前進ReadOffset：

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

從特定的消息 ID 和最後消費的消息中讀取可以被視為安全操作，可確定消費附加到流的所有消息。使用最新的消息進行讀取可以跳過輪詢操作處于死時間狀态時添加到流中的消息。輪詢引入了一個死區時間，其中消息可以在各個輪詢指令之間到達。流消費不是線性連續讀取，而是拆分為重複XREAD調用。

序列化

發送到流的任何記錄都需要序列化為其二進制格式。由于流與散列資料結構的接近性，流鍵、字段名稱和值使用在RedisTemplate.

請確定檢查RedisSerializers in use 并注意，如果您決定不使用任何序列化程式，則需要確定這些值已經是二進制的。

對象映射

簡單值

StreamOperations允許通過 将簡單值ObjectRecord直接附加到流中，而無需将這些值放入Map結構中。然後将該值配置設定給有效載荷字段，并且可以在讀回該值時提取該值。

ObjectRecord<String, String> record = StreamRecords.newRecord()

.in("my-stream")
.ofObject("my-value");
           

redisTemplate()

.opsForStream()
.add(record); 
           

List<ObjectRecord<String, String>> records = redisTemplate()

.opsForStream()
.read(String.class, StreamOffset.fromStart("my-stream"));           

XADD my-stream * "_class" "java.lang.String" "_raw" "my-value"

ObjectRecords 與所有其他記錄都經過完全相同的序列化過程，是以也可以使用傳回 a 的無類型讀取操作擷取 Record MapRecord。

複數值

可以通過 3 種方式向流中添加複雜值：

使用例如轉換為簡單值。一個字元串 JSON 表示。

使用合适的RedisSerializer.

Map使用 a将值轉換為适合序列化的值HashMapper。

第一個變體是最直接的變體，但忽略了流結構提供的字段值功能，流中的值仍然可以被其他消費者讀取。第二個選項與第一個選項具有相同的好處，但可能會導緻非常具體的消費者限制，因為所有消費者都必須實作完全相同的序列化機制。該HashMapper方法使用蒸汽散列結構稍微複雜一點，但将源扁平化。隻要選擇了合适的序列化程式組合，其他消費者仍然能夠讀取記錄。

HashMappers 将有效負載轉換為Map具有特定類型的 a。確定使用能夠（反）序列化散列的散列鍵和散列值序列化程式。

ObjectRecord<String, User> record = StreamRecords.newRecord()

.in("user-logon")
.ofObject(new User("night", "angel"));
           

.opsForStream()
.add(record); 
           

List<ObjectRecord<String, User>> records = redisTemplate()

.opsForStream()
.read(User.class, StreamOffset.fromStart("user-logon"));           

XADD 使用者登入 * "_class" "com.example.User" "firstname" "night" "lastname" "angel"

StreamOperations預設使用ObjectHashMapper。您可以HashMapper在擷取時提供一個适合您的要求StreamOperations。

.opsForStream(new Jackson2HashMapper(true))
.add(record);            

XADD 使用者登入 * "firstname" "night" "@class" "com.example.User" "lastname" "angel"

AStreamMessageListenerContainer可能不知道@TypeAlias域類型上使用的任何内容，因為那些需要通過MappingContext. 確定RedisMappingContext使用initialEntitySet.

@Bean

RedisMappingContext redisMappingContext() {

RedisMappingContext ctx = new RedisMappingContext();
ctx.setInitialEntitySet(Collections.singleton(Person.class));
return ctx;           

RedisConverter redisConverter(RedisMappingContext mappingContext) {

return new MappingRedisConverter(mappingContext);           

ObjectHashMapper hashMapper(RedisConverter converter) {

return new ObjectHashMapper(converter);           

StreamMessageListenerContainer streamMessageListenerContainer(RedisConnectionFactory connectionFactory, ObjectHashMapper hashMapper) {

StreamMessageListenerContainerOptions<String, ObjectRecord<String, Object>> options = StreamMessageListenerContainerOptions.builder()
        .objectMapper(hashMapper)
        .build();

return StreamMessageListenerContainer.create(connectionFactory, options);           

10.12.Redis 事務

Redis的提供支援的交易通過multi，exec和discard指令。這些操作在 上可用RedisTemplate。但是，RedisTemplate不能保證在同一個連接配接中運作事務中的所有操作。

Spring Data Redis 提供了SessionCallback接口，供需要對同一個 執行多個操作connection時使用，例如使用Redis 事務時。以下示例使用該multi方法：

//execute a transaction

List

operations.multi();
operations.opsForSet().add("key", "value1");

// This will contain the results of all operations in the transaction
return operations.exec();           

System.out.println("Number of items added to set: " + txResults.get(0));

RedisTemplate使用其值、散列鍵和散列值序列化器exec在傳回之前反序列化所有結果。還有一種exec方法可以讓您為事務結果傳遞自定義序列化程式。

從 1.1 版開始，exec對RedisConnection和的方法進行了重要更改RedisTemplate。以前，這些方法直接從連接配接器傳回事務的結果。這意味着資料類型通常與從 的方法傳回的資料類型不同RedisConnection。例如，zAdd傳回一個布爾值，訓示元素是否已添加到排序集中。大多數連接配接器将此值傳回為 long，并且 Spring Data Redis 執行轉換。另一個常見的差別是，大多數連接配接器OK為諸如set. 這些回複通常會被 Spring Data Redis 丢棄。在 1.1 之前，未對exec. 此外，結果沒有反序列化RedisTemplate，是以它們通常包含原始位元組數組。如果此更改破壞了您的應用程式，請設定

convertPipelineAndTxResults為falseon 您RedisConnectionFactory以禁用此行為。

10.12.1.@交易支援

預設情況下，RedisTemplate不參與托管 Spring 事務。如果您想RedisTemplate在使用@Transactional或時使用 Redis 事務TransactionTemplate，則需要RedisTemplate通過設定為每個顯式啟用事務支援

setEnableTransactionSupport(true)。啟用事務支援綁定RedisConnection到由ThreadLocal. 如果事務完成且沒有錯誤，Redis 事務将使用 送出EXEC，否則使用 復原DISCARD。Redis 事務是面向批處理的。在正在進行的事務期間發出的指令被排隊，并且僅在送出事務時應用。

Spring Data Redis 在正在進行的事務中區分隻讀和寫指令。隻讀指令，例如KEYS，通過管道傳輸到新的（非線程綁定）RedisConnection以允許讀取。寫入指令由RedisTemplate送出排隊并在送出時應用。

以下示例顯示了如何配置事務管理：

示例 3. 啟用事務管理的配置

@Configuration

@EnableTransactionManagement

public class RedisTxContextConfiguration {

public StringRedisTemplate redisTemplate() {

StringRedisTemplate template = new StringRedisTemplate(redisConnectionFactory());
// explicitly enable transaction support
template.setEnableTransactionSupport(true);              
return template;           

public RedisConnectionFactory redisConnectionFactory() {

// jedis || Lettuce           

public PlatformTransactionManager transactionManager() throws SQLException {

return new DataSourceTransactionManager(dataSource());              

public DataSource dataSource() throws SQLException {

// ...           

配置 Spring Context 以啟用聲明式事務管理。

配置RedisTemplate為通過将連接配接綁定到目前線程來參與事務。

事務管理需要一個

PlatformTransactionManager. Spring Data Redis 不附帶PlatformTransactionManager實作。假設您的應用程式使用 JDBC，Spring Data Redis 可以使用現有的事務管理器參與事務。

以下示例分别示範了使用限制：

示例 4. 使用限制

// must be performed on thread-bound connection

template.opsForValue().set("thing1", "thing2");

// read operation must be run on a free (not transaction-aware) connection

template.keys("*");

// returns null as values set within a transaction are not visible

template.opsForValue().get("thing1");

10.13.流水線

Redis 提供對流水線的支援，這涉及向伺服器發送多個指令而無需等待回複，然後一步讀取回複。當您需要連續發送多個指令時，流水線可以提高性能，例如将許多元素添加到同一個 List。

Spring Data Redis 提供了多種RedisTemplate在管道中運作指令的方法。如果你不關心流水線操作的結果，你可以使用标準execute方法，傳遞true的pipeline參數。這些executePipelined方法運作提供的RedisCallback或SessionCallback在管道中并傳回結果，如以下示例所示：

//pop a specified number of items from a queue

public Object doInRedis(RedisConnection connection) throws DataAccessException {
  StringRedisConnection stringRedisConn = (StringRedisConnection)connection;
  for(int i=0; i< batchSize; i++) {
    stringRedisConn.rPop("myqueue");
  }
return null;           

前面的示例從管道中的隊列中批量右彈出項目。在results List包含了所有的彈出項目。RedisTemplate在傳回之前使用其值、哈希鍵和哈希值序列化器對所有結果進行反序列化，是以前面示例中的傳回項是字元串。還有其他executePipelined方法可讓您為流水線結果傳遞自定義序列化程式。

請注意，從 傳回的值RedisCallback必須是null，因為為了傳回流水線指令的結果而丢棄該值。

Lettuce 驅動程式支援細粒度的重新整理控制，允許在指令出現時重新整理、緩沖或在連接配接關閉時發送它們。

LettuceConnectionFactory factory = // ...

factory.setPipeliningFlushPolicy(PipeliningFlushPolicy.buffered(3));

在本地緩沖并在每第三個指令後重新整理。

10.14.Redis 腳本

Redis 2.6 及更高版本支援通過eval和evalsha指令運作 Lua 腳本。Spring Data Redis 為運作腳本提供了進階抽象，該腳本處理序列化并自動使用 Redis 腳本緩存。

腳本可以通過調用運作execute的方法RedisTemplate和ReactiveRedisTemplate。兩者都使用可配置ScriptExecutor（或ReactiveScriptExecutor）來運作提供的腳本。預設情況下，ScriptExecutor(or ReactiveScriptExecutor) 負責序列化提供的鍵和參數并反序列化腳本結果。這是通過模闆的鍵和值序列化器完成的。還有一個額外的重載，允許您為腳本參數和結果傳遞自定義序列化程式。

預設ScriptExecutor通過檢索腳本的 SHA1 并嘗試首先運作來優化性能，如果腳本尚未出現在 Redis 腳本緩存中evalsha，eval則回退到。

以下示例使用 Lua 腳本運作常見的“檢查并設定”場景。這是 Redis 腳本的理想用例，因為它需要原子地運作一組指令，并且一個指令的行為受另一個指令的結果影響。

public RedisScript script() {

ScriptSource scriptSource = new ResourceScriptSource(new ClassPathResource("META-INF/scripts/checkandset.lua"));

return RedisScript.of(scriptSource, Boolean.class);

public class Example {

@Autowired

RedisScript script;

public boolean checkAndSet(String expectedValue, String newValue) {

return redisTemplate.execute(script, singletonList("key"), asList(expectedValue, newValue));           

-- checkandset.lua

local current = redis.call('GET', KEYS[1])

if current == ARGV[1]

then redis.call('SET', KEYS[1], ARGV[2])

return true

end

return false

前面的代碼配置了一個RedisScript指向名為 的檔案checkandset.lua，它應該傳回一個布爾值。該腳本resultType應該是一個Long，Boolean，List或反序列化的值類型。null如果腳本傳回丢棄狀态（特别是OK），也可能是這樣。

最好DefaultRedisScript在應用程式上下文中配置單個執行個體，以避免在每次腳本運作時重新計算腳本的 SHA1。

然後checkAndSet上面的方法運作腳本。腳本可以SessionCallback作為事務或管道的一部分在内部運作。有關更多資訊，請參閱“ Redis 事務”和“流水線”。

Spring Data Redis 提供的腳本支援還允許您使用 Spring Task 和 Scheduler 抽象來安排 Redis 腳本定期運作。有關更多詳細資訊，請參閱Spring 架構文檔。

10.14.1.Redis緩存

在 2.0 中更改

Spring Redis通過包提供了 Spring緩存抽象的

org.springframework.data.redis.cache實作。要将 Redis 用作支援實作，請添加RedisCacheManager到您的配置中，如下所示：

public RedisCacheManager cacheManager(RedisConnectionFactory connectionFactory) {

return RedisCacheManager.create(connectionFactory);           

RedisCacheManager可以使用 配置行為RedisCacheManagerBuilder，讓您設定預設RedisCacheConfiguration、事務行為和預定義緩存。

RedisCacheManager cm = RedisCacheManager.builder(connectionFactory)

.cacheDefaults(defaultCacheConfig())
.withInitialCacheConfigurations(singletonMap("predefined", defaultCacheConfig().disableCachingNullValues()))
.transactionAware()
.build();           

如前面的示例所示，RedisCacheManager允許在每個緩存的基礎上定義配置。

RedisCachecreated with的行為是用RedisCacheManager定義的RedisCacheConfiguration。該配置允許您設定密鑰到期時間、字首和RedisSerializer實作與二進制存儲格式之間的轉換，如以下示例所示：

RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()

.entryTtl(Duration.ofSeconds(1))
.disableCachingNullValues();           

RedisCacheManager預設為無鎖RedisCacheWriter讀取和寫入二進制值。無鎖緩存提高了吞吐量。缺少條目鎖定可能會導緻putIfAbsent和clean方法出現重疊的非原子指令，因為這些指令需要将多個指令發送到 Redis。鎖定對應物通過設定顯式鎖定密鑰并檢查此密鑰的存在來防止指令重疊，這會導緻額外的請求和潛在的指令等待時間。

鎖定适用于緩存級别，而不是每個緩存條目。

可以選擇加入鎖定行為，如下所示：

RedisCacheManager cm = RedisCacheManager.build(RedisCacheWriter.lockingRedisCacheWriter(connectionFactory))

.cacheDefaults(defaultCacheConfig())
...           

預設情況下，key緩存條目的any以實際緩存名稱作為字首，後跟兩個冒号。此行為可以更改為靜态和計算字首。

以下示例顯示了如何設定靜态字首：

// static key prefix

RedisCacheConfiguration.defaultCacheConfig().prefixKeysWith("( ͡° ᴥ ͡°)");

The following example shows how to set a computed prefix:

// computed key prefix

RedisCacheConfiguration.defaultCacheConfig().computePrefixWith(cacheName -> "¯\_(ツ)_/¯" + cacheName);

緩存實作預設使用KEYS和DEL清除緩存。KEYS可能會導緻大鍵空間的性能問題。是以，RedisCacheWriter可以使用 a 建立預設值BatchStrategy以切換到SCAN基于 - 的批處理政策。該SCAN政策需要批量大小以避免過多的 Redis 指令往返：

RedisCacheManager cm = RedisCacheManager.build(RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory, BatchStrategies.scan(1000)))

.cacheDefaults(defaultCacheConfig())
...           

該KEYS批次政策是使用任何驅動程式和Redis的操作模式（獨立，叢集）的全面支援。SCAN使用 Lettuce 驅動程式時完全支援。JedisSCAN僅支援非叢集模式。

下表列出了 的預設設定RedisCacheManager：

預設情況下RedisCache，統計資訊被禁用。使用

RedisCacheManagerBuilder.enableStatistics()收集當地的命中和未命中通過RedisCache#getStatistics()，傳回所收集資料的快照。