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; 
 } 

 @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; 
 } 

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(460795365@qq.com) 
 * @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"); 
  } 
  
}