redisson中如何使用分布式集合

這篇文章將為大家詳細講解有關redisson中如何使用分布式集合，文章內容質量較高，因此小編分享給大家做個參考，希望大家閱讀完這篇文章后對相關知識有一定的了解。

redisson分布式集合

1.映射（Map）

Redisson的分布式Map結構的RMap Java對象實現了java.util.concurrent.ConcurrentMap接口和java.util.Map接口。同時還保持了元素的插入順序。該對象的最大容量受Redis限制，最大元素數量是4 294 967 295個。

RMap<String, SomeObject> map = redisson.getMap("anyMap");
SomeObject prevObject = map.put("123", new SomeObject());
SomeObject currentObject = map.putIfAbsent("323", new SomeObject());
SomeObject obj = map.remove("123");

map.fastPut("321", new SomeObject());
map.fastRemove("321");

Future<SomeObject> putAsyncFuture = map.putAsync("321");
Future<Void> fastPutAsyncFuture = map.fastPutAsync("321");

map.fastPutAsync("321", new SomeObject());
map.fastRemoveAsync("321");

1.1. 映射（Map）淘汰機制（Eviction）

Redisson的分布式的RMapCache Java對象在基于RMap的前提下實現了針對單個元素的淘汰機制。同時仍然保留了元素的插入順序。由于RMapCache是基于RMap實現的，使它同時繼承了java.util.concurrent.ConcurrentMap接口和java.util.Map接口。Redisson提供的Spring Cache整合 正是基于這樣的對象結構來實現的。

目前的Redis自身并不支持哈希（Hash）當中的元素淘汰，因此所有過期元素都是通過org.redisson.EvictionScheduler實例來實現定期清理的。為了保證資源的有效利用，每次運行最多清理100個過期元素。任務的啟動時間將根據上次實際清理數量自動調整，間隔時間趨于1秒到2小時之間。比如該次清理時刪除了100條元素，那么下次執行清理的時間將在1秒以后（最小間隔時間）。一旦該次清理數量少于上次清理數量，時間間隔將增加1.5倍。

RMapCache<String, SomeObject> map = redisson.getMapCache("anyMap");
// 有效時間 ttl = 10分鐘
map.put("key1", new SomeObject(), 10, TimeUnit.MINUTES);
// 有效時間 ttl = 10分鐘, 最長閑置時間 maxIdleTime = 10秒鐘
map.put("key1", new SomeObject(), 10, TimeUnit.MINUTES, 10, TimeUnit.SECONDS);

// 有效時間 = 3 秒鐘
map.putIfAbsent("key2", new SomeObject(), 3, TimeUnit.SECONDS);
// 有效時間 ttl = 40秒鐘, 最長閑置時間 maxIdleTime = 10秒鐘
map.putIfAbsent("key2", new SomeObject(), 40, TimeUnit.SECONDS, 10, TimeUnit.SECONDS);

1.2 本地緩存映射（LocalCachedMap）

在特定的場景下，高度頻繁的讀取操作讓網絡通信都被視為瓶頸的情況時，使用Redisson提供的分布式Map本地緩存RLocalCachedMapJava對象會是一個很好的選擇。RLocalCachedMap會根據配置參數，針對一個分布式Map中元素的進行有條件緩存，從而實現高效率的讀取操作。以下配置參數可以用來創建這個實例：

LocalCachedMapOptions options = LocalCachedMapOptions.defaults()
      // 淘汰機制有LFU, LRU和NONE這幾種算法策略可供選擇
     .evictionPolicy(EvictionPolicy.LFU)
     .cacheSize(1000)
      // 如果該值是`真(true)`時，在該實例執行更新和刪除操作的同時，將向其他所有的相同實例發
      // 送針對該元素的淘汰消息。其他相同實例在收到該消息以后，會同時刪除自身的緩存。下次讀取
      // 該元素時會從Redis服務器獲取。
     .invalidateEntryOnChange(false)
      // 每個Map本地緩存里元素的有效時間，默認毫秒為單位
     .timeToLive(10000)
      // 或者
     .timeToLive(10, TimeUnit.SECONDS)
      // 每個Map本地緩存里元素的最長閑置時間，默認毫秒為單位
     .maxIdle(10000)
      // 或者
     .maxIdle(10, TimeUnit.SECONDS);
RLocalCachedMap<String, Integer> map = redisson.getLocalCachedMap("test", options);

map.put("1", 1);
map.put("2", 2);

map.fastPut("3", 4);
當不再使用Map本地緩存對象的時候應該自行手動銷毀，如果Redisson對象被關閉（shutdown）了，則不用手動銷毀。

