搭建千万级 Feed 流系统

步骤一：初始化 Timeline 模型

通过 Timeline SDK 分别初始化存储库（Store，发件箱）、同步库（Sync，收件箱）和元数据库（Meta）。存储库配置多元索引，支持按发送者和消息内容检索。

TimelineStoreFactory serviceFactory = new TimelineStoreFactoryImpl(syncClient);

// Initialize store (outbox): stores all published messages
TimelineIdentifierSchema idSchema = new TimelineIdentifierSchema.Builder()
        .addStringField("timeline_id").build();

IndexSchema timelineIndex = new IndexSchema();
timelineIndex.addFieldSchema(new FieldSchema("sender", FieldType.KEYWORD)
        .setIndex(true).setEnableSortAndAgg(true).setStore(true));
timelineIndex.addFieldSchema(new FieldSchema("send_time", FieldType.LONG)
        .setIndex(true).setEnableSortAndAgg(true).setStore(true));
timelineIndex.addFieldSchema(new FieldSchema("text", FieldType.TEXT)
        .setIndex(true).setStore(true).setAnalyzer(FieldSchema.Analyzer.MaxWord));

TimelineSchema storeSchema = new TimelineSchema("feed_timeline_store_table", idSchema)
        .autoGenerateSeqId()
        .setTimeToLive(-1)
        .withIndex("feed_timeline_store_index", timelineIndex);

TimelineStore store = serviceFactory.createTimelineStore(storeSchema);

// Initialize sync (inbox): receives pushed messages from followed users
TimelineSchema syncSchema = new TimelineSchema("feed_timeline_sync_table", idSchema)
        .autoGenerateSeqId()
        .setTimeToLive(-1);

TimelineStore sync = serviceFactory.createTimelineStore(syncSchema);

// Initialize meta: stores timeline metadata
IndexSchema metaIndex = new IndexSchema();
metaIndex.setFieldSchemas(Arrays.asList(
        new FieldSchema("type", FieldType.KEYWORD)
                .setIndex(true).setEnableSortAndAgg(true).setStore(true),
        new FieldSchema("user", FieldType.TEXT)
                .setIndex(true).setStore(true).setAnalyzer(FieldSchema.Analyzer.MaxWord)
));

TimelineMetaSchema metaSchema = new TimelineMetaSchema("feed_timeline_meta_table", idSchema)
        .withIndex("feed_timeline_meta_index", metaIndex);

TimelineMetaStore metaStore = serviceFactory.createMetaStore(metaSchema);

// Create all tables (call once during initialization)
store.prepareTables();
sync.prepareTables();
metaStore.prepareTables();

步骤二：关注用户

调用基础 SDK 的 PutRow 接口写入用户关系表，建立关注关系。

public void follow(String userA, String userB) {
    PrimaryKey primaryKey = PrimaryKeyBuilder.createPrimaryKeyBuilder()
            .addPrimaryKeyColumn("main_user", PrimaryKeyValue.fromString(userA))
            .addPrimaryKeyColumn("sub_user", PrimaryKeyValue.fromString(userB))
            .build();

    RowPutChange rowPutChange = new RowPutChange("feed_user_relation_table", primaryKey);
    rowPutChange.addColumn("timestamp", ColumnValue.fromLong(System.currentTimeMillis()));

    syncClient.putRow(new PutRowRequest(rowPutChange));
}

步骤三：发布消息

发布消息时，先写入发送者的存储库（发件箱），再推送到所有关注者的同步库（收件箱）。这是推拉结合模式中的"推"操作。

public void sendMessage(String userId, Message message, List<String> followers) {
    // 1. Write to sender's store (outbox)
    TimelineIdentifier identifier = new TimelineIdentifier.Builder()
            .addField("timeline_id", message.getTimelineId())
            .build();
    store.createTimelineQueue(identifier).store(message.getTimelineMessage());

    // 2. Push to each follower's sync (inbox)
    for (String friend : followers) {
        TimelineIdentifier friendId = new TimelineIdentifier.Builder()
                .addField("timeline_id", friend)
                .build();
        sync.createTimelineQueue(friendId).store(message.getTimelineMessage());
    }
}

步骤四：读取 Feed 流

读取 Feed 流时，从同步库（收件箱）按自增序列号顺序扫描消息。这是推拉结合模式中的"拉"操作。客户端记录上次读取的 lastSequenceId，下次请求时从该位置继续扫描。

public List<Post> fetchFeed(String userId, long lastSequenceId) {
    TimelineIdentifier identifier = new TimelineIdentifier.Builder()
            .addField("timeline_id", userId)
            .build();

    ScanParameter parameter = new ScanParameter()
            .scanForward(lastSequenceId)
            .maxCount(30);

    Iterator<TimelineEntry> iterator = sync.createTimelineQueue(identifier).scan(parameter);

    List<Post> posts = new LinkedList<>();
    int counter = 0;
    while (iterator.hasNext() && counter++ <= 30) {
        TimelineEntry entry = iterator.next();
        TimelineMessage msg = entry.getMessage();

        Post post = new Post()
                .setName(msg.getString("sender"))
                .setTime(msg.getLong("send_time"))
                .setContent(msg.getString("text"))
                .setSequenceId(entry.getSequenceID());
        posts.add(post);
    }
    return posts;
}

查看指定用户发布的所有消息（用户首页）时，从存储库（Store）而非同步库（Sync）扫描即可，代码结构相同。