Redis實現多人多聊天室功能

本文為大家分享了Redis支持多人多聊天室功能的設計代碼，供大家參考，具體內容如下

設計原理

左邊的一個數據域,代表兩個聊天室,聊天室id分別是827,729

在聊天室827裡,有2個人,分別是jason22,jeff24他們分別已經閱讀過聊天室內的id為5和6的消息

右邊的一個數據域,代表了用戶在不同的聊天室,jason22參與了827與729聊天室,在這兩個聊天室裡,他分別閱讀到了id為5和id為10的消息

另外827聊天室內id為5的消息與729聊天室內id為5的消息不一樣。

同時還有三個域
msgs:chatid
這是一個zset,有序集合,member是消息體,score是消息id
代表的是某個聊天室內已經發出的消息
另外 這裡面存的是有用的消息,已經被所有人都閱讀的消息就會被刪除

ids:chatid
是一個String型的數據,裡面放的是最新的消息的編號(發消息時,自增這個字段,即可獲得最新的值)

ids:chat:
是一個String型的數據,裡面放的是最新的聊天室的編號(創建聊天室時,自增這個字段)

代碼

OK 開始看代碼 

public String createChat(Jedis conn, String sender, Set<String> recipients, String message) { 
//啟動的時候redis裡是沒有ids:chat:這個鍵的 
//自增之後返回1 
    String chatId = String.valueOf(conn.incr("ids:chat:")); 
    return createChat(conn, sender, recipients, message, chatId); 
  } 
 
 
  /** 
  * 
  * @param conn 
  * @param sender 發送消息的人 
  * @param recipients 接受消息的人 
  * @param message 待發送的消息 
  * @param chatId 聊天室的編號 
  * @return 
  */ 
  public String createChat( Jedis conn, String sender,  
    Set<String> recipients, String message, String chatId){ 
  //自己發的消息 自己也能接受到 
    recipients.add(sender); 
 
 
    Transaction trans = conn.multi(); 
    for (String recipient : recipients){ 
  //聊天室的成員 最開始時 都閱讀的是0號信息 
      trans.zadd("chat:" + chatId, 0, recipient); 
  //記錄每個人參加的聊天室 
      trans.zadd("seen:" + recipient, 0, chatId); 
    } 
    trans.exec(); 
 
 
    return sendMessage(conn, chatId, sender, message); 
  } 
 
 
  public String sendMessage(Jedis conn, String chatId, String sender, String message) { 
   
  //鎖住聊天室 為啥? 人員變動了咋辦 
  //這個acquireLock見上一章 
    String identifier = acquireLock(conn, "chat:" + chatId); 
    if (identifier == null){ 
      throw new RuntimeException("Couldn't get the lock"); 
    } 
    try { 
    //給要發布的消息設定一個最新的編號 第一次時 返回的是1 
      long messageId = conn.incr("ids:" + chatId); 
      HashMap<String,Object> values = new HashMap<String,Object>(); 
      values.put("id", messageId); 
      values.put("ts", System.currentTimeMillis()); 
      values.put("sender", sender); 
      values.put("message", message); 
      String packed = new Gson().toJson(values); 
       
      //某個聊天室的消息列表 
      //最舊的消息----消息json  
      //默認的zset是按照score的值從小到大排序 
      conn.zadd("msgs:" + chatId, messageId, packed); 
    }finally{ 
      releaseLock(conn, "chat:" + chatId, identifier); 
    } 
    return chatId; 
  } 

 發消息現在就OK了,剩下的就是用戶去拉取未讀的消息了。這個比較麻煩,恩,相當的麻煩 

 @SuppressWarnings("unchecked") 
  public List<ChatMessages> fetchPendingMessages(Jedis conn, String recipient) { 
   
  //獲得用戶在各個聊天室 已經看到的最新消息的id 
  //有幾個聊天室 seenSet的size就是幾 
    Set<Tuple> seenSet = conn.zrangeWithScores("seen:" + recipient, 0, -1); 
    List<Tuple> seenList = new ArrayList<Tuple>(seenSet); 
 
 
    Transaction trans = conn.multi(); 
    for (Tuple tuple : seenList){ 
      String chatId = tuple.getElement(); 
      int seenId = (int)tuple.getScore(); 
      //獲取每個聊天室裡 未讀的所有消息 
      //min 和 max 可以是 -inf 和 +inf 
      trans.zrangeByScore("msgs:" + chatId, String.valueOf(seenId + 1), "inf"); 
    } 
    //我參加了幾個聊天室 results的長度就是幾 
    List<Object> results = trans.exec(); 
 
 
    //com.google.gson.Gson jar包自己下載吧 
    Gson gson = new Gson(); 
    Iterator<Tuple> seenIterator = seenList.iterator(); 
    Iterator<Object> resultsIterator = results.iterator(); 
 
 
    //用戶最後成功拉取的未讀消息 存放在chatMessages 
    List<ChatMessages> chatMessages = new ArrayList<ChatMessages>(); 
    List<Object[]> seenUpdates = new ArrayList<Object[]>(); 
    List<Object[]> msgRemoves = new ArrayList<Object[]>(); 
     
     
    //這個大的while循環 用戶參與了幾個聊天室 就循環幾次 
    while (seenIterator.hasNext()){ 
      Tuple seen = seenIterator.next(); 
      Set<String> messageStrings = (Set<String>)resultsIterator.next(); 
      if (messageStrings.size() == 0){ 
      //沒有未讀的消息 
        continue; 
      } 
 
 
      //代碼運行到這裡 
      //說明 我在某個聊天室 還有未讀的消息 
      //seedid記錄我已經拉取到的消息 初始為0 
      int seenId = 0; 
      //當前處理的是哪個聊天室 
      String chatId = seen.getElement(); 
       
      List<Map<String,Object>> messages = new ArrayList<Map<String,Object>>(); 
       
      //我在聊天室未讀的消息列表 
      for (String messageJson : messageStrings){ 
        Map<String,Object> message = (Map<String,Object>)gson.fromJson( 
          messageJson, new TypeToken<Map<String,Object>>(){}.getType()); 
        int messageId = ((Double)message.get("id")).intValue(); 
 
 
        if (messageId > seenId){ 
          seenId = messageId; 
        } 
        message.put("id", messageId); 
        //加入到成功拉取的列表裡 
        messages.add(message); 
      } 
      //更新我在這個聊天室讀到的最新消息 
      conn.zadd("chat:" + chatId, seenId, recipient); 
       
      //記錄我在某個聊天室讀到的最新記錄 
      seenUpdates.add(new Object[]{"seen:" + recipient, seenId, chatId}); 
 
 
      //取出第0個member-score 
      Set<Tuple> minIdSet = conn.zrangeWithScores("chat:" + chatId, 0, 0); 
      //為啥刪除呢? 每個聊天室是一個zset表 第一條記錄代表的就是 所有用戶至少都讀了的消息 
      if (minIdSet.size() > 0){ 
      Tuple tuple=minIdSet.iterator().next(); 
      System.out.println("要刪除的 tuple:"+tuple.getElement()+"--"+tuple.getScore()); 
        msgRemoves.add(new Object[]{"msgs:" + chatId, tuple.getScore()}); 
      } 
      chatMessages.add(new ChatMessages(chatId, messages)); 
    } 
 
 
    trans = conn.multi(); 
    for (Object[] seenUpdate : seenUpdates){ 
      trans.zadd( 
        (String)seenUpdate[0], 
        (Integer)seenUpdate[1], 
        (String)seenUpdate[2]); 
    } 
    for (Object[] msgRemove : msgRemoves){ 
      trans.zremrangeByScore( 
        (String)msgRemove[0], 0, ((Double)msgRemove[1]).intValue()); 
    } 
    trans.exec(); 
 
 
    //返回的是我這次拉取獲得的 最新的消息 
    return chatMessages; 
  } 

