INM消息处理

消息处理 总流程图

statement_list

namespace inm
{
	enum StatementType
	{
		S_UNKNOWN,	/**< 非法类型 */
		S_INSTANCE,	/**< 实例增删改 */
		S_SCHEMA,	/**< 模式增删改 */
		S_UDT,		/**< 自定义类型增删改 */
		S_QUERY,  	/**< 查询 */
		S_UPDATE,        /**< 更新 */
		S_SUPDATE,
		S_LOGIN,	/**< 登录类型提示消息，与客户端对应，值=7 位置不可变>*/
		S_FUPDATE,
		S_FUNCTION,
		S_INVOKE,
		S_VIEW,
		S_REFRESH       /**物化视图更新 xj**/
		
	};
}

scheduler::schedule

将语句根据类型放到对应的vector中，分别执行对应的exec函数。

例：
input:query1、query2、query3、insert1、insert2、create1、create2。

经过语法分析后：
vector<scheclasspre*> :create1、create2。->execschmlist
vector<query*>：query1、query2、query3。->execquerylist
vector<instance*>：insert1、insert2。->execinstancelist

vector<statement_list>::iterator iter = yy_extra.smt_lists.begin();
		for(; iter != yy_extra.smt_lists.end(); ++iter) {
			switch(iter->type) {
				case S_SCHEMA:
					execSchmList((vector<SchmClassPre *> *)iter->list);
					break;
				case S_VIEW:
					execViewList((vector<ViewClass *> *)iter->list);
					break;
			       case S_REFRESH:
					execRefreshList((vector<UpdateViewClass *> *)iter->list);
					break;
				case S_INSTANCE:
					execInstList((InstanceList *)iter->list);
					break;
				case S_QUERY:
					request->getInstOperator()->syncDirty();
					execQueryList((list<Query *> *)iter->list);
					request->getInstOperator()->freeDirty();
					break;
				case S_UPDATE:
					execUpdateList((list<Update *> *)iter->list);
					break;
				case S_UDT:
					execUdtList((vector<SchmUserType *> *)iter->list);
					break;
				case S_SUPDATE:
					execSchmUpdateList((vector<SchmUpdate *> *)iter->list);
					break;
         	                case S_FUNCTION:
					execFunctionList((vector<FunctionClass *> *)iter->list);
					break;
				case S_INVOKE:
					execFunctionInvkList((vector<FunctionInvk *> *)iter->list);
					break;
    	                        case S_FUPDATE:
					execFunctionUpdateList((vector<FunctionUpdate *> *)iter->list);
					break;
				default:
					log.error(C_INVAL) << "Unknown Type: " << iter->type;
					break;
			}
		}

operator

可以看到并不是scheduler直接操作数据，而是交由各个operator处理。
scheme有operator,instance也有operator…

void	Scheduler::execSchmList(vector<SchmClassPre *> *schms)
	{
		request->setStatementType(S_SCHEMA);
		vector<SchmClassPre *>::iterator iter = schms->begin();
		for(; iter != schms->end(); ++iter) {
			...
			request->getSchmOperator()->insert(sc);
			...
		}
		delete schms;
	}

创建模式

INM采用的是缓冲区+底层存储的方式。

缓冲区：map<id,node> 缓冲区可以被重载成各种类型，node可以是schemeclass* queryclass*。
底层存储：每种类型都由主数据库和索引数据库组成。索引数据库用于用户查询，主数据库用于序列化后保存。

插入更新操作一般会先交由缓冲区处理，但考虑到其他用户可能一会就要查询，因此会在处理完后被更新到索引数据库中。等待合适的时机再序列化到主数据库中。

/** 插入模式
	 *
	 * 将模式插入到缓冲区中，并立即更新索引。
	 * 因为之后所有可能的操作都可能通过名字查找模式，因此要立即更新索引。
	 * @param schm 要插入的模式
	 */
	void	SchmOperator::insert(SchmClass	*schm)
	{
		db.updateName(schm);
		buffer.insert(schm);
	}

