用C++的高级模版特性实现一个不需要IDL的RPC

作者: rockeet 发表日期: 2006年12月09日分类: C++序列化评论: 4 条阅读次数: 4,546 次

当然，首要的是为了有用，其次是挑战自己驾驭 C++ 的能力，目前已经全部完成，并且取得了非常好的效果。使用该 RPC 的简短示例代码：

//////////////////////////////////////////////////////////////////////////
// sample usage...
// test.h 

typedef std::vector<unsigned> vint_vec;

class AsyncInterface : public GlobaleScope {
public:
	BEGIN_RPC_ADD_MF(AsyncInterface)
		RPC_ADD_MF(get_val)
		RPC_ADD_MF(get_len)
		RPC_ADD_MF(squareVec)
		RPC_ADD_MF(multiVec)
	END_RPC_ADD_MF()

	RPC_DECLARE_MF(get_val, (rpc_in<int> x))
	RPC_DECLARE_MF(get_len, (const std::string& x))
	RPC_DECLARE_MF(squareVec, (vint_vec& x))
	RPC_DECLARE_MF(multiVec, (vint_vec& z, vint_vec& x, vint_vec& y))
};
RPC_TYPEDEF_PTR(AsyncInterface);

// more convenient way than AsyncInterface...
BEGIN_RPC_INTERFACE(SampleRPC_Interface2, SessionScope)
	RPC_ADD_MF(get_val)
	RPC_ADD_MF(get_len)
END_RPC_ADD_MF()
	RPC_DECLARE_MF(get_val, (rpc_in<int> x))
	RPC_DECLARE_MF(get_len, (const std::string& x))
END_RPC_INTERFACE()

//////////////////////////////////////////////////////////////////////////

// sample usage...

// test.h

typedef std::vector<unsigned> vint_vec;

class AsyncInterface : public GlobaleScope {

public:

BEGIN_RPC_ADD_MF(AsyncInterface)

RPC_ADD_MF(get_val)

RPC_ADD_MF(get_len)

RPC_ADD_MF(squareVec)

RPC_ADD_MF(multiVec)

END_RPC_ADD_MF()

RPC_DECLARE_MF(get_val, (rpc_in<int> x))

RPC_DECLARE_MF(get_len, (const std::string& x))

RPC_DECLARE_MF(squareVec, (vint_vec& x))

RPC_DECLARE_MF(multiVec, (vint_vec& z, vint_vec& x, vint_vec& y))

};

RPC_TYPEDEF_PTR(AsyncInterface);

// more convenient way than AsyncInterface...

BEGIN_RPC_INTERFACE(SampleRPC_Interface2, SessionScope)

RPC_ADD_MF(get_val)

RPC_ADD_MF(get_len)

END_RPC_ADD_MF()

RPC_DECLARE_MF(get_val, (rpc_in<int> x))

RPC_DECLARE_MF(get_len, (const std::string& x))

END_RPC_INTERFACE()

// async_client.cpp

#include <iostream>
#include <febird/rpc/client.h>
#include <febird/io/SocketStream.h>

using namespace std;
using namespace febird;
using namespace febird::rpc;

#include "../test.h"
#pragma comment(lib, "Ws2_32.lib")

void printVec(const vint_vec& vec) {
	for (int i = 0; i != vec.size(); ++i) {
		cout << "vec[" << i << "]=" << vec[i] << "/n";
	}
}
class AsyncImpl : public AsyncInterface {
public:
	AsyncImpl() {
		this->multiVec.set_async_callback(&AsyncImpl::on_multiVec);
	}
private:
	void on_multiVec(const client_packet_base& packet, vint_vec& z, vint_vec& x, vint_vec& y) {
		printf("AsyncImpl::on_multiVec/n");
		printf("ret=%u, z=%u, x=%u, y=%u/n", packet.retv, z.size(), x.size(), y.size());
	}
};
RPC_TYPEDEF_PTR(AsyncImpl);