OK,咱們看看測試代碼:

package redisinaction; 
 
 
import java.util.Arrays; 
import java.util.HashSet; 
import java.util.Iterator; 
import java.util.List; 
import java.util.Map; 
import java.util.Set; 
 
 
import org.junit.BeforeClass; 
import org.junit.Test; 
 
 
import jedis.redis_in_action.Chapter06; 
import jedis.redis_in_action.Chapter06.ChatMessages; 
import redis.clients.jedis.Jedis; 
import redis.clients.jedis.Tuple; 
 
 
/**  
 * This class is used for ...  
 * @author dlf([email protected]) 
 * @version 1.0, 2016年10月17日 下午10:15:58  
 */ 
public class Chapter06Test { 
  static Jedis conn = null; 
  static Chapter06 c=null; 
   
  @BeforeClass 
  public static void initConn(){ 
    System.out.println("test before"); 
    conn = new Jedis("10.150.0.80"); 
    conn.auth("dlf123123"); 
     
    c=new Chapter06(); 
  } 
   
 
 
   
  @Test 
   public void testMultiRecipientMessaging() { 
      System.out.println("\n----- testMultiRecipientMessaging -----"); 
      conn.del("ids:chat:", "msgs:1", "ids:1", "seen:joe", "seen:jeff", "seen:jenny"); 
 
 
      System.out.println("Let's create a new chat session with some recipients..."); 
      Set<String> recipients = new HashSet<String>(); 
      recipients.add("jeff"); 
      recipients.add("jenny"); 
      String chatId = c.createChat(conn, "joe", recipients, "message 1"); 
      System.out.println("Now let's send a few messages..."); 
      for (int i = 2; i < 5; i++){ 
        c.sendMessage(conn, chatId, "joe", "message " + i); 
      } 
      System.out.println(); 
      System.out.println("看看消息庫"); 
      //消息庫裡的所有消息 
      Set<Tuple> messageFromBase=conn.zrangeWithScores("msgs:"+chatId, 0, -1); 
      Iterator<Tuple> iterator=messageFromBase.iterator(); 
      while(iterator.hasNext()){ 
        Tuple tuple=iterator.next(); 
        System.out.println(tuple.getElement()+"　－－　　"+tuple.getScore()); 
      } 
      System.out.println("And let's get the messages that are waiting for jeff and jenny..."); 
       
       
      List<ChatMessages> r1 = c.fetchPendingMessages(conn, "jeff"); 
      List<ChatMessages> r2 = c.fetchPendingMessages(conn, "jenny"); 
    //當我拉取了joe的未讀信息後 就會刪除msgs:1裡面的信息 
    //為什麼?想明白了麼? 
      List<ChatMessages> r3 = c.fetchPendingMessages(conn, "joe"); 
      System.out.println("They are the same? " + r1.equals(r2)); 
       
      System.out.println("Those messages are:"); 
      for(ChatMessages chat : r1){ 
        System.out.println(" chatId: " + chat.chatId); 
        System.out.println("  messages:"); 
        for(Map<String,Object> message : chat.messages){ 
          System.out.println("   " + message); 
        } 
      } 
 
 
      System.out.println("看看還有沒"); 
      messageFromBase=conn.zrangeWithScores("msgs:"+chatId, 0, -1); 
       iterator=messageFromBase.iterator(); 
      while(iterator.hasNext()){ 
        Tuple tuple=iterator.next(); 
        System.out.println(tuple.getElement()+"　－－　　"+tuple.getScore()); 
      } 
      conn.del("ids:chat:", "msgs:1", "ids:1", "seen:joe", "seen:jeff", "seen:jenny"); 
    } 
   
} 

搞定了,大家不妨把代碼復制一份,自己看看
下面的是測試的結果

test before

----- testMultiRecipientMessaging -----
Let's create a new chat session with some recipients...
Now let's send a few messages...

看看消息庫
{"sender":"joe","id":1,"message":"message 1","ts":1477276890018}　－－　　1.0
{"sender":"joe","id":2,"message":"message 2","ts":1477276890113}　－－　　2.0
{"sender":"joe","id":3,"message":"message 3","ts":1477276890115}　－－　　3.0
{"sender":"joe","id":4,"message":"message 4","ts":1477276890116}　－－　　4.0
And let's get the messages that are waiting for jeff and jenny...
要刪除的 tuple:jenny--0.0
要刪除的 tuple:joe--0.0
要刪除的 tuple:jeff--4.0
They are the same? true
Those messages are:
  chatId: 1
    messages:
      {sender=joe, id=1, message=message 1, ts=1.477276890018E12}
      {sender=joe, id=2, message=message 2, ts=1.477276890113E12}
      {sender=joe, id=3, message=message 3, ts=1.477276890115E12}
      {sender=joe, id=4, message=message 4, ts=1.477276890116E12}
看看還有沒

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持。