RLocalCachedMap<String, Integer> map = ...
map.destroy();

1.3. 映射（Map）數據分片

Map數據分片是Redis集群模式下的一個功能。Redisson提供的分布式RClusteredMap Java對象也是基于RMap實現的。在這里可以獲取更多的信息。

RClusteredMap<String, SomeObject> map = redisson.getClusteredMap("anyMap");

SomeObject prevObject = map.put("123", new SomeObject());
SomeObject currentObject = map.putIfAbsent("323", new SomeObject());
SomeObject obj = map.remove("123");

map.fastPut("321", new SomeObject());
map.fastRemove("321");

當使用RClusteredMap在特定的場景下，高度頻繁的讀取操作讓網絡通信都被視為瓶頸的情況時，使用Redisson提供的集群模式下的分布式Map本地緩存RClusteredLocalCachedMapJava對象會是一個很好的選擇。RClusteredLocalCachedMap會根據配置參數，針對一個集群模式下的分布式Map中元素的進行有條件緩存，從而實現高效率的讀取操作。以下配置參數可以用來創建這個實例：

LocalCachedMapOptions options = LocalCachedMapOptions.defaults()
      // LFU, LRU and NONE policies are available
     .evictionPolicy(EvictionPolicy.LFU)
     .cacheSize(1000)
      // 如果該值是`真(true)`時，在該實例執行更新和刪除操作的同時，將向其他所有的相同實例發
      // 送針對該元素的淘汰消息。其他相同實例在收到該消息以后，會同時刪除自身的緩存。下次讀取
      // 該元素時會從Redis服務器獲取。
     .invalidateEntryOnChange(false)
      // 每個Map本地緩存里元素的有效時間，默認毫秒為單位
     .timeToLive(10000)
      // 或者
     .timeToLive(10, TimeUnit.SECONDS)
      // 每個Map本地緩存里元素的最長閑置時間，默認毫秒為單位
     .maxIdle(10000)
      // 或者
     .maxIdle(10, TimeUnit.SECONDS);
RClusteredLocalCachedMap<String, Integer> map = redisson.getClusteredLocalCachedMap("test", options);
        
map.put("1", 1);
map.put("2", 2);

map.fastPut("3", 4);

當不再使用Map本地緩存對象的時候應該自行手動銷毀，如果Redisson對象被關閉（shutdown）了，則不用手動銷毀。

RClusteredLocalCachedMap<String, Integer> map = ...
map.destroy();

除了RClusteredLocalCachedMap以外，Redisson還提供了另一種集群模式下的分布式映射（Map），它不僅提供了透明的數據分片功能，還為每個元素提供了淘汰機制。RClusteredMapCache類分別同時提供了RClusteredMap和RMapCache這兩個接口的實現。與RClusteredLocalCachedMap不同的是，前者的淘汰機制是針對保持在Redis里的元素而言的，而后者的淘汰機制是針對的本地緩存里的元素而言。

RClusteredMapCache<String, SomeObject> map = redisson.getClusteredMapCache("anyMap");
// 有效時間 ttl = 10分鐘
map.put("key1", new SomeObject(), 10, TimeUnit.MINUTES);
// 有效時間 ttl = 10分鐘, 最長閑置時間 maxIdleTime = 10秒鐘
map.put("key1", new SomeObject(), 10, TimeUnit.MINUTES, 10, TimeUnit.SECONDS);

// 有效時間 = 3 秒鐘
map.putIfAbsent("key2", new SomeObject(), 3, TimeUnit.SECONDS);
// 有效時間 ttl = 40秒鐘, 最長閑置時間 maxIdleTime = 10秒鐘
map.putIfAbsent("key2", new SomeObject(), 40, TimeUnit.SECONDS, 10, TimeUnit.SECONDS);

該功能僅限于Redisson PRO版本。

2. 多值映射（Multimap）

Redisson的分布式RMultimap Java對象允許Map中的一個字段值包含多個元素。 字段總數受Redis限制，每個Multimap最多允許有4 294 967 295個不同字段。

2.1. 基于集（Set）的多值映射（Multimap）

基于Set的Multimap不允許一個字段值包含有重復的元素。

RSetMultimap<SimpleKey, SimpleValue> map = redisson.getSetMultimap("myMultimap");
map.put(new SimpleKey("0"), new SimpleValue("1"));
map.put(new SimpleKey("0"), new SimpleValue("2"));
map.put(new SimpleKey("3"), new SimpleValue("4"));

