当前位置:编程学习 > C/C++ >>

C++多进程并发框架

三年来一直从事服务器程序开发,一直都是忙忙碌碌,不久前结束了职业生涯的第一份工作,有了一个礼拜的休息时间,终于可以写写总结了。于是把以前的开源代码做了整理和优化,这就是FFLIB。虽然这边总结看起来像日记,有很多废话,但是此文仍然是有很大针对性的。针对服务器开发中常见的问题,如多线程并发、消息转发、异步、性能优化、单元测试,提出自己的见解。
面对的问题
从事开发工程中,遇到过不少问题,很多时候由于时间紧迫,没有使用优雅的方案。在跟业内的一些朋友交流过程中,我也意识到有些问题是大家都存在的。简单列举如下:
●多线程与并发
●异步消息/接口调用
●消息的序列化与Reflection
●性能优化
●单元测试
多线程与并发
现在是多核时代,并发才能实现更高的吞吐量、更快的响应,但也是把双刃剑。总结如下几个用法:
●多线程+显示锁;接口是被多线程调用的,当被调用时,显示加锁,再操作实体数据。悲剧的是,工程师为了优化会设计多个锁,以减少锁的粒度,甚至有些地方使用了原子操作。这些都为领域逻辑增加了额外的设计负担。最坏的情况是会出现死锁。
●多线程+任务队列;接口被多线程调用,但请求会被暂存到任务队列,而任务队列会被单线程不断执行,典型生产者消费者模式。它的并发在于不同的接口可以使用不同的任务队列。这也是我最常用的并发方式。
这是两种最常见的多线程并发,它们有个天生的缺陷——Scalability。一个机器的性能总是有瓶颈的。两个场景的逻辑虽然由多个线程实现了并发,但是运算量十分有可能是一台机器无法承载的。如果是多进程并发,那么可以分布式把其部署到其他机器(也可部署在一台机器)。所以多进程并发比多线程并发更加Scalability。另外采用多进程后,每个进程单线程设计,这样的程序更加Simplicity。多进程的其他优点如解耦、模块化、方便调试、方便重用等就不赘言了。
异步消息/接口调用
提到分布式,就要说一下分布式的通讯技术。常用的方式如下:
●类RPC;包括WebService、RPC、ICE等,特点是远程同步调用。远程的接口和本地的接口非常相似。但是游戏服务器程序一般非常在意延迟和吞吐量,所以这些阻塞线程的同步远程调用方式并不常用。但是我们必须意识到他的优点,就是非常利于调用和测试。
●全异步消息;当调用远程接口的时候,异步发送请求消息,接口响应后返回一个结果消息,调用方的回调函数处理结果消息继续逻辑操作。所以有些逻辑就会被切割成ServiceStart和ServiceCallback两段。有时异步会讲领域逻辑变得支离破碎。另外消息处理函数中一般会写一坨的switch/case 处理不同的消息。最大的问题在于单元测试,这种情况传统单元测试根本束手无策。
消息的序列化与Reflection
实现消息的序列化和反序列化的方式有很多,常见的有Struct、json、Protobuff等都有很成功的应用。我个人倾向于使用轻量级的二进制序列化,优点是比较透明和高效,一切在掌握之中。在FFLIB 中实现了bin_encoder_t 和 bin_decoder_t 轻量级的消息序列化,几十行代码而已。
性能优化
已经写过关于性能方面的总结,参见
http://www.cnblogs.com/zhiranok/archive/2012/06/06/cpp_perf.html
有的网友提到profiler、cpuprofiler、callgrind等工具。这些工具我都使用过,说实话,对于我来说,我太认同它有很高的价值。第一他们只能用于开发测试阶段,可以初步得到一些性能上参考数据。第二它们如何实现跟踪人们无从得知。运行其会使程序变慢,不能反映真实数据。第三重要的是,开发测试阶段性能和上线后的能一样吗?Impossible !
关于性能,原则就是数据说话,详见博文,不在赘述。
单元测试
关于单元测试,前边已经谈论了一些。游戏服务器程序一般都比较庞大,但是不可思议的是,鄙人从来没见有项目(c++ 后台架构的)有完整单元测试的。由于存在着异步和多线程,传统的单元测试框架无法胜任,而开发支持异步的测试框架又是不现实的。我们必须看到的是,传统的单元测试框架已经取得了非常大的成功。据我了解,使用web 架构的游戏后台已经对于单元测试的使用已经非常成熟,取得了极其好的效果。所以我的思路是利用现有的单元测试框架,将异步消息、多线程的架构做出调整。
已经多次谈论单元测试了。其实在开发FFLIB的思路很大程度来源于此,否则可能只是一个c++ 网络库而已。我决定尝试去解决这个问题的时候,把FFLIB 定位于框架。
先来看一段非常简单的单元测试的代码 :
1
Assert(2 == Add(1, 1));
请允许我对这行代码做些解释,对Add函数输入参数,验证返回值是否是预期的结果。这不就是单元测试的本质吗?在想一下我们异步发送消息的过程,如果每个输入消息约定一个结果消息包,每次发送请求时都绑定一个回调函数接收和验证结果消息包。这样的话就恰恰满足了传统单元测试的步骤了。最后还需解决一个问题,Assert是不能处理异步的返回值的。幸运的是,future机制可以化异步为同步。不了解future 模式的可以参考这里:
http://blog.chinaunix.net/uid-23093301-id-190969.html
http://msdn.microsoft.com/zh-cn/library/dd764564.aspx#Y300
来看一下在FFLIB框架下远程调用echo 服务的示例:

