使用protobuf(c++)的构建通用的数据系统(转)

使用protobuf(c++)的构建通用的数据系统(转)

2012-04-11 发表评论

 

本文以C++语言举例，其他语言应该也有类似的解法。

假定我们定义了一个Persion的message type，我们的用法可能如下

定义person.proto文件，并用protoc编译出person.pb.h 和 person.pb.cc

package tutorial;
message Person {
required int32 id = 1;
optional string name = 2;
optional string email = 3;
}

编写自己的代码进行数据的填充和序列化，并引入person.pb.h,和person.pb.cc

Person person;
person.set_id(1234);
person.set_name("John Doe");
person.set_email("jdoe@example.com");
string output;
person.SerializeToString(&output);

 

假设我们的系统应用场景是：数据都是持久化在mysql中，同时提供memcached作为缓存。在使用中通常缓存miss后会从mysql中读取数据后进行序列化并set到memcached中再返回给用户。

 

还是以上面的message type为例，在mysql中还有一张表持久化存储了对应的数据。我们把mysql定义的数据的属性名跟proto定义的属性名都是一样的。现在我们想把mysql中读出来的某条记录(mysqlpp::Row)转换成对应的protobuf的消息

Create table Person {
id int(11) not null,
name varchar(64),
emal varchar(255),
........
}

如果我们想把数据库读出来的一条记录进行序列化，需要先转换成protobuf定义的Message，然后调用Message的SerializeToString进行序列化

我们的转换函数可能是这么写的

void fill(mysqlpp::Row& row, tutorial::Person&  persion){
persion.set_id((int)row[“id”]);
persion.set_name((std::string)row[“name”]);
persion.set_email((std::string)row[“email”]);
}

这种方法要求我们针对一个新来的message type就需要新写一段这样的代码。如果我们的系统支撑了很多不同的消息类型，每个消息类型的名称，参数个数，名称以及类型都不同。在填充数据的时候针对每个类型都需要写类似的代码，必然会制约我们系统的扩展性和维护性。幸好protobuf提供了反射的机制，给我们提供了更多发挥想象力的空间。

 

所有的message type(比如tutorial::Persion)都是继承自Message class。

Message class 定义了 New() 虚函数，用以返回本对象的一份新实例，类型与本对象的真实类型相同。也就是说，拿到 Message* 指针，不用知道它的具体类型，就能创建和它类型一样的具体 Message Type 的对象。

Message class 定义了GetReflection虚函数，可以返回一个Reflection对象，这个对象可以动态的访问和修改message里面的field内容

每个Message Type对应一个Descriptor对象,通过这个对象我们可以在运行时得到Message Type包含的field，以及field的类型描述。

那我们如何拿到一个Message Type对应的Message* 指针了？

//在上面的例子中使用， typeName = "tutorial::Persion"
Message* createMessage(const std::string& typeName)
{
Message* message = NULL;
const Descriptor* descriptor = DescriptorPool::generated_pool()->FindMessageTypeByName(typeName);
if (descriptor)
{
const Message* prototype = MessageFactory::generated_factory()->GetPrototype(descriptor);
if (prototype)
{
message = prototype->New();
}
}
return message;
}

这样我们就可以根据Message Type的名字获取到对应的Message* 指针

那又如何动态的把mysql中的数据填充进去了？

 

void convert(const mysqlpp::Row& row, std::string& data){
Message* message = createMessage(message_type);
const Reflection* reflection = message->GetReflection();
const Descriptor* descriptor = message->GetDescriptor();

for(int i=0; i < descriptor->field_count(); ++i){
reflectionFill(message,descriptor->field(i),reflection,row);
}
message->SerializeToString(&data);
delete message;
}

void reflectionFill(Message* message, const FieldDescriptor* descriptor, const Reflection* reflection, const mysqlpp::Row& row){

assert(descriptor != NULL);
//current not support for repeated label
if(descriptor->label() == FieldDescriptor::LABEL_REPEATED){
return;
}

const char* name = descriptor->name().c_str();

switch(descriptor->type()){
case FieldDescriptor::TYPE_FIXED64 :
case FieldDescriptor::TYPE_INT64 :
reflection->SetInt64(message,descriptor,(long long)row[name]);break;
case FieldDescriptor::TYPE_UINT64 :
reflection->SetInt64(message,descriptor,(unsigned long long)row[name]);break;

case FieldDescriptor::TYPE_FIXED32:
case FieldDescriptor::TYPE_INT32 :
reflection->SetInt32(message,descriptor,(int)row[name]);break;
case FieldDescriptor::TYPE_UINT32 :
reflection->SetInt32(message,descriptor,(unsigned int)row[name]);break;

case FieldDescriptor::TYPE_STRING :
reflection->SetString(message,descriptor,(std::string)row[name]);break;

case FieldDescriptor::TYPE_DOUBLE :
reflection->SetDouble(message,descriptor,(double)row[name]);break;
case FieldDescriptor::TYPE_FLOAT :
reflection->SetFloat(message,descriptor,(float)row[name]);break;
case FieldDescriptor::TYPE_BOOL :
reflection->SetBool(message,descriptor,(bool)row[name]);break;
default : std::cerr << "not support type " << descriptor->type() << std::endl;
}
}

这种方式要求我们必须在编译的时候就把对应的 xxx.proto 生成的 xxx.pb.h和 xxx.pb.cc编译进去，这样我们在增加proto的时候每次都需要重新编译。

不过protobuf提供了动态编译的功能，让我们在程序运行期也可以动态的编译xxx.proto，这样我们在运行期使用动态生成的Messag class。

class MockErrorCollector : public MultiFileErrorCollector {
public:
MockErrorCollector() {}
~MockErrorCollector() {}

std::string text_;

// implements ErrorCollector ---------------------------------------
void AddError(const std::string& filename, int line, int column,
const std::string& message) {
std::cerr << "error filename "<<filename <<" message "<<message <<std::endl;
}
};
MockErrorCollector errorCollector;
google::protobuf::compiler::DiskSourceTree sourceTree;
google::protobuf::compiler::Importer importer(&sourceTree, &errorCollector);
sourceTree.MapPath("", "./");
importer.Import("person.proto");

使用这种方式可以在运行期动态编译新添加proto，不过Message*的获取稍有不同,是因为之前的获取方式只能用于编译期添加的Message class

需要使用进行如下的修改

Message* createMessage(const std::string& type_name)
{
Message* message = NULL;
const Descriptor* descriptor =
DescriptorPool::generated_pool()->FindMessageTypeByName(type_name);
if (descriptor)
{
const Message* prototype =
MessageFactory::generated_factory()->GetPrototype(descriptor);
if (prototype)
{
message = prototype->New();
}
}else {
descriptor = importer.pool()->FindMessageTypeByName(type_name    DynamicMessageFactory * dynamicMessageFactory=new DynamicMessageFactory();
if(descriptor){
message = dynamicMessageFactory->GetPrototype(descriptor)->New();
}else{
std::cerr << "not found message type " << type_name << std::endl;
}
}
return message;
}