Set<SimpleValue> allValues = map.get(new SimpleKey("0"));

List<SimpleValue> newValues = Arrays.asList(new SimpleValue("7"), new SimpleValue("6"), new SimpleValue("5"));
Set<SimpleValue> oldValues = map.replaceValues(new SimpleKey("0"), newValues);

Set<SimpleValue> removedValues = map.removeAll(new SimpleKey("0"));

2.2. 基于列表（List）的多值映射（Multimap）

基于List的Multimap在保持插入順序的同時允許一個字段下包含重復的元素。

RListMultimap<SimpleKey, SimpleValue> map = redisson.getListMultimap("test1");
map.put(new SimpleKey("0"), new SimpleValue("1"));
map.put(new SimpleKey("0"), new SimpleValue("2"));
map.put(new SimpleKey("0"), new SimpleValue("1"));
map.put(new SimpleKey("3"), new SimpleValue("4"));

List<SimpleValue> allValues = map.get(new SimpleKey("0"));

Collection<SimpleValue> newValues = Arrays.asList(new SimpleValue("7"), new SimpleValue("6"), new SimpleValue("5"));
List<SimpleValue> oldValues = map.replaceValues(new SimpleKey("0"), newValues);

List<SimpleValue> removedValues = map.removeAll(new SimpleKey("0"));

2.3. 多值映射（Multimap）淘汰機制（Eviction）

Multimap對象的淘汰機制是通過不同的接口來實現的。它們是RSetMultimapCache接口和RListMultimapCache接口，分別對應的是Set和List的Multimaps。

所有過期元素都是通過org.redisson.EvictionScheduler實例來實現定期清理的。為了保證資源的有效利用，每次運行最多清理100個過期元素。任務的啟動時間將根據上次實際清理數量自動調整，間隔時間趨于1秒到2小時之間。比如該次清理時刪除了100條元素，那么下次執行清理的時間將在1秒以后（最小間隔時間）。一旦該次清理數量少于上次清理數量，時間間隔將增加1.5倍。

RSetMultimapCache的使用范例：

RSetMultimapCache<String, String> multimap = redisson.getSetMultimapCache("myMultimap");
multimap.put("1", "a");
multimap.put("1", "b");
multimap.put("1", "c");

multimap.put("2", "e");
multimap.put("2", "f");

multimap.expireKey("2", 10, TimeUnit.MINUTES);

3.集（Set）

Redisson的分布式Set結構的RSet Java對象實現了java.util.Set接口。通過元素的相互狀態比較保證了每個元素的唯一性。該對象的最大容量受Redis限制，最大元素數量是4 294 967 295個。

RSet<SomeObject> set = redisson.getSet("anySet");
set.add(new SomeObject());
set.remove(new SomeObject());

Redisson PRO版本中的Set對象還可以在Redis集群環境下支持單集合數據分片。

3.1. 集（Set）淘汰機制（Eviction）

Redisson的分布式RSetCache Java對象在基于RSet的前提下實現了針對單個元素的淘汰機制。由于RSetCache是基于RSet實現的，使它還集成了java.util.Set接口。

目前的Redis自身并不支持Set當中的元素淘汰，因此所有過期元素都是通過org.redisson.EvictionScheduler實例來實現定期清理的。為了保證資源的有效利用，每次運行最多清理100個過期元素。任務的啟動時間將根據上次實際清理數量自動調整，間隔時間趨于1秒到2小時之間。比如該次清理時刪除了100條元素，那么下次執行清理的時間將在1秒以后（最小間隔時間）。一旦該次清理數量少于上次清理數量，時間間隔將增加1.5倍。

RSetCache<SomeObject> set = redisson.getSetCache("anySet");
// ttl = 10 seconds
set.add(new SomeObject(), 10, TimeUnit.SECONDS);

3.2. 集（Set）數據分片

Set數據分片是Redis集群模式下的一個功能。Redisson提供的分布式RClusteredSet Java對象也是基于RSet實現的。在這里可以獲取更多的信息。

RClusteredSet<SomeObject> set = redisson.getClusteredSet("anySet");
set.add(new SomeObject());
set.remove(new SomeObject());

除了RClusteredSet以外，Redisson還提供了另一種集群模式下的分布式集（Set），它不僅提供了透明的數據分片功能，還為每個元素提供了淘汰機制。RClusteredSetCache類分別同時提供了RClusteredSet和RSetCache這兩個接口的實現。當然這些都是基于java.util.Set的接口實現上的。

該功能僅限于Redisson PRO版本。

4. 有序集（SortedSet）

