关于如何构建一个微博型广播 二（转）

原文：http://codecampo.com/topics/196

首先是这个主题的前篇链接 http://codecampo.com/topics/4d7f18bf9f328ba60e000006

前篇文章构思了一个用户广播的实现，并且给出了伪代码。现在 codecampo 已经实现了一个基于 Mongodb + redis 的状态广播，所以可以补充一下前篇没有描述清楚的地方。

0 Timeline 用查询还是缓存？

上篇说到由于广播规则的复杂性，timeline 最好使用一个队列，新增 status 使用投递方式而不依赖数据库查询。

具体看例子，campo 当前的 status 数据会是这样的：

> db.status_bases.findOne({ _type : "Status::Topic" })
{
    "_id" : ObjectId("4df484bde7444a4597000002"),
    "_type" : "Status::Topic",
    "created_at" : ISODate("2011-02-19T12:14:53Z"),
    "tags" : [ ],
    "topic_id" : ObjectId("4d5fb43d9f328b666500000a"),
    "user_id" : ObjectId("4d5fb41b9f328b6665000006")
}

> db.status_bases.findOne({ _type : "Status::Reply" })
{
    "_id" : ObjectId("4df484c0e7444a45970003a7"),
    "_type" : "Status::Reply",
    "created_at" : ISODate("2011-05-21T15:31:30Z"),
    "reply_id" : ObjectId("4dd7dad29f328b74df000018"),
    "targeted" : true,
    "topic_id" : ObjectId("4d5fb94c9f328b666500001f"),
    "user_id" : ObjectId("4d5e8dfc9f328bd543000002")
}

当前有两种类型的 status，一类跟主题创建相关，叫做 Status::Topic，一类跟回复创建相关，叫做 Status::Reply。这两类数据存在同一个 collection 中，数据有相同的地方，比如：userid，topicid；也有各自特性的数据，比如：reply_id，targeted（是否以@开头的直接回复），tags（缓存主题的tags）。

Timeline 的规则是：1、不显示自己的 status 2、不显示 targeted 为 true 的直接回复 3、显示 following 用户的 status 4、显示出现自己喜爱标签的 status 5、显示自己关注主题和自己创建的主题的回复 status 6、按时间排列

如果用数据库查询怎么实现 Timeline？用 mongoid 查询看起来会是这样的：

mark_topic_ids = Topic.where(:marker_ids => @user.id).only(:_id).map(&:_id) 
self_topic_ids = @user.topics.only(:_id).map(&:_id)
topic_ids = (mark_topic_ids + self_topic_ids).uniq
status_ids = Status::Base.where(:targeted.ne => true, :user_id.ne => @user.id).any_of({:user_id.in => @user.following_ids.to_a}, {:tags.in => @user.favorite_tags.to_a}, {:topic_id.in => topic_ids}).asc(:created_at).limit(Stream.status_limit)

生成的 Mongo Query 可能让人吓一跳，因为用来 $in 查询的 followingids 和 favoritetags 还有 topic_ids 会非常长。虽然过早考虑性能不是一个好习惯，但我认为每次都用查询来获取一个不变的列表非常不“自然”。

所以可以考虑建造一个 Timeline 队列缓存，当前有一个非常适合存放 Timeline 的内存型数据库：redis。

1 用 redis 储存 Timeline

先介绍一下 redis：

Redis is an open source, advanced key-value store. It is often referred to as a data structure server since keys can contain strings, hashes, lists, sets and sorted sets.

redis 对 lists 数据的支持很好，优于 mongodb。例如我没找到让 mongodb 简单插入一条数据到 List 头部并且限制长度、丢弃老数据的好方法。

我对 Timeline list 操作的需求如下：

1. 可以插入一个 status id 到列表头部
2. 如果 list 长度超过设定值（比如800），就删除尾部的数据
3. 可以类似翻页式的获取某区间的 ids

campo 实现的 timeline 操作封装在 app/model/stream.rb 文件中，完整代码可以在这里看到。

下面分析一下实现

push status

def push_status(status)
  $redis.lpush store_key, status.id
  $redis.ltrim store_key, 0, Stream.status_limit - 1
end

push_status 操作先用 lpush 操作将 id 从列表左边 push 进去，然后用 ltrim 抛弃列表右边超过指定数量的 id。

获取某区间 ids

def status_ids(start = 0, stop = -1)
  $redis.lrange store_key, start, stop
end

redis 的 lrange 操作可以分段读取 list 数据。实际读取 Timeline 时，先获取 ids，然后再到 mongodb 获取文档数据。具体实现看 Stream#fetch_statuses。

2 重建 Timeline

有两种情况需要重建 Timeline：1、服务崩溃导致队列丢失 2、用户新增订阅。

这时候可以用前面提到的 Mongodb 查询重建 Timeline。重建可以作为后台任务进行，这样无论规则多么复杂都不会阻塞用户的新增订阅的操作。

详细可以看 Stream#rebuild_later 和 Stream#rebuild 的实现。

3 关于数据完整性？

接触 NoSQL 应用之后，经常听到的一个问题是数据完整性。campo 当前的实现有完整性问题么？有的，比如删除一个 status 的时候 Timeline 里面会遗留无效的 id。但根据情况的不同，web 应用通常可以忽略这些完整性：读写需求远大于删除需求、用户本身不在乎数据完整性。

campo 的 Timeline 里面遇到无效 id 的时候，会导致某页的 status 数量不足分页数量，但这不是什么大问题。可以在用户下次触发 Timeline 重建的时候丢弃，或者随着时间的推移被新 status 推后直至丢弃。

当然通过 redis 缓存 + mongodb 也可以查询一个没有缺憾的 Timeline

# slow than fetch_statuses, but complete than fetch_statuses
def statuses
  Status::Base.where(:_id.in => status_ids).desc(:created_at)
end

但是用一个 800 ids 的 $in 查询我觉得不太优雅，所以实际中并没有调用这个方法。

4 小结

现在已经实现了上篇主题中提到的第二阶段 Timeline，而第三阶段的“忽略不活跃用户”，目前 campo 还没有达到这个用户量，就不过度设计了。

对于现在的信息过载的互联网，订阅和广播模式是很好的信息过滤模式。用户应该允许只关注自己感兴趣的内容，并且屏蔽不感兴趣的内容。campo 接下来还会实现用户 block 和主题 mute 功能。

订阅模式在互联网上已经出现很久了，但是具体实现的文章不多，希望本篇给查找此类信息的人一点帮助。