Event Store框架探究

摘要：

　　游戲開發中，經常會越到千奇百怪的Bug。后台程序都是以demon 方式運行，要么GDB，要么Log。一些確定性的bug可以直接使用GDB調試，比如特定請求會Crash。如果是運行一段時間，Bug才會出現，無明顯規律，那么也只能使用Log了。但是從成千上萬條日志中Grep、分析、定位，然后修改代碼、測試，這個過程效率極其低，有的時候挫折感倍強，想罵娘都。經過一些總結后，我們希望程序能夠具有完整跟蹤用戶行為的功能。用戶的行為被完整的記錄下來，針對領域對象提供類似“快照”的功能，當程序出現問題時，我們可以從某個正確的“快照”為起點，回放用戶的操作，這樣Bug可以被重現，修復bug后也可以通過回放用戶操作來驗證正確與否。

挑戰：

event store的相關概念：

　　這樣的”用戶行為回放系統”以前還真沒搞過， 最近對這個功能非常感興趣，查閱了一些資料，前輩們在這方面還是有不少研究的。而且我自己也在實踐DDDD，非常關注DDD社區的活動。DDD社區最近流行的設計思想CQRS強調命令與查詢分離，並且成熟的幾套框架都有集成Even Store。何為Event Store，其就是用來記錄用戶行為的對象。DDD強調關注點集中在和領域問題相關的幾個領域對象上。用戶的操作實際就對應着領域對象的修改。那么將領域對象的每次修改抽象成一個event事件，把這些event都存儲在event Store, 當需要重新構造領域對象時， 遍歷相應的event增量式的構造出領域對象。

推薦：Martin Fowler 的http://martinfowler.com/bliki/CQRS.html

snapshot 快照功能：

　　如果把領域對象的所有操作都記錄為event，隨着時間推移，event可能積累的很多，當構造領域對象時可能需要花費較大的開銷。在領域對象修改到一定量時，snapshot 為領域對象設置快照，那么恢復領域對象時只需從最近的快照開始回放event即可。snapshot保存了領域對象關鍵的修改點，它是對回放event構造領域對象的優化。

如何序列話對象：

　　主要設計到兩種對象，一個是event對象，一個是領域實體entity。event記錄了用戶行為，被event_store按時間（又稱version）順序記錄，entity序列化發生在會entity設置快照時。

　　有序列化當然有反序列化，實體對象必須能夠從序列化的數據中（即snapshot數據）回建對象。並把event在實體對象上回放，也需要將event從序列化數據中回建。

詳細設計：

簡化模型：

　　我們用entity_user對象模擬玩家的操作，假設有兩個接口，inc_gold 和 inc_level，即玩家增漲金錢、增長等級。

類設計:

1. 序列化基類serializer_i， encode用來序列化對象，decode用於反序列化對象

   class serializer_i
   {
   public:
       virtual ~serializer_i(){}
       virtual int decode(const json_value_t& jval_) = 0;
       virtual string encode() const          = 0;
   };

1. 領域實體對象基類 entity_i，領域實體對象必須用於唯一的id以方便進行索引，並且如前文所述，實體對象必須是可以序列化、反序列話的，當對其進行snapshot時需要對其進行序列化，當構造該對象時需要從快照數據中反序列化構建對象。

   class entity_i: public serializer_i
   {
   public:
       entity_i(uint64_t id_):
           m_id(id_)
       {}
       virtual ~entity_i(){}
       uint64_t id() const { return m_id; }
   
   protected:
       uint64_t m_id;
   };

1. 事件基類 event_i，所有對領域對象的修改都是通過Raise一個特定事件完成的，由於C++是強類型的語言並且支持重載，領域對象針對每個event都有一個特定的apply接口。event也繼承自serializer_i，當其被存儲到eventStore中時需要序列化，當要回放event恢復entity時需要反序列化event。

   class event_i: public serializer_i
   {
   public:
       virtual ~event_i(){}
   };

1. 事件派發器event_dispather_i，實際上就是用來實現反射功能，反序列化event時需要根據不同類型的event調用不同的entity的apply接口，此對象能夠保證event會被正確的被apply調用。

   class event_dispather_i
   {
   public:
       ~event_dispather_i(){}
       virtual int dispath(const string& json_) = 0;
   };

1. 事件倉庫基類，其提供功能有三

   class event_store_i
   {
   public:
       virtual ~event_store_i(){}
       virtual int save_event(uint64_t entity_id_, const event_i& event_) = 0;
       virtual int snapshot_entity(const entity_i& entity_) = 0;
       virtual int constuct_snapshot_last(entity_i& entity_, event_dispather_i& event_dispacher_, int version_ = 1) = 0;
   };

　　1. 存儲event事件

　　2.  保存領域對象實體的快照數據

　　3.  通過某個版本的快照，回建領域對象

1. 結構圖如下

   

7. 示例代碼 