Redisson的分布式RSortedSet Java對象實現了java.util.SortedSet接口。在保證元素唯一性的前提下，通過比較器（Comparator）接口實現了對元素的排序。

RSortedSet<Integer> set = redisson.getSortedSet("anySet");
set.trySetComparator(new MyComparator()); // 配置元素比較器
set.add(3);
set.add(1);
set.add(2);

set.removeAsync(0);
set.addAsync(5);

5. 計分排序集（ScoredSortedSet）

Redisson的分布式RScoredSortedSet Java對象是一個可以按插入時指定的元素評分排序的集合。它同時還保證了元素的唯一性。

RScoredSortedSet<SomeObject> set = redisson.getScoredSortedSet("simple");

set.add(0.13, new SomeObject(a, b));
set.addAsync(0.251, new SomeObject(c, d));
set.add(0.302, new SomeObject(g, d));

set.pollFirst();
set.pollLast();

int index = set.rank(new SomeObject(g, d)); // 獲取元素在集合中的位置
Double score = set.getScore(new SomeObject(g, d)); // 獲取元素的評分

6. 字典排序集（LexSortedSet）

Redisson的分布式RLexSortedSet Java對象在實現了java.util.Set<String>接口的同時，將其中的所有字符串元素按照字典順序排列。它公式還保證了字符串元素的唯一性。

RLexSortedSet set = redisson.getLexSortedSet("simple");
set.add("d");
set.addAsync("e");
set.add("f");

set.lexRangeTail("d", false);
set.lexCountHead("e");
set.lexRange("d", true, "z", false);

7. 列表（List)

Redisson分布式列表（List）結構的RList Java對象在實現了java.util.List接口的同時，確保了元素插入時的順序。該對象的最大容量受Redis限制，最大元素數量是4 294 967 295個。

RList<SomeObject> list = redisson.getList("anyList");
list.add(new SomeObject());
list.get(0);
list.remove(new SomeObject());

8. 列隊（Queue）

Redisson分布式無界列隊（Queue）結構的RQueue Java對象實現了java.util.Queue接口。盡管RQueue對象無初始大小（邊界）限制，但對象的最大容量受Redis限制，最大元素數量是4 294 967 295個。

RQueue<SomeObject> queue = redisson.getQueue("anyQueue");
queue.add(new SomeObject());
SomeObject obj = queue.peek();
SomeObject someObj = queue.poll();

9. 雙端隊列（Deque）

Redisson分布式無界雙端隊列（Deque）結構的RDeque Java對象實現了java.util.Deque接口。盡管RDeque對象無初始大小（邊界）限制，但對象的最大容量受Redis限制，最大元素數量是4 294 967 295個。

RDeque<SomeObject> queue = redisson.getDeque("anyDeque");
queue.addFirst(new SomeObject());
queue.addLast(new SomeObject());
SomeObject obj = queue.removeFirst();
SomeObject someObj = queue.removeLast();

10. 阻塞隊列（Blocking Queue）

Redisson分布式無界阻塞隊列（Blocking Queue）結構的RBlockingQueue Java對象實現了java.util.concurrent.BlockingQueue接口。盡管RBlockingQueue對象無初始大小（邊界）限制，但對象的最大容量受Redis限制，最大元素數量是4 294 967 295個。

RBlockingQueue<SomeObject> queue = redisson.getBlockingQueue("anyQueue");
queue.offer(new SomeObject());

SomeObject obj = queue.peek();
SomeObject someObj = queue.poll();
SomeObject ob = queue.poll(10, TimeUnit.MINUTES);

poll, pollFromAny, pollLastAndOfferFirstTo和take方法內部采用話題訂閱發布實現，在Redis節點故障轉移（主從切換）或斷線重連以后，內置的相關話題監聽器將自動完成話題的重新訂閱。

11. 有界阻塞列隊（Bounded Blocking Queue）

Redisson分布式有界阻塞列隊（Bounded Blocking Queue）結構的RBoundedBlockingQueue Java對象實現了java.util.concurrent.BlockingQueue接口。該對象的最大容量受Redis限制，最大元素數量是4 294 967 295個。列隊的初始容量（邊界）必須在使用前設定好。

RBoundedBlockingQueue<SomeObject> queue = redisson.getBoundedBlockingQueue("anyQueue");
// 如果初始容量（邊界）設定成功則返回`真（true）`，
// 如果初始容量（邊界）已近存在則返回`假（false）`。
queue.trySetCapacity(2);

queue.offer(new SomeObject(1));
queue.offer(new SomeObject(2));
// 此時容量已滿，下面代碼將會被阻塞，直到有空閑為止。
queue.put(new SomeObject());