int main0(int argc, char* argv[])
try {
	auto_ptr<SocketStream> cs(ConnectSocket("127.0.0.1:8001"));
	rpc_client<PortableDataInput, PortableDataOutput> client(cs.get());

	AsyncImplPtr obj1;
	SampleRPC_Interface2Ptr obj2;

	client.create(obj1, "obj1");
	client.create(obj2, "obj2");

	int ret;
	ret = client.retrieve(obj2, "obj2");

	rpc_ret_t val = obj1->get_val(100);
	cout << "obj1->get_val(100)=" << val << "/n";

	val = obj1->get_len("hello, world!");
	cout << "obj1->get_len(/"hello, world!/")=" << val << "/n";

	std::vector<unsigned> vec;
	vec.push_back((1));
	vec.push_back((2));
	vec.push_back((3));
	vec.push_back((4));
	vec.push_back((11));
	vec.push_back((22));
	vec.push_back((33));
	vec.push_back((44));

	val = obj1->squareVec(vec);
	printVec(vec);
	val = obj1->squareVec(vec);
	printVec(vec);

	std::vector<unsigned> vec2;
	for (int i = 0; i != vec.size(); ++i) {
		vec2.push_back(i + 1);
	}
//	obj1->multiVec.on_return = &AsyncInterface::on_multiVec;
	for (int i = 0; i < 5; ++i) {
		std::vector<unsigned> vec3;
		obj1->multiVec.async(vec3, vec, vec2);
		cout << "obj1->multiVec(vec3, vec, vec2) = " << val << "/n";
		printVec(vec3);
	}
	client.wait_pending_async();
	return 0;
}
catch (const std::exception& exp) {
	printf("exception: what=%s/n", exp.what());
	return 1;
}

int main(int argc, char* argv[]) {
#if defined(_WIN32) || defined(_WIN64)
	WSADATA information;
	WSAStartup(MAKEWORD(2, 2), &information);
	int ret = main0(argc, argv);
	WSACleanup();
	return ret;
#else
	return main0(argc, argv);
#endif
}

// async_client.cpp

#include <iostream>

#include <febird/rpc/client.h>

#include <febird/io/SocketStream.h>

using namespace std;

using namespace febird;

using namespace febird::rpc;

#include "../test.h"

#pragma comment(lib, "Ws2_32.lib")

