一步一步打造WebIM(2)——消息缓存

WebIM系列文章

在一步一步打造WebIM(1)一文中，已经介绍了如何实现一个简单的WebIM，但是，这个WebIM有一个问题，就是每一次添加消息监听器时，都必须访问一次数据库去查询是否有消息，显然，如果用户比较多时，必然对数据库的压力比较大。解决这个问题的一个方法就是先将消息缓存在内存中，不立即写入数据库，等到缓存满了才写入数据库。本文将介绍如何实现消息缓存。

基本思路

实现一个消息缓存管理类，以用户为单位缓存所有消息，每一个用户对应着一个List<Message>，保存着该用户新收到的消息，消息缓存管理用一个Hashtable保存着所有用户对应的List<Message>。

具体实现代码如下：

public class MessageCacheManagement
{
    static MessageCacheManagement m_Instance = new MessageCacheManagement();

    static public MessageCacheManagement Instance
    {
        get { return m_Instance; }
    }

    private MessageCacheManagement()
    {
    }

    Int32 m_Count = 0;
    Hashtable m_Cache = new Hashtable();

    List<Message> GetUserMessageCache(String user)
    {
        if (!m_Cache.ContainsKey(user))
        {
            m_Cache.Add(user, new List<Message>());
        }

        return m_Cache[user] as List<Message>;
    }

    /// <summary>
    /// 清除缓存
    /// </summary>
    public void Clear()
    {
        lock (m_Cache)
        {
            List<Message> msgs = new List<Message>();
            foreach (DictionaryEntry ent in m_Cache)
            {
                (ent.Value as List<Message>).Clear();
            }
            m_Count = 0;
        }
    }

    /// <summary>
    /// 获取所有缓存的消息
    /// </summary>
    /// <returns></returns>
    public List<Message> GetAll()
    {
        lock (m_Cache)
        {
            List<Message> msgs = new List<Message>();
            foreach (DictionaryEntry ent in m_Cache)
            {
                foreach (Message msg in ent.Value as List<Message>)
                {
                    msgs.Add(msg);
                }
            }
            return msgs;
        }
    }

    /// <summary>
    /// 获取某一用户缓存的消息的最小时间 
    /// </summary>
    public Nullable<DateTime> GetMinCreatedTime(string user)
    {
        lock (m_Cache)
        {
            List<Message> userMsgs = GetUserMessageCache(user);
            return userMsgs.Count == 0 ? null : new Nullable<DateTime>(userMsgs[0].CreatedTime);
        }
    }


    /// <summary>
    /// 在缓存中插入一条消息
    /// </summary>
    /// <param name="user"></param>
    /// <param name="msg"></param>
    public void Insert(String user, Message msg)
    {
        List<Message> userMsgs = null;

        lock (m_Cache)
        {
            userMsgs = GetUserMessageCache(user);
        }

        lock (userMsgs)
        {
            userMsgs.Add(msg);
            m_Count++;
        }
    }

    /// <summary>
    /// 查找缓存中接受者为user，发送时间大于from的消息
    /// </summary>
    public List<Message> Find(String user, DateTime from)
    {
        List<Message> userMsgs = null;

        lock (m_Cache)
        {
            userMsgs = GetUserMessageCache(user);
        }

        lock (userMsgs)
        {
            List<Message> msgs = new List<Message>();

            int i = 0;
            while (i < userMsgs.Count && userMsgs[i].CreatedTime <= from) i++;

            while (i < userMsgs.Count) { msgs.Add(userMsgs[i]); i++; }

            return msgs;
        }
    }

    /// <summary>
    /// 获取消息总量
    /// </summary>
    public Int32 Count
    {
        get { return m_Count; }
    }
}

添加消息监听器

增加消息缓存后，添加消息监听器的流程也要修改，具体思路是先获取消息接收者在缓存中发送时间最早的消息的发送时间，显然，如果监听器的From大于或等于这个最小发送时间时，无需访问数据库，可以直接访问缓存。具体代码修改为：