struct lambda_t
{
  static void callback(echo_t::out_t& msg_)
  {
    echo_t::in_t in;
    in.value = "XXX_echo_test_XXX";
    singleton_t<msg_bus_t>::instance()
       .get_service_group("echo")
       ->get_service(1)->async_call(in, &lambda_t::callback);
  }
};
echo_t::in_t in;
in.value = "XXX_echo_test_XXX";
singleton_t<msg_bus_t>::instance().get_service_group("echo")->get_service(1)->async_call(in, &lambda_t::callback);
当需要调用远程接口时,async_call(in, &lambda_t::callback); 异步调用必须绑定一个回调函数,回调函数接收结果消息,可以触发后续操作。这样的话,如果对echo 的远程接口做单元测试,可以这样做:

rpc_future_t< echo_t::out_t> rpc_future;
echo_t::in_t in;
in.value = "XXX_echo_test_XXX";
const echo_t::out_t& out = rpc_future.call(
    singleton_t<msg_bus_t>::instance()
        .get_service_group("echo")->get_service(1), in);
Assert(in.value == out.value);
这样所有的远程接口都可以被单元测试覆盖。
FFLIB 介绍
FFLIB 结构图
C++多进程并发框架

如图所示,Client 不会直接和Service 相连接,而是通过Broker 中间层完成了消息传递。关于Broker 模式可以参见:http://blog.chinaunix.net/uid-23093301-id-90459.html
进程间通信采用TPC,而不是多线程使用的共享内存方式。Service 一般是单线程架构的,通过启动多进程实现相对于多线程的并发。由于Broker模式天生石分布式的,所以有很好的Scalability。
消息时序图
C++多进程并发框架

如何注册服务和接口
来看一下Echo 服务的实现:

struct echo_service_t
{
public:
    void echo(echo_t::in_t& in_msg_, rpc_callcack_t<echo_t::out_t>& cb_)
    {
        logtrace((FF, "echo_service_t::echo done value<%s>", in_msg_.value.c_str()));
        echo_t::out_t out;
        out.value = in_msg_.value;
        cb_(out);
    }
};
 
int main(int argc, char* argv[])
{
    int g_index = 1;
    if (argc > 1)
    {
        g_index = atoi(argv[1]);
    }
    char buff[128];
    snprintf(buff, sizeof(buff), "tcp://%s:%s", "127.0.0.1", "10241");
 
    msg_bus_t msg_bus;
    assert(0 == singleton_t<msg_bus_t>::instance().open("tcp://127.0.0.1:10241") && "can't connnect to broker");
 
    echo_service_t f;
 
    singleton_t<msg_bus_t>::instance().create_service_group("echo");
    singleton_t<msg_bus_t>::instance().create_service("echo", g_index)
            .bind_service(&f)
            .reg(&echo_service_t::echo);
 

补充:软件开发 , C++ ,
CopyRight © 2012 站长网 编程知识问答 www.zzzyk.com All Rights Reserved
部份技术文章来自网络,