void printVec(const vint_vec& vec) {

for (int i = 0; i != vec.size(); ++i) {

cout << "vec[" << i << "]=" << vec[i] << "/n";

}

class AsyncImpl : public AsyncInterface {

public:

AsyncImpl() {

this->multiVec.set_async_callback(&AsyncImpl::on_multiVec);

}

private:

void on_multiVec(const client_packet_base& packet, vint_vec& z, vint_vec& x, vint_vec& y) {

printf("AsyncImpl::on_multiVec/n");

printf("ret=%u, z=%u, x=%u, y=%u/n", packet.retv, z.size(), x.size(), y.size());

}

};

RPC_TYPEDEF_PTR(AsyncImpl);

int main0(int argc, char* argv[])

try {

auto_ptr<SocketStream> cs(ConnectSocket("127.0.0.1:8001"));

rpc_client<PortableDataInput, PortableDataOutput> client(cs.get());

AsyncImplPtr obj1;

SampleRPC_Interface2Ptr obj2;

client.create(obj1, "obj1");

client.create(obj2, "obj2");

int ret;

ret = client.retrieve(obj2, "obj2");

rpc_ret_t val = obj1->get_val(100);

cout << "obj1->get_val(100)=" << val << "/n";

val = obj1->get_len("hello, world!");

cout << "obj1->get_len(/"hello, world!/")=" << val << "/n";

std::vector<unsigned> vec;

vec.push_back((1));

vec.push_back((2));

vec.push_back((3));

vec.push_back((4));

vec.push_back((11));

vec.push_back((22));

vec.push_back((33));

vec.push_back((44));

val = obj1->squareVec(vec);

printVec(vec);

val = obj1->squareVec(vec);

printVec(vec);

std::vector<unsigned> vec2;

for (int i = 0; i != vec.size(); ++i) {

vec2.push_back(i + 1);

}

// obj1->multiVec.on_return = &AsyncInterface::on_multiVec;

for (int i = 0; i < 5; ++i) {

std::vector<unsigned> vec3;

obj1->multiVec.async(vec3, vec, vec2);

cout << "obj1->multiVec(vec3, vec, vec2) = " << val << "/n";

printVec(vec3);

}

client.wait_pending_async();

return 0;

}

catch (const std::exception& exp) {

printf("exception: what=%s/n", exp.what());

return 1;

}

int main(int argc, char* argv[]) {

#if defined(_WIN32) || defined(_WIN64)

WSADATA information;

WSAStartup(MAKEWORD(2, 2), &information);

int ret = main0(argc, argv);

WSACleanup();

return ret;

#else

return main0(argc, argv);

#endif

}

// async_server.cpp

#include <febird/rpc/server.h>
#include <febird/io/SocketStream.h>
#include <iostream>

using namespace febird;
using namespace febird::rpc;

#include "../test.h"
#pragma comment(lib, "Ws2_32.lib")

// rpc 函数的返回值是 rpc_ret_t，主要是为了避免专门为 void 返回值的函数编写特化代码
// use macro for convenient
BEGIN_RPC_IMP_INTERFACE(SampleRPC_Imp1, AsyncInterface)
	rpc_ret_t get_val(rpc_in<int> x) {
		std::cout << "AsyncInterface::get_val(rpc_in<int> x=" << x.r << ")/n";
		return x.r;
	}
	rpc_ret_t get_len(const std::string& x) {
		std::cout << "AsyncInterface::get_len(const std::string& x=/"" << x << "/")/n";
		return x.size();
	}
	rpc_ret_t squareVec(vint_vec& x) {
		for (vint_vec::iterator i = x.begin(); i != x.end(); ++i) {
			*i *= *i;
		}
		return x.size();
	}
	rpc_ret_t multiVec(vint_vec& z, vint_vec& x, vint_vec& y) {
		z.clear();
		for (int i = 0; i != x.size(); ++i) {
			z.push_back(x[i] * y[i]);
		}
		return 0x12345678;
	}
END_RPC_IMP_INTERFACE()

// don't use macro for more control
class SampleRPC_Imp2 : public SampleRPC_Interface2 {
	rpc_ret_t get_val(rpc_in<int> x) {
		std::cout << BOOST_CURRENT_FUNCTION << "x=" <<x.r << "/n";
		return x.r;
	}
	rpc_ret_t get_len(const std::string& x) {
		std::cout << BOOST_CURRENT_FUNCTION << "x=" << x << "/n";
		return x.size();
	}
public:
	// if use macro, such as SampleRPC_Imp1, app can not use custom create
	static remote_object* create() {
		SampleRPC_Imp2* p = new SampleRPC_Imp2;
		// set p
		// ...
		return p;
	}
};

int main0(int argc, char* argv[]) {
	try {
		SocketAcceptor acceptor("0.0.0.0:8001");
		rpc_server<PortableDataInput, PortableDataOutput> server(&acceptor);

		// register rpc implementation class...
		RPC_SERVER_AUTO_CREATE(server, SampleRPC_Imp1);
		server.auto_create((SampleRPC_Imp2*)0, &SampleRPC_Imp2::create);

		server.start();
	}
	catch (const std::exception& exp) {
		printf("exception: what=%s/n", exp.what());
	}
	return 0;
}

int main(int argc, char* argv[]) {
#if defined(_WIN32) || defined(_WIN64)
	WSADATA information;
	WSAStartup(MAKEWORD(2, 2), &information);
	int ret = main0(argc, argv);
	WSACleanup();
	return ret;
#else
	return main0(argc, argv);
#endif
}

// async_server.cpp

#include <febird/rpc/server.h>

#include <febird/io/SocketStream.h>

#include <iostream>

using namespace febird;

using namespace febird::rpc;

#include "../test.h"

#pragma comment(lib, "Ws2_32.lib")

// rpc 函数的返回值是 rpc_ret_t，主要是为了避免专门为 void 返回值的函数编写特化代码

// use macro for convenient

BEGIN_RPC_IMP_INTERFACE(SampleRPC_Imp1, AsyncInterface)

rpc_ret_t get_val(rpc_in<int> x) {

std::cout << "AsyncInterface::get_val(rpc_in<int> x=" << x.r << ")/n";

return x.r;

}

rpc_ret_t get_len(const std::string& x) {

std::cout << "AsyncInterface::get_len(const std::string& x=/"" << x << "/")/n";

return x.size();

}

rpc_ret_t squareVec(vint_vec& x) {

for (vint_vec::iterator i = x.begin(); i != x.end(); ++i) {

*i *= *i;

}

return x.size();

}

rpc_ret_t multiVec(vint_vec& z, vint_vec& x, vint_vec& y) {

z.clear();

for (int i = 0; i != x.size(); ++i) {

z.push_back(x[i] * y[i]);

}

return 0x12345678;

}

END_RPC_IMP_INTERFACE()

// don't use macro for more control

class SampleRPC_Imp2 : public SampleRPC_Interface2 {

rpc_ret_t get_val(rpc_in<int> x) {

std::cout << BOOST_CURRENT_FUNCTION << "x=" <<x.r << "/n";

return x.r;

}

rpc_ret_t get_len(const std::string& x) {

std::cout << BOOST_CURRENT_FUNCTION << "x=" << x << "/n";

return x.size();

}

public:

// if use macro, such as SampleRPC_Imp1, app can not use custom create

static remote_object* create() {

SampleRPC_Imp2* p = new SampleRPC_Imp2;

// set p

// ...

return p;

}

};

int main0(int argc, char* argv[]) {

try {

SocketAcceptor acceptor("0.0.0.0:8001");

rpc_server<PortableDataInput, PortableDataOutput> server(&acceptor);

// register rpc implementation class...

RPC_SERVER_AUTO_CREATE(server, SampleRPC_Imp1);

server.auto_create((SampleRPC_Imp2*)0, &SampleRPC_Imp2::create);

server.start();

}

catch (const std::exception& exp) {

printf("exception: what=%s/n", exp.what());

}

return 0;

}

int main(int argc, char* argv[]) {

#if defined(_WIN32) || defined(_WIN64)

WSADATA information;

WSAStartup(MAKEWORD(2, 2), &information);

int ret = main0(argc, argv);

WSACleanup();

return ret;

#else

return main0(argc, argv);

#endif

}

主要思路就是在 Server 端，用宏生成函数声明和工厂代码，把类注册到工厂中，启动server后，就可以自动接收client端的函数调用了。client发来的函数调用请求会派发到server端的用户定义函数体。当然，client端没有用户定义的函数体。

在client端，同样名字的宏，生成了和server端不同的代码，这些代码生成stub函数，把函数调用的参数序列化到网络。

其中主要使用了boost::tuple，和 boost::any，来生成stub的虚函数。

这个实验很有趣，它说明了C++强大的表达能力，即使没有IDL，仅使用语言本身就可以实现RPC。

4 个回复

zm0011说道：

2008年2月9日下午10:38

在开发网络通讯程序中很有用，正想自己写一个，竟然百度出来了。顺便看了一下老兄的其它文章，数量不多但水平比较高，而且涉足的领域很多哦，从WIN到LINUX、从C++壳到数据库都有。希望能看到你奉献更多的好东东，谢谢！

登录以回复
whinah说道：

2008年2月23日下午3:32

客气、客气，这个RPC之后我也再没继续深入。打造一个稳定而强大的基础设施需要太大的精力。目前我们使用ICE，有兴趣可以去关心一下。前不久我在boost sandbox 中看到了一个rpc，没细看，估计在参数传递方面和我的设计思想差不多，不过它底层用的是boost.asio。

登录以回复
angkorcn说道：

2008年10月24日上午10:58

思路不错，不过说实话实在不喜欢宏这个东西，带来的问题太多。对于RPC我还是倾向于使用IDL编译，其代码生成能力是没有限制的，类型安全也可以做的更好。宏只是预处理，其代码生成能力一般，类型安全极差。用C++的模板作为代码生成器，能力很差，不过安全性好。除非在0x标准里有新的东西产生，不然还是倾向于IDL。

登录以回复
whinah说道：

2008年10月27日下午5:29

当然，使用 IDL 是普遍的用法，并且已经很成熟。宏，模板，成也萧何，败也萧何，用得好，就一切很OK，用得不好，诡异的行为和漫长的编译很够大家受的。
任何东西都有它的生存空间，IDL是不错，不但自动化，还是跨语言的，但是数数市面上的IDL，比较主流的就不下几十种，加上那些杂七杂八的，几百种都不止。
我这个东西做出来两年多了，一直都扔在那里没用，最近做一个小东西，用了一下，效果竟然很好。

登录以回复

发表评论

您必须登录后才能发表评论。