1. User實體對象：User維護兩個成員變量gold和level，用來表示當前用戶的金錢和等級，inc_gold和inc_level是兩個Cmd接口，驗證參數有效Raise一個inc_gold_event事件，參見示例代碼：

   class event_store_i
   {
   public:
       virtual ~event_store_i(){}
       virtual int save_event(uint64_t entity_id_, const event_i& event_) = 0;
       virtual int snapshot_entity(const entity_i& entity_) = 0;
       virtual int constuct_snapshot_last(entity_i& entity_, event_dispather_i& event_dispacher_, int version_ = 1) = 0;
   };

　　為簡化操作，我們假設金錢是經常變更的，而等級變化較慢，level變化為關鍵變化，每當level改變我們都會為實體建立新的快照：

int entity_user_t::inc_level(int32_t level_)
{
    if (level_ <= 0)
        return -1;
    inc_level_event_t event(level_);
    apply(event);
    m_event_store->save_event(this->id(), event);
    m_event_store->snapshot_entity(*this);
    return 0;
}

而對應的apply接口則非常的簡單，因為參數已經進過驗證，apply是實實在在的改變對象狀態的內部方法：

void entity_user_t::apply(const inc_level_event_t& event_)
{
    m_level += event_.level;
}

1. event 對象除了用於特定的數據字段，最主要的當屬decode和encode接口。這里為了方便調試我使用了json序列化和反序列化方式，json的decode和encode有不小的開銷，基於二進制的序列化和反序列化可以達到很高的實時性，存在很大的優化空間。

   struct inc_level_event_t: public event_i
   {
       inc_level_event_t():
           level(0)
       {}
       inc_level_event_t(int32_t level_):
       level(level_)
       {}
       int decode(const json_value_t& jval_)
       {
           json_instream_t in("inc_level_event_t");
           in.decode("level", jval_["level"], level);
           return 0;
       }
       string encode() const
       {
           rapidjson::Document::AllocatorType allocator;
           rapidjson::StringBuffer            str_buff;
           json_value_t                       ibj_json(rapidjson::kObjectType);
           json_value_t                       ret_json(rapidjson::kObjectType);
           
           json_outstream_t out(allocator);
           out.encode("level", ibj_json, level);
           ret_json.AddMember("inc_level_event_t", ibj_json, allocator);
           
           rapidjson::Writer<rapidjson::StringBuffer> writer(str_buff, &allocator);
           ret_json.Accept(writer);
           string output(str_buff.GetString(), str_buff.GetSize());
           return output;
       }
       int32_t level;
   };

   
   
2. event store 的實現，為了簡便起見，本示例框架並沒有將序列化的數據落盤，而是直接存儲在內存中。真是的eventStore可以采用寫文件的方式或者Sqlite也是很好的方案。

   class event_store_mem_t: public event_store_i
   {
       typedef vector<string>                event_record_t;
       typedef map<uint64_t, event_record_t> entity_event_map_t;
       
       struct  snapshot_info_t
       {
           long   event_version;
           string data;
       };
       typedef vector<snapshot_info_t>       entity_record_t;
       typedef map<uint64_t, entity_record_t>entity_snapshot_map_t;
   public:
       event_store_mem_t();
       ~ event_store_mem_t();
       
       int save_event(uint64_t entity_id_, const event_i& event_);
       int snapshot_entity(const entity_i& entity_);
       int constuct_snapshot_last(entity_i& entity_, event_dispather_i& event_dispacher_, int version_ = 1);
   
   private:
       entity_event_map_t    m_entity_events;
       entity_snapshot_map_t m_entity_snapshot;
   };

下面時基於內存的eventStore的實現：

int event_store_mem_t::save_event(uint64_t entity_id_, const event_i& event_)
{
    m_entity_events[entity_id_].push_back(event_.encode());
    return 0;
}

int event_store_mem_t::snapshot_entity(const entity_i& entity_)
{
    snapshot_info_t info;
    info.event_version = m_entity_events[entity_.id()].size();
    info.data          = entity_.encode();
    m_entity_snapshot[entity_.id()].push_back(info);
    return 0;
}

int event_store_mem_t::constuct_snapshot_last(entity_i& entity_, event_dispather_i& event_dispacher_, int version_)
{
    int index = m_entity_snapshot.size() - version_;
    if (index >=0 && index < (int)m_entity_snapshot.size())
    {
        snapshot_info_t& info = m_entity_snapshot[entity_.id()][index];
        
        json_dom_t document;
        if (document.Parse<0>(info.data.c_str()).HasParseError())
        {
            throw msg_exception_t("json format not right");
        }
        if (false == document.IsObject() && false == document.Empty())
        {
            throw msg_exception_t("json must has one field");
        }

        entity_.decode(document.MemberBegin()->value);
        
        for (size_t i = info.event_version; i < m_entity_events[entity_.id()].size(); ++i)
        {
            event_dispacher_.dispath(m_entity_events[entity_.id()][i]);
        }
    }
    return 0;
}

1. 恕不絮煩，詳細實現代碼參見google code

　　svn co  https://ffown.googlecode.com/svn/trunk/example/event_store

8. 待改進之處

• 序列化可以采用二進制方式
• event_dispacher可以使用idl分析器實現，思路參見blog：http://www.cnblogs.com/zhiranok/archive/2012/02/28/CPP_message_dispath.html

• event Store 需要將數據落盤,可以寫文件、levelDB或這Sqlite等。