《三次給你聊清楚Redis》之說說簡單的應用

《三天給你聊清楚redis》第3天說說redis大概怎么用，和面試題（18000字）

五、實戰

5.1基礎實戰

5.1.1實戰

功能隨處可見，我們都知道是一個非常高頻的操作，redis就非常適合做這種工作。

實現效果：

分析：三種類型：給帖子，給評論，給回復

我們只實現查看數量的話，只要一個int記錄一下就可以，但是我們之后還想查看的人，所以要把每一個的信息都記錄好，方便后面的功能繼續做出來。

思路：

：把的信息放進去。

取消：把的信息刪除。

在此之前，我們要封裝一個get到key的類，方便后面復用。

package com.now.community.community.util; public class RedisKeyUtil { private static final String SPLIT = ":"; private static final String PREFIX_ENTITY_LIKE = "like:entity"; private static final String PREFIX_USER_LIKE = "like:user"; // 某個實體的贊 // like:entity:entityType:entityId -> set(userId) public static String getEntityLikeKey(int entityType, int entityId) { return PREFIX_ENTITY_LIKE + SPLIT + entityType + SPLIT + entityId; } // 某個用戶的贊 // like:user:userId -> int public static String getUserLikeKey(int userId) { return PREFIX_USER_LIKE + SPLIT + userId; } }

業務：

// public void like(int userId, int entityType, int entityId, int entityUserId) { redisTemplate.execute(new SessionCallback() { @Override public Object execute(RedisOperations operations) throws DataAccessException { String entityLikeKey = RedisKeyUtil.getEntityLikeKey(entityType, entityId); boolean isMember = operations.opsForSet().isMember(entityLikeKey, userId); operations.multi(); if (isMember) { operations.opsForSet().remove(entityLikeKey, userId); } else { operations.opsForSet().add(entityLikeKey, userId); } return operations.exec(); } }); }

我們要查找是否，還有數量，方便頁面顯示：

// 查詢某實體的數量 public long findEntityLikeCount(int entityType, int entityId) { String entityLikeKey = RedisKeyUtil.getEntityLikeKey(entityType, entityId); return redisTemplate.opsForSet().size(entityLikeKey); } // 查詢某人對某實體的狀態 public int findEntityLikeStatus(int userId, int entityType, int entityId) { String entityLikeKey = RedisKeyUtil.getEntityLikeKey(entityType, entityId); return redisTemplate.opsForSet().isMember(entityLikeKey, userId) ? 1 : 0; }

LikeController

@RequestMapping(path = "/like", method = RequestMethod.POST) @ResponseBody public String like(int entityType, int entityId,int entityUserId,int postId) { User user = hostHolder.getUser(); // likeService.like(user.getId(), entityType, entityId,entityUserId); // 數量 long likeCount = likeService.findEntityLikeCount(entityType, entityId); // 狀態 int likeStatus = likeService.findEntityLikeStatus(user.getId(), entityType, entityId); // 返回的結果 Map map = new HashMap<>(); map.put("likeCount", likeCount); map.put("likeStatus", likeStatus); return CommunityUtil.getJSONString(0, null, map); }

5.1.2實戰關注

效果：

思路：很好想，把自己的粉絲和自己關注的人都存起來（set即可），做增刪改查。

package com.now.Community.community.service; import com.now.community.community.entity.User; import com.now.community.community.util.CommunityConstant; import com.now.community.community.util.RedisKeyUtil; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.dao.DataAccessException; import org.springframework.data.redis.core.RedisOperations; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.SessionCallback; import org.springframework.stereotype.Service; import java.util.*; @Service public class FollowService implements CommunityConstant { @Autowired private RedisTemplate redisTemplate; @Autowired private UserService userService; public void follow(int userId, int entityType, int entityId) { redisTemplate.execute(new SessionCallback() { @Override public Object execute(RedisOperations operations) throws DataAccessException { String followeeKey = RedisKeyUtil.getFolloweeKey(userId, entityType); String followerKey = RedisKeyUtil.getFollowerKey(entityType, entityId); operations.multi(); operations.opsForZSet().add(followeeKey, entityId, System.currentTimeMillis()); operations.opsForZSet().add(followerKey, userId, System.currentTimeMillis()); return operations.exec(); } }); } public void unfollow(int userId, int entityType, int entityId) { redisTemplate.execute(new SessionCallback() { @Override public Object execute(RedisOperations operations) throws DataAccessException { String followeeKey = RedisKeyUtil.getFolloweeKey(userId, entityType); String followerKey = RedisKeyUtil.getFollowerKey(entityType, entityId); operations.multi(); operations.opsForZSet().remove(followeeKey, entityId); operations.opsForZSet().remove(followerKey, userId); return operations.exec(); } }); } // 查詢關注的實體的數量 public long findFolloweeCount(int userId, int entityType) { String followeeKey = RedisKeyUtil.getFolloweeKey(userId, entityType); return redisTemplate.opsForZSet().zCard(followeeKey); } // 查詢實體的粉絲的數量 public long findFollowerCount(int entityType, int entityId) { String followerKey = RedisKeyUtil.getFollowerKey(entityType, entityId); return redisTemplate.opsForZSet().zCard(followerKey); } // 查詢當前用戶是否已關注該實體 public boolean hasFollowed(int userId, int entityType, int entityId) { String followeeKey = RedisKeyUtil.getFolloweeKey(userId, entityType); return redisTemplate.opsForZSet().score(followeeKey, entityId) != null; } // 查詢某用戶關注的人 public List> findFollowees(int userId, int offset, int limit) { String followeeKey = RedisKeyUtil.getFolloweeKey(userId, ENTITY_TYPE_USER); Set targetIds = redisTemplate.opsForZSet().reverseRange(followeeKey, offset, offset + limit - 1); if (targetIds == null) { return null; } List> list = new ArrayList<>(); for (Integer targetId : targetIds) { Map map = new HashMap<>(); User user = userService.findUserById(targetId); map.put("user", user); Double score = redisTemplate.opsForZSet().score(followeeKey, targetId); map.put("followTime", new Date(score.longValue())); list.add(map); } return list; } // 查詢某用戶的粉絲 public List> findFollowers(int userId, int offset, int limit) { String followerKey = RedisKeyUtil.getFollowerKey(ENTITY_TYPE_USER, userId); Set targetIds = redisTemplate.opsForZSet().reverseRange(followerKey, offset, offset + limit - 1); if (targetIds == null) { return null; } List> list = new ArrayList<>(); for (Integer targetId : targetIds) { Map map = new HashMap<>(); User user = userService.findUserById(targetId); map.put("user", user); Double score = redisTemplate.opsForZSet().score(followerKey, targetId); map.put("followTime", new Date(score.longValue())); list.add(map); } return list; } }

5.1.3實戰統計訪問量

過于簡單不解釋

5.1.4實戰排行榜

const REDIS_TB_NAME='user:actId'; //表名 const REDIS_SEP=":"; //命名分隔符 const REDIS_FIELDS="username|regtime"; //表字段名稱 const REDIS_FIELD_RANK="rank"; //排行 const REDIS_FIELD_ID="id"; //表的自增ID //插入排行榜數據 for($i=0;$iget(REDIS_TB_NAME.REDIS_SEP.REDIS_FIELD_ID); //事務機制,插入用戶信息及排行信息，自增id $ret=$redis->multi() //開始事務 ->hMset(REDIS_TB_NAME.REDIS_SEP.$redis_increase_id,array($fields[0]=>"username".$redis_increase_id, $fields[1]=>(time()+intval(rand(0,1000))))) //username 用戶名 //regtime 注冊時間 ->Zadd(REDIS_TB_NAME.REDIS_SEP.REDIS_FIELD_RANK,intval(rand(1,100)),$redis_increase_id) //插入排行 ->incr(REDIS_TB_NAME.REDIS_SEP.REDIS_FIELD_ID) //自增id ->exec(); //執行事務 if($ret==false) //插入失敗，重新插入 { $i--; } } echo "插入".$i."條記錄成功
";

Redis本身支持一些簡單的組合型的命令，比如以NX結尾命令都是判斷在這個值沒有時才進行某個命令

????????Redis支持自定義的命令組合，通過MULTI和EXEC，將幾個命令組合起來執行

????????如：插入排行數據和用戶信息，并自增id

$redis->multi() ->hmset("user:1",array("username"=>"hirryli","regtime"=>1234123483)) ->Zadd("user:rank",$scores,$userId) ->incr("user:id") ->exec();

5.2實戰優化小項目

這是我們之前項目的業務流程，做一下簡單介紹。

登錄：

用戶輸入賬號、密碼、驗證碼。我們先判斷用戶輸入的驗證碼是不是我們session存的驗證碼，然后去查賬號密碼是否正確。

如果登錄成功，發送給用戶一張憑證（ticket）。

登錄后

之后的每次請求，用戶攜帶ticket，服務器得到后，根據ticket去login_ticket表中查找登錄信息，并且根據登錄信息再查user表獲得更多的用戶信息。

使用Redis存儲驗證碼

- 驗證碼需要頻繁的訪問與刷新，對性能要求較高。

- 驗證碼不需永久保存，通常在很短的時間后就會失效。

- 分布式部署時，存在Session共享的問題。

我們重構思路：進入登錄頁面會訪問驗證碼方法，此方法會自動生成一個驗證碼和圖片，將驗證碼和圖片輸出給瀏覽器，并且下發一個cookies，這個cookies里面存的是一段隨機數，這段隨機數作為key存在redis里面（之前是存session），value就是驗證碼，并設置一個過期時間；

//驗證碼 @RequestMapping(path = "/kaptcha", method = RequestMethod.GET) public void getKaptcha(HttpServletResponse response/*, HttpSession session*/) { // 生成驗證碼 String text = kaptchaProducer.createText(); BufferedImage image = kaptchaProducer.createImage(text); // 將驗證碼存入session //session.setAttribute("kaptcha", text); //驗證碼的歸屬 String owner= CommunityUtil.generateUUID(); Cookie cookie=new Cookie("kaptchaOwner",owner); cookie.setMaxAge(60); cookie.setPath(contextPath); response.addCookie(cookie); //存入redis String redisKey= RedisKeyUtil.getKaptchaKey(owner); redisTemplate.opsForValue().set(redisKey,text,60, TimeUnit.SECONDS); // 將圖片輸出給瀏覽器 response.setContentType("image/png"); try { OutputStream os = response.getOutputStream(); ImageIO.write(image, "png", os); } catch (IOException e) { logger.error("響應驗證碼失敗:" + e.getMessage()); } }

@RequestMapping(path = "/login",method = RequestMethod.POST) public String login(String username,String password,String code,boolean rememberme, Model model,/*HttpSession session,*/HttpServletResponse response, @CookieValue("kaptchaOwner") String kaptchaOwner){ // 檢查驗證碼 //String kaptcha = (String) session.getAttribute("kaptcha"); String kaptcha=null; if(StringUtils.isNotBlank(kaptchaOwner)){ String redisKey=RedisKeyUtil.getKaptchaKey(kaptchaOwner); kaptcha=(String) redisTemplate.opsForValue().get(redisKey); } if(StringUtils.isBlank(kaptcha) || StringUtils.isBlank(code) || !kaptcha.equalsIgnoreCase(code)){ model.addAttribute("codeMsg", "驗證碼不正確!"); return "/site/login"; } // 檢查賬號,密碼 int expiredSeconds = rememberme ? REMEMBER_EXPIRED_SECONDS : DEFAULT_EXPIRED_SECONDS; Map map = userService.login(username, password, expiredSeconds); if (map.containsKey("ticket")) { Cookie cookie = new Cookie("ticket", map.get("ticket").toString()); cookie.setPath(contextPath); cookie.setMaxAge(expiredSeconds); response.addCookie(cookie); return "redirect:/index"; } else { ... } }

使用Redis存儲登錄憑證

- 處理每次請求時，都要查詢用戶的登錄憑證，訪問的頻率非常高。

登錄時不存MySQL里，存redis里

public Map login(String username,String password,int expiredSeconds){ Map map=new HashMap<>(); // 生成登錄憑證 LoginTicket loginTicket = new LoginTicket(); loginTicket.setUserId(user.getId()); loginTicket.setTicket(CommunityUtil.generateUUID()); loginTicket.setStatus(0); loginTicket.setExpired(new Date(System.currentTimeMillis() + expiredSeconds * 1000)); //loginTicketMapper.insertLoginTicket(loginTicket); String redisKey= RedisKeyUtil.getTicketKey(loginTicket.getTicket()); redisTemplate.opsForValue().set(redisKey,loginTicket); ... }

查找

退出時也是改redis

public void logout(String ticket) { //loginTicketMapper.updateStatus(ticket, 1); String redisKey= RedisKeyUtil.getTicketKey(ticket); LoginTicket loginTicket=(LoginTicket) redisTemplate.opsForValue().get(redisKey); loginTicket.setStatus(1); redisTemplate.opsForValue().set(redisKey,loginTicket); }

使用redis緩存用戶信息

- 處理每次請求時，都要根據憑證查詢用戶信息，訪問的頻率非常高。

緩存用戶信息：因為會經常根據userid來查詢user對象，所以使用redis來緩存提高服務器性能。使用redis的String類型，存入user對象，會自動將整個對象轉換成json字符串，同時設置過期時間；

取值：優先從redis中取，取不到的時候從mysql中取，并將數據初始化到redis中

更新：更新的時候先更新mysql中的值，然后清除緩存數據；

// 1.優先從緩存中取值 private User getCache(int userId) { String redisKey = RedisKeyUtil.getUserKey(userId); return (User) redisTemplate.opsForValue().get(redisKey); } // 2.取不到時初始化緩存數據 private User initCache(int userId) { User user = userMapper.selectById(userId); String redisKey = RedisKeyUtil.getUserKey(userId); redisTemplate.opsForValue().set(redisKey, user, 3600, TimeUnit.SECONDS); return user; } // 3.數據變更時清除緩存數據 private void clearCache(int userId) { String redisKey = RedisKeyUtil.getUserKey(userId); redisTemplate.delete(redisKey); }

public User findUserById(int id) { // return userMapper.selectById(id); User user = getCache(id); if (user == null) { user = initCache(id); } return user; }

public int updateHeader(int userId, String headerUrl) { //return userMapper.updateHeader(userId, headerUrl); int rows=userMapper.updateHeader(userId, headerUrl); clearCache(userId); return rows; }

5.3討論一下為啥用redis解決會話？

什么是會話？

會話可簡單理解為：用戶開一個瀏覽器，點擊多個超鏈接，訪問服務器多個web資源，然后關閉瀏覽器，整個過程稱之為一個會話。

?會話過程中要解決的一些問題？

–每個用戶不可避免各自會產生一些數據，程序要想辦法為每個用戶保存這些數據。

–例如：用戶點擊超鏈接通過一個servlet購買了一個商品，程序應該想辦法保存用戶購買的商品，以便于用戶點結帳servlet時，結帳servlet可以得到用戶購買的商品為用戶結帳。

?Cookie

–Cookie是客戶端技術，程序把每個用戶的數據以cookie的形式寫給用戶各自的瀏覽器。當用戶使用瀏覽器再去訪問服務器中的web資源時，就會帶著各自的數據去。這樣，web資源處理的就是用戶各自的數據了。

?HttpSession

–Session是服務器端技術，利用這個技術，服務器在運行時可以為每一個用戶的瀏覽器創建一個其獨享的HttpSession對象，由于session為用戶瀏覽器獨享，所以用戶在訪問服務器的web資源時，可以把各自的數據放在各自的session中，當用戶再去訪問服務器中的其它web資源時，其它web資源再從用戶各自的session中取出數據為用戶服務。

總結：cookie存在客戶端，session存在服務器端

通常結合使用。

我們先用sprintboot演示一下cookie和session操作

@RequestMapping(path = "/cookie/set",method = RequestMethod.GET) @ResponseBody public String setCookie(HttpServletResponse httpServletResponse){ Cookie cookie=new Cookie("code", CommunityUtil.generateUUID()); cookie.setPath("/community/alpha"); cookie.setMaxAge(60*10); httpServletResponse.addCookie(cookie); return "set cookie"; } @RequestMapping(path = "/cookie/get",method = RequestMethod.GET) @ResponseBody public String getCookie(@CookieValue("code") String code){ System.out.println(code); return "get cookie"; } @RequestMapping(path = "/session/set", method = RequestMethod.GET) @ResponseBody public String setSession(HttpSession session){ session.setAttribute("id",1); session.setAttribute("name","Test"); return "set session"; } @RequestMapping(path = "/session/get", method = RequestMethod.GET) @ResponseBody public String getSession(HttpSession session) { System.out.println(session.getAttribute("id")); System.out.println(session.getAttribute("name")); return "get session"; }

隨著服務器要處理的請求越來越多，我們不得不分布式部署，減小服務器壓力。

為了負載均衡，我們一般采用nginx來分發請求給各個服務器處理

但是這樣session是無法共享的。

（粘性session）

你可以設置nginx的分配策略，下次同一個還讓同一個服務器來處理

但是很顯然，這就和分布式和nginx初衷違背了：負載很難保證均衡。

（同步session）

一臺服務器的session給所有服務器復制一份

第一，性能不好。第二，產生了一定的耦合

（專門session）

專門一臺服務器來解決，存session，其它服務器來這個服務器取session再用。

但是也有問題：你這個服務器掛了怎么辦？別的服務器都是依賴這個服務器工作的。我們分布式部署本來就是為了解決性能的瓶頸啊。

很容易想到，我們把那個處理session的服務器搞個集群：

更不行，想想就知道，本來就是為了解決分布式部署的問題，你把單獨解決session的服務器又搞集群，和之前有什么區別呢？還不如一個服務器存一份簡單呢。

（存數據庫）

可以，但是傳統的關系數據庫是存到硬盤里，速度太慢。

（nosql）

最終，我們的主流辦法使用nosql數據庫，比如redis，來解決這個問題的，如果有不同意見，歡迎討論。

5.4插曲：RedLock小專欄

概念

Redis 官方站這篇文章提出了一種權威的基于 Redis 實現分布式鎖的方式名叫?Redlock，此種方式比原先的單節點的方法更安全。它可以保證以下特性：

安全特性：互斥訪問，即永遠只有一個 client 能拿到鎖

避免死鎖：最終 client 都可能拿到鎖，不會出現死鎖的情況，即使原本鎖住某資源的 client crash 了或者出現了網絡分區

容錯性：只要大部分 Redis 節點存活就可以正常提供服務

單節點實現

SET resource_name my_random_value NX PX 30000

主要依靠上述命令，該命令僅當 Key 不存在時（NX保證）set 值，并且設置過期時間 3000ms （PX保證），值 my_random_value 必須是所有 client 和所有鎖請求發生期間唯一的，釋放鎖的邏輯是：

if redis.call("get",KEYS[1]) == ARGV[1] then return redis.call("del",KEYS[1]) else return 0 end

上述實現可以避免釋放另一個client創建的鎖，如果只有 del 命令的話，那么如果 client1 拿到 lock1 之后因為某些操作阻塞了很長時間，此時 Redis 端 lock1 已經過期了并且已經被重新分配給了 client2，那么 client1 此時再去釋放這把鎖就會造成 client2 原本獲取到的鎖被 client1 無故釋放了，但現在為每個 client 分配一個 unique 的 string 值可以避免這個問題。至于如何去生成這個 unique string，方法很多隨意選擇一種就行了。

redlock算法

算法很易懂，起 5 個 master 節點，分布在不同的機房盡量保證可用性。為了獲得鎖，client 會進行如下操作：

得到當前的時間，微秒單位

嘗試順序地在 5 個實例上申請鎖，當然需要使用相同的 key 和 random value，這里一個 client 需要合理設置與 master 節點溝通的 timeout 大小，避免長時間和一個 fail 了的節點浪費時間

當 client 在大于等于 3 個 master 上成功申請到鎖的時候，且它會計算申請鎖消耗了多少時間，這部分消耗的時間采用獲得鎖的當下時間減去第一步獲得的時間戳得到，如果鎖的持續時長（lock validity time）比流逝的時間多的話，那么鎖就真正獲取到了。

如果鎖申請到了，那么鎖真正的 lock validity time 應該是 origin（lock validity time） - 申請鎖期間流逝的時間

如果 client 申請鎖失敗了，那么它就會在少部分申請成功鎖的 master 節點上執行釋放鎖的操作，重置狀態

失敗重試

如果一個 client 申請鎖失敗了，那么它需要稍等一會在重試避免多個 client 同時申請鎖的情況，最好的情況是一個 client 需要幾乎同時向 5 個 master 發起鎖申請。另外就是如果 client 申請鎖失敗了它需要盡快在它曾經申請到鎖的 master 上執行 unlock 操作，便于其他 client 獲得這把鎖，避免這些鎖過期造成的時間浪費，當然如果這時候網絡分區使得 client 無法聯系上這些 master，那么這種浪費就是不得不付出的代價了。

放鎖

放鎖操作很簡單，就是依次釋放所有節點上的鎖就行了

性能、崩潰恢復

如果我們的節點沒有持久化機制，client 從 5 個 master 中的 3 個處獲得了鎖，然后其中一個重啟了，這是注意?整個環境中又出現了 3 個 master 可供另一個 client 申請同一把鎖！?違反了互斥性。如果我們開啟了 AOF 持久化那么情況會稍微好轉一些，因為 Redis 的過期機制是語義層面實現的，所以在 server 掛了的時候時間依舊在流逝，重啟之后鎖狀態不會受到污染。但是考慮斷電之后呢，AOF部分命令沒來得及刷回磁盤直接丟失了，除非我們配置刷回策略為 fsnyc = always，但這會損傷性能。解決這個問題的方法是，當一個節點重啟之后，我們規定在 max TTL 期間它是不可用的，這樣它就不會干擾原本已經申請到的鎖，等到它 crash 前的那部分鎖都過期了，環境不存在歷史鎖了，那么再把這個節點加進來正常工作。

Redis 分布式

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。