[TOC] # redis-spring-boot-starter 前面項目中，咱們使用了[db-core]([http://www.hmoore.net/owenwangwen/open-capacity-platform/1048264](http://www.hmoore.net/owenwangwen/open-capacity-platform/1048264))為整個項目提供通用的數據庫處理， db-core提供了druid數據源，mybatis的操作，也支持了redis的操作，功能不單一，將代碼分離，自定義redis-spring-boot-starter ## 功能 * lettuce連接池 * redis操作工具類 * geohash 經緯度定位 * redisson分布式鎖 ## 代碼分析 * 代碼一覽  * RedisAutoConfig  回顧db-spring-boot-starter的章節 * @EnableAutoConfiguration 作用：從classpath中搜索所有META-INF/spring.factories配置文件 并且其中org.springframework.boot.autoconfigure.EnableAutoConfiguration key對應的配置項加載到spring容器 ## 序列化 ### redis序列化配置  ### oauth序列化配置  ### 測試腳本 原生序列化方案[序列化100000次]耗時：14389ms size:=1276000000 原生序列化方案[反序列化100000次]耗時：16298ms size:=1276000000 hessian序列化方案[序列化100000次]耗時：10479ms size:=792000000 hessian序列化方案[反序列化100000次]耗時：5480ms size:=792000000 Kryo序列化方案[序列化100000次]耗時：9330ms size:=1013000000 Kryo序列化方案[反序列化100000次]耗時：8582ms size:=1013000000 ``` public static void main(String[] args) { Map map = new HashMap(); for(int i = 0 ; i <=100 ; i++) { map.put("a"+i, "a"+i); } Serializer jdkSerializer = SerializerManager.getSerializer(SerializerManager.JDK) ; Serializer kryoObjectSerializer = SerializerManager.getSerializer(SerializerManager.KRYO) ; Serializer hessianSerializer = SerializerManager.getSerializer(SerializerManager.HESSIAN2) ; long size = 0; long time1 = System.currentTimeMillis(); byte[] jdkserialize = null ; byte[] redisserialize = null ; byte[] kryoserialize = null ; for (int i = 0; i < 1000000; i++) { jdkserialize =jdkSerializer.serialize(map); size += jdkserialize.length; } System.out.println("原生序列化方案[序列化100000次]耗時：" + (System.currentTimeMillis() - time1) + "ms size:=" + size); long time2 = System.currentTimeMillis(); for (int i = 0; i < 1000000; i++) { Map aa =(Map) jdkSerializer.deserialize(jdkserialize) ; } System.out.println("原生序列化方案[反序列化100000次]耗時：" + (System.currentTimeMillis() - time2) + "ms size:=" + size); long time3 = System.currentTimeMillis(); size = 0; for (int i = 0; i < 1000000; i++) { redisserialize =hessianSerializer.serialize(map); size += redisserialize.length; } System.out.println("hessian序列化方案[序列化100000次]耗時：" + (System.currentTimeMillis() - time3) + "ms size:=" + size); long time4 = System.currentTimeMillis(); for (int i = 0; i < 1000000; i++) { Map aa =(Map) hessianSerializer.deserialize(redisserialize) ; } System.out.println("hessian序列化方案[反序列化100000次]耗時：" + (System.currentTimeMillis() - time4) + "ms size:=" + size); long time5 = System.currentTimeMillis(); size = 0; for (int i = 0; i < 1000000; i++) { kryoserialize =kryoObjectSerializer.serialize(map); size += kryoserialize.length; } System.out.println("Kryo序列化方案[序列化100000次]耗時：" + (System.currentTimeMillis() - time5) + "ms size:=" + size); long time6 = System.currentTimeMillis(); for (int i = 0; i < 1000000; i++) { Map aa =(Map) kryoObjectSerializer.deserialize(kryoserialize) ; } System.out.println("Kryo序列化方案[反序列化100000次]耗時：" + (System.currentTimeMillis() - time6) + "ms size:=" + size); } ``` ## redis cluster中的代碼優化 項目中 3主3從 redis集群出現單節點宕機，造成master遷移，但是發現應用無法正常連接redis ，分析了代碼，發現默認Lettuce是不會刷新拓撲io.lettuce.core.cluster.models.partitions.Partitions#slotCache，最終造成槽點查找節點依舊找到老的節點，自然訪問不了了。  ## geohash 經緯度定位 ~~~ /** * 添加經緯度信息 map.put("北京" ,new Point(116.405285 ,39.904989)) //redis 命令：geoadd * cityGeo 116.405285 39.904989 "北京" */ public Long addGeoPoint(String key, Map<Object, Point> map) { return redisTemplate.opsForGeo().add(key, map); } /** * 查找指定key的經緯度信息 redis命令：geopos cityGeo 北京 * * @param key * @param member * @return */ public Point geoGetPoint(String key, String member) { List<Point> lists = redisTemplate.opsForGeo().position(key, member); return lists.get(0); } /** * 返回兩個地方的距離，可以指定單位 redis命令：geodist cityGeo 北京 上海 * * @param key * @param srcMember * @param targetMember * @return */ public Distance geoDistance(String key, String srcMember, String targetMember) { Distance distance = redisTemplate.opsForGeo().distance(key, srcMember, targetMember, Metrics.KILOMETERS); return distance; } /** * 根據指定的地點查詢半徑在指定范圍內的位置 redis命令：georadiusbymember cityGeo 北京 100 km WITHDIST * WITHCOORD ASC COUNT 5 * * @param key * @param member * @param distance * @return */ public GeoResults geoRadiusByMember(String key, String member, double distance) { return redisTemplate.opsForGeo().radius(key, member, new Distance(distance, Metrics.KILOMETERS), RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs().includeDistance().includeCoordinates() .sortAscending()); } /** * 根據給定的經緯度，返回半徑不超過指定距離的元素 redis命令：georadius cityGeo 116.405285 39.904989 * 100 km WITHDIST WITHCOORD ASC COUNT 5 * * @param key * @param circle * @param distance * @return */ public GeoResults geoRadiusByCircle(String key, Circle circle, double distance) { return redisTemplate.opsForGeo().radius(key, circle, new Distance(distance, Metrics.KILOMETERS), RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs().includeDistance().includeCoordinates() .sortAscending()); } //刪除元素 public Long deleteGeoMember(String key, String member) { return redisTemplate.boundZSetOps(key).remove(member); } ~~~ ## 分布式鎖redssion ### 集成方式  ### 大致使用  ### 代碼分析 * 獲取鎖  > 調用getLock()方法后實際返回一個RedissonLock對象 * 加鎖   > 在RedissonLock對象的lock()方法主要調用tryAcquire()方法，由于leaseTime == -1，于是走tryLockInnerAsync()方法， * 加鎖細節  > 結合上面的參數聲明，我們可以知道，這里KEYS\[1\]就是getName()，ARGV\[2\]是getLockName(threadId),假設前面獲取鎖時傳的name是“anyLock”，假設調用的線程ID是Thread-1，假設成員變量UUID類型的id是85b196ce-e6f2-42ff-b3d7-6615b6748b5d:65那么KEYS\[1\]=anyLock，ARGV\[2\]=85b196ce-e6f2-42ff-b3d7-6615b6748b5d:Thread-1 ,因此，這段腳本的意思是1、判斷有沒有一個叫“anyLock”的key2、如果沒有，則在其下設置一個字段為“85b196ce-e6f2-42ff-b3d7-6615b6748b5d:Thread-1”，值為“1”的鍵值對 ，并設置它的過期時間3、如果存在，則進一步判斷“85b196ce-e6f2-42ff-b3d7-6615b6748b5d:Thread-1”是否存在，若存在，則其值加1，并重新設置過期時間4、返回“anyLock”的生存時間（毫秒） * 加鎖redis結構  > 這里用的數據結構是hash，hash的結構是： key? 字段1? 值1 字段2? 值2? 。。。用在鎖這個場景下，key就表示鎖的名稱，也可以理解為臨界資源，字段就表示當前獲得鎖的線程所有競爭這把鎖的線程都要判斷在這個key下有沒有自己線程的字段，如果沒有則不能獲得鎖，如果有，則相當于重入，字段值加1（次數） * 解鎖  > 我們還是假設name=anyLock，假設線程ID是Thread-1，同理，我們可以知道KEYS\[1\]是getName()，即KEYS\[1\]=anyLock，KEYS\[2\]是getChannelName()，即KEYS\[2\]=redisson\_lock\_\_channel:{anyLock}，ARGV\[1\]是LockPubSub.unlockMessage，即ARGV\[1\]=0，ARGV\[2\]是生存時間，ARGV\[3\]是getLockName(threadId)，即ARGV\[3\]=85b196ce-e6f2-42ff-b3d7-6615b6748b5d:Thread-1，因此，上面腳本的意思是：1、判斷是否存在一個叫“anyLock”的key2、如果不存在，向Channel中廣播一條消息，廣播的內容是0，并返回1。3、如果存在，進一步判斷字段85b196ce-e6f2-42ff-b3d7-6615b6748b5d:Thread-1是否存在。4、若字段不存在，返回空，若字段存在，則字段值減1，5、若減完以后，字段值仍大于0，則返回0。6、減完后，若字段值小于或等于0，則廣播一條消息，廣播內容是0，并返回1；可以猜測，廣播0表示資源可用，即通知那些等待獲取鎖的線程現在可以獲得鎖了 * 等待 ``` private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException { long threadId = Thread.currentThread().getId(); Long ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired if (ttl == null) { return; } RFuture<RedissonLockEntry> future = subscribe(threadId); if (interruptibly) { commandExecutor.syncSubscriptionInterrupted(future); } else { commandExecutor.syncSubscription(future); } try { while (true) { ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired if (ttl == null) { break; } // waiting for message if (ttl >= 0) { try { future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { if (interruptibly) { throw e; } future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } } else { if (interruptibly) { future.getNow().getLatch().acquire(); } else { future.getNow().getLatch().acquireUninterruptibly(); } } } } finally { unsubscribe(future, threadId); } // get(lockAsync(leaseTime, unit)); } ``` > 這里會訂閱Channel，當資源可用時可以及時知道，并搶占，防止無效的輪詢而浪費資源當資源可用用的時候，循環去嘗試獲取鎖，由于多個線程同時去競爭資源，所以這里用了信號量，對于同一個資源只允許一個線程獲得鎖，其它的線程阻塞 * 總結   ## Cache Aside Pattern * 讀的時候，先讀緩存，緩存沒有的話，那么就讀數據庫，然后取出數據后放入緩存，同時返回響應  * 更新的時候，先刪除緩存，然后再更新數據庫