/// <summary>
/// 添加消息监听器，如果查找到符合监听器条件的消息，返回false，此时不会添加监听器
/// 如果没有查找到符合监听器条件的消息，返回true，此时监听器将被添加到m_Listeners中
/// </summary>
public bool AddListener(String receiver, String sender, Nullable<DateTime> from, WebIM_AsyncResult asynResult)
{
    MessageListener listener = new MessageListener(receiver, sender, from, asynResult);
    lock (m_Lock)
    {
        if (!m_Listeners.ContainsKey(receiver))
        {
            m_Listeners.Add(receiver, new List<MessageListener>());
        }
        List<MessageListener> listeners = m_Listeners[receiver] as List<MessageListener>;

        //获取用户receiver缓存的消息的最小发送时间
        Nullable<DateTime> min = MessageCacheManagement.Instance.GetMinCreatedTime(receiver);

        List<Message> messages = new List<Message>();

        //当from >= 缓存在内存中的消息的最小时间时，不必查询数据库
        if (min == null || from == null || from.Value < min.Value)
        {
            //查询数据库
            messages.AddRange(Find(receiver, sender, from));
        }

        //在缓存中查询
        messages.AddRange(MessageCacheManagement.Instance.Find(receiver, from.Value));

        if (messages.Count == 0)
        {
            //插入监听器
            listeners.Add(listener);
        }
        else
        {
            //发送消息
            listener.Send(messages);
        }
        return messages.Count == 0;
    }
}

发送消息

增加消息缓存后，发送消息的流程也要修改，具体思路是：先将消息保存到缓存中，之后判断缓存的消息的总数，如果超过设定的上限，就将消息写入数据库。具体代码修改为（您可以通过修改MAX_CACHE_COUNT修改缓存消息数的上限）：

/// <summary>
/// 插入新的消息，插入消息后将查询m_Listeners中是否有符合条件的监听器，如存在，同时将消息发送出去
/// </summary>
public Message NewMessage(String receiver, String sender, DateTime createdTime, String content)
{
    lock (m_Lock)
    {
        Message message = new Message(sender, receiver, content, createdTime, ++m_MaxKey);

        List<Message> messages = new List<Message>();
        messages.Add(message);

        if (m_Listeners.ContainsKey(receiver))
        {
            List<MessageListener> listeners = m_Listeners[receiver] as List<MessageListener>;
            List<MessageListener> removeListeners = new List<MessageListener>();
            foreach (MessageListener listener in listeners)
            {
                if ((listener.Sender == "*" || String.Compare(listener.Sender, sender, true) == 0) &&
                    (listener.From == null || message.CreatedTime > listener.From))
                {
                    listener.Send(messages);
                    removeListeners.Add(listener);

                    System.Threading.ThreadPool.QueueUserWorkItem(new System.Threading.WaitCallback(listener.Complete));
                }
            }

            foreach (MessageListener listener in removeListeners)
            {
                //移除监听器
                listeners.Remove(listener);
            }
        }

        MessageCacheManagement.Instance.Insert(receiver, message);

        if (MessageCacheManagement.Instance.Count >= MAX_CACHE_COUNT)
        {//超过缓存的最大值，将缓存中的消息全部写入数据库
            //启动事务
            SQLiteTransaction trans = m_Conn.BeginTransaction();

            try
            {
                List<Message> cacheMsgs = MessageCacheManagement.Instance.GetAll();

                foreach (Message msg in cacheMsgs)
                {
                    SQLiteCommand cmd = new SQLiteCommand(
                        "insert into Message (Receiver,Sender,Content,CreatedTime,Key) values (?,?,?,?,?)",
                        m_Conn
                    );
                    cmd.Parameters.Add("Receiver", DbType.String).Value = msg.Receiver;
                    cmd.Parameters.Add("Sender", DbType.String).Value = msg.Sender;
                    cmd.Parameters.Add("Content", DbType.String).Value = msg.Content;
                    cmd.Parameters.Add("CreatedTime", DbType.DateTime).Value = msg.CreatedTime;
                    cmd.Parameters.Add("Key", DbType.Int64).Value = msg.Key;

                    cmd.ExecuteNonQuery();
                }

                trans.Commit();
            }
            catch
            {
                trans.Rollback();
            }

            MessageCacheManagement.Instance.Clear();
        }

        return message;
    }
}

源代码下载（此次源代码仅修改了MessageManagement.cs文件，如果您有任何问题，可通过WebIM和我联系）

如果本文对您有所帮助，别忘了点推荐