与MongoDB交互方案

inm 用户输入statement 经过词法/语法分析->statement_list 经过序列化/反序列化->bson->mongocxx驱动->mongodb

1. INM用户输入statement

2.词法/语法分析

void	Scheduler::init()
	{
		yyscanner = scanner_init(statement, &yy_extra.inl_yy_extra);
		parser_init(&yy_extra);
		inl_yyparse(yyscanner);
		scanner_finish(yyscanner);
	}

3. 序列化

代替原有的序列化方案，将INM数据结构->bson格式。

	...
	class SchmAttribute;
	class SchmRelation;
	class SchmRelationPre;
	class SchmClassPre;
	class SchmUpdatePath;
	class SchmUpdatePathResult;
	class SchmUpdateDefinition;
	class Instance;
	class ViewModifyTuple;
	class SchmClass
	{
		private:
			string className;	/*<< 模式名 */
			set<uint> superClass;	/*<< 父类 */
			set<uint> subClass;	/*<< 子类 */
			vector<SchmAttribute *> attributes;	/*<< 属性 */
			vector<SchmRelation *> relations;	/*<< 关系 */
			string version;		/*<< version值 */
			string oldClassName;	/*<< old模式名 */
			string oldVersion;	/*<< old模式version */
			uint id;		/*<< 模式id号 */
			SchmClassType classType;/*<< 模式类型 */
			vector<vector<string>*> keys; //keys
			bool isCloneFlag;
			map<uint,ViewModifyTuple *> modifyTuples;//modifyList的结构，初始化以及管理
            //hhd 20150207 调用标记的id，
            //调用标记用于指明哪些实例的规则推导需要调用属于当前类的实例
            uint markId;
            ...

4. mongocxx驱动

mongoDB的官方CPP驱动。
当通过序列化将INM的数据结构序列换成bson以后，可以通过mongoDB的官方CPP驱动进行CRUD的操作.

#include <iostream>
#include <bsoncxx/builder/basic/document.hpp>
#include <bsoncxx/builder/basic/kvp.hpp>
#include <bsoncxx/json.hpp>
#include <mongocxx/client.hpp>
#include <mongocxx/instance.hpp>
#include <mongocxx/uri.hpp>
using bsoncxx::builder::basic::kvp;
using bsoncxx::builder::basic::make_document;
using bsoncxx::to_json;
using namespace mongocxx;
int main() {
    // The mongocxx::instance constructor and destructor initialize and shut down the driver,
    // respectively. Therefore, a mongocxx::instance must be created before using the driver and
    // must remain alive for as long as the driver is in use.
    mongocxx::instance inst{};
    mongocxx::client conn{mongocxx::uri{"mongodb://localhost/?replicaSet=replset"}};
    // By default, a session is causally consistent. Pass options::client_session to override
    // causal consistency.
    auto session = conn.start_session();
    auto coll = conn["db"]["collection"];
    auto result = coll.update_one(session,
                                  make_document(kvp("_id", 1)),
                                  make_document(kvp("$inc", make_document(kvp("x", 1)))));
    std::cout << "Updated " << result->modified_count() << " documents" << std::endl;
    // Read from secondary. In a causally consistent session the data is guaranteed to reflect the
    // update we did on the primary. The query may block waiting for the secondary to catch up,
    // or time out and fail after 2 seconds.
    options::find opts;
    read_preference secondary;
    secondary.mode(read_preference::read_mode::k_secondary);
    opts.read_preference(secondary).max_time(std::chrono::milliseconds(2000));
    auto cursor = coll.find(session, make_document(kvp("_id", 1)), opts);
    for (auto&& doc : cursor) {
        std::cout << bsoncxx::to_json(doc) << std::endl;
    }
}

最后

以上就是曾经画笔为你收集整理的序列化成BSON，实现与MongoDB交互INM消息处理与MongoDB交互方案的全部内容，希望文章能够帮你解决序列化成BSON，实现与MongoDB交互INM消息处理与MongoDB交互方案所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。