SomeObject obj = queue.peek();
SomeObject someObj = queue.poll();
SomeObject ob = queue.poll(10, TimeUnit.MINUTES);

poll, pollFromAny, pollLastAndOfferFirstTo和take方法內部采用話題訂閱發布實現，在Redis節點故障轉移（主從切換）或斷線重連以后，內置的相關話題監聽器將自動完成話題的重新訂閱。

12. 阻塞雙端列隊（Blocking Deque）

Redisson分布式無界阻塞雙端列隊（Blocking Deque）結構的RBlockingDeque Java對象實現了java.util.concurrent.BlockingDeque接口。盡管RBlockingDeque對象無初始大小（邊界）限制，但對象的最大容量受Redis限制，最大元素數量是4 294 967 295個。

RBlockingDeque<Integer> deque = redisson.getBlockingDeque("anyDeque");
deque.putFirst(1);
deque.putLast(2);
Integer firstValue = queue.takeFirst();
Integer lastValue = queue.takeLast();
Integer firstValue = queue.pollFirst(10, TimeUnit.MINUTES);
Integer lastValue = queue.pollLast(3, TimeUnit.MINUTES);

poll, pollFromAny, pollLastAndOfferFirstTo和take方法內部采用話題訂閱發布實現，在Redis節點故障轉移（主從切換）或斷線重連以后，內置的相關話題監聽器將自動完成話題的重新訂閱。

13. 阻塞公平列隊（Blocking Fair Queue）

Redisson分布式無界阻塞公平列隊（Blocking Fair Queue）結構的RBlockingFairQueue Java對象在實現Redisson分布式無界阻塞隊列（Blocking Queue）結構RBlockingQueue接口的基礎上，解決了多個列隊消息的處理者在復雜的網絡環境下，網絡延時的影響使“較遠”的客戶端最終收到消息數量低于“較近”的客戶端的問題。從而解決了這種現象引發的個別處理節點過載的情況。

以分布式無界阻塞隊列為基礎，采用公平獲取消息的機制，不僅保證了poll和take方法獲取消息的先入順序，還能讓列隊里的消息被均勻的發布到處在復雜分布式環境中的各個處理節點里。

RBlockingFairQueue queue = redisson.getBlockingFairQueue("myQueue");
queue.offer(new SomeObject());

SomeObject obj = queue.peek();
SomeObject someObj = queue.poll();
SomeObject ob = queue.poll(10, TimeUnit.MINUTES);

14. 延遲列隊（Delayed Queue）

Redisson分布式延遲列隊（Delayed Queue）結構的RDelayedQueue Java對象在實現了RQueue接口的基礎上提供了向列隊按要求延遲添加項目的功能。該功能可以用來實現消息傳送延遲按幾何增長或幾何衰減的發送策略。

RQueue<String> distinationQueue = ...
RDelayedQueue<String> delayedQueue = getDelayedQueue(distinationQueue);
// 10秒鐘以后將消息發送到指定列隊
delayedQueue.offer("msg1", 10, TimeUnit.SECONDS);
// 一分鐘以后將消息發送到指定列隊
delayedQueue.offer("msg2", 1, TimeUnit.MINUTES);
在該對象不再需要的情況下，應該主動銷毀。僅在相關的Redisson對象也需要關閉的時候可以不用主動銷毀。

RDelayedQueue<String> delayedQueue = ...
delayedQueue.destroy();

15. 優先隊列（Priority Queue）

基于Redis的分布式優先隊列（Priority Queue）Java對象實現了java.util.Queue的接口。可以通過比較器（Comparator）接口來對元素排序。

RPriorityQueue<Integer> queue = redisson.getPriorityQueue("anyQueue");
queue.trySetComparator(new MyComparator()); // 指定對象比較器
queue.add(3);
queue.add(1);
queue.add(2);

queue.removeAsync(0);
queue.addAsync(5);

queue.poll();

16. 優先雙端隊列（Priority Deque）

基于Redis的分布式優先雙端隊列（Priority Deque）Java對象實現了java.util.Deque的接口。可以通過比較器（Comparator）接口來對元素排序。

RPriorityDeque<Integer> queue = redisson.getPriorityDeque("anyQueue");
queue.trySetComparator(new MyComparator()); // 指定對象比較器
queue.addLast(3);
queue.addFirst(1);
queue.add(2);

queue.removeAsync(0);
queue.addAsync(5);

queue.pollFirst();
queue.pollLast();

關于redisson中如何使用分布式集合就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。