通过使用C++变参模板动态生成重载函数来解决Tars框架的消息转发问题

Fantastic

C++14有很多新特性,但是最让人着迷的,莫过于变参模板了,变参模板把函数式的编程思想在C++中发挥得淋漓尽致,让C++的灵活性提升了一个level,使用变参模板,可以写出看起来像是动态语言特性的代码.
最近遇到一个问题,如果没有变参模板,还真不知道该如何解决.

Question

我们的后台程序使用了Tars框架,他的IDL生成出来的C++代码中,有一个Callback类用来表示异步调用的回调,定义如下:

1
2
3
4
5
6
7
8
9
class FakeCallbackBase {
protected:
    virtual void callback_fakeRequestName(int a, const FakeResponse2 &a1, const FakeResponse& a2) {
        std::cout << "no implement" << std::endl;
    }
    virtual void callback_fakeRequestName_exception(int a, const FakeResponse2 &a1, const FakeResponse& a2) {
        std::cout << "no implement" << std::endl;
    }
};

一眼看上去似乎也没什么问题,每个请求对应一个虚函数,要处理这个请求的callback,只需要重载对应的虚函数即可.
就处理业务逻辑来说,其实没啥太大问题.
但是,我现在要写的是一个Proxy,主要负责把内部的Tars接口使用开放的HTTP协议对外暴露,所以对每一个接口的业务逻辑都是一样的.我可不想把同一份代码不停的Copy & Paste.
So… 我想到一个办法,变参模板!

Solution

分析上面的回调函数的定义,发现参数列表有如下规律:

1
第一个参数是RPC的返回值,类似于函数的返回值,只有一个,后面的所有参数表示的是RPC的出参,是一个*可变列表*

定义这样一个宏,方便用来给每一个RPC生成统一的Callback类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
#ifndef DEFINE_FORWARD_CALLBACK
#define DEFINE_FORWARD_CALLBACK(requestName) \
class ForwardFor_##requestName {     \
public:                              \
template<typename... TREQS> \
class Forward { \
public: \
    template<typename TR>\
    class Return {\
        public:\
        template<typename TF, typename TCONTEXT, typename... TRSPS> \
        class Callback : public TF {\
        public:                                                         \
            Callback(std::shared_ptr<typename ForwardHelper::Forward<TREQS...>::template Return<TR>::template CallbackProcesserBase<TCONTEXT, TRSPS...>> processer, const TCONTEXT& context, const TREQS&... tReqs) \
            : m_processer(processer), m_context(context) {                     \
                m_processer->onRequest(m_context, std::string(#requestName), tReqs...); \
            }                                                           \
            virtual void callback_##requestName(TR ret, TRSPS... rsps) { \
                m_processer->onCallback(m_context, std::string(#requestName), ret, rsps...); \
            }                                                           \
            virtual void callback_##requestName##_exception(TR ret) {\
                m_processer->onCallbackException(m_context, std::string(#requestName), ret); \
            }\
        private:\
            std::shared_ptr<typename ForwardHelper::Forward<TREQS...>::template Return<TR>::template CallbackProcesserBase<TCONTEXT, TRSPS...>> m_processer; \
            TCONTEXT m_context;                                         \
        };\
    private:\
        Return(){}\
        Return(const Return&){}\
    };\
private:\
    Forward(){}                                    \
    Forward(const ForwardFor_##requestName&){}                  \
};                                                              \
}
#endif

使用嵌套类是为了区分请求参数类型列表和回包(出参)参数类型列表
Forward类的模板参数是请求参数列表
Callback类的模板参数是回包参数列表
TCONTEXT是callback允许外带的一个上下文,方便发包和收包的逻辑衔接
TF是这个宏生成出来的类的基类,本例中就是上面出现的FakeCallbackBase
m_processer是用来统一处理所有回包的一个处理器,我们把静态的代码转换为动态的代码,靠的就是这个m_processer

下面来看一下Processer的代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
namespace ForwardHelper {
    template<typename... TREQS>
    class Forward {
    public:
        template<typename TR>
        class Return {
        public:
            template<typename TCONTEXT, typename... TRSPS>
            class CallbackProcesserBase {
            public:
                virtual void onRequest(TCONTEXT& m_context, const std::string& requestName, const TREQS&... tReqs) = 0;
                virtual void onCallbackException(TCONTEXT& m_context, const std::string& requestName, TR t) = 0;
                virtual void onCallback(TCONTEXT& m_context, const std::string& requestName, TR t, TRSPS... rsps) = 0;
            };
        };
    };
}

CallbackProcesserBase 就是Processer的基类,因为我们要实现的是HTTP的转发,所以使用一个派生类来实现以上的接口定义,代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
template<typename... TREQS>
class HTTPForward {
public:
    template<typename TR>
    class Return {
    public:
        template<typename... TRSPS>
        class MultiReqRspGenericHttpProcesser : public ForwardHelper::Forward<TREQS...>::template Return<TR>::template CallbackProcesserBase<tars::TarsCurrentPtr, TRSPS...> {
        public:
            virtual void onRequest(tars::TarsCurrentPtr &current, const std::string& requestName, const TREQS&... tReqs) {
                m_startTime = TNOWMS;
                getStringStreamByMultiParam(m_ossReq, tReqs...);
                FDLOG("protocol") << requestName << "|REQ|" << m_ossReq.str() << endl;
            }
            virtual void onCallbackException(tars::TarsCurrentPtr &current, const std::string& requestName, TR ret) {
                ERRORLOG << requestName << "|REQ|" << m_ossReq.str() << "|" << ret << "|" << getCostInMS() << endl;
            }
            void processCallback(std::vector<std::string> &bodies, std::ostringstream &oss) {
            }
            template<typename T>
            void processCallback(std::vector<std::string> &bodies, std::ostringstream &oss, T lastOneOfRSPS) {
                bodies.emplace_back();
                tarsEncode(lastOneOfRSPS, *bodies.rbegin());
                oss << lastOneOfRSPS;
            }
            template<typename T, typename... TRSPS2>
            void processCallback(std::vector<std::string> &bodies, std::ostringstream &oss, T oneOfRSPS,  TRSPS2... rsps) {
                bodies.emplace_back();
                tarsEncode(oneOfRSPS, *bodies.rbegin());
                oss << oneOfRSPS << "|";
                processCallback(bodies, oss, rsps...);
            }
            virtual void onCallback(tars::TarsCurrentPtr &current, const std::string& requestName, TR ret, TRSPS... rsps) {
                //打包rsps, 然后使用current对proxy的调用方回包,从而实现统一的转发
                std::ostringstream ossRsp;  //日志使用
                ResponseModern rsp;
                processCallback(rsp.outParams, ossRsp, rsps...); //这一行对rsps进行拆包,然后一个一个Encode到一个std::vector<std::string>(也就是rsp.outParams)中
                rsp.result.retCode = ret;
                string body;
                tarsEncode(rsp,body); //最后把最外层的rsp Encode,然后直接http返回
                TC_HttpResponse httpResp;
                httpResp.setContentType("application/octet-stream;charset=UTF-8");
                httpResp.setCacheControl("no-cache");
                httpResp.setResponse(200, "OK", body);
                body = httpResp.encode();
                if( !current->isResponse())
                {
                    current->sendResponse(body.data(),body.size());
                    FDLOG("protocol")<< requestName<< "|REQ|" << m_ossReq.str() <<
                        "|RSP|" << ossRsp.str() << "|"<< ret <<"|"<< getCostInMS() << "|" << body.size() << endl;
                }
            }
            template<typename TFORWARDFOR, typename TF, typename TCONTEXT>
            static TF* makeCallback(const TCONTEXT& context, const TREQS&... treqs) {
                //为了避免多次填写模板参数,方便生成callback对象,所以使用了一个独立的静态工厂函数,实现模板参数的推导
                auto pThis = std::make_shared<typename HTTPForward<TREQS...>::template Return<TR>::template MultiReqRspGenericHttpProcesser<TRSPS...>>();
                TF* base = new typename TFORWARDFOR::template Forward<TREQS...>::template Return<TR>::template Callback<TF, TCONTEXT, TRSPS...>(
                    std::dynamic_pointer_cast<typename ForwardHelper::Forward<TREQS...>::template Return<TR>::template CallbackProcesserBase<TCONTEXT, TRSPS...>>(pThis),
                    context,
                    treqs...);
                return base;
            }
    protected:
            int64_t m_startTime;
            std::ostringstream m_ossReq;
            int64_t getCostInMS() const {
                return TNOWMS - m_startTime;
            }
        };
    };
};

最后,通过变参模板,我实现了一个接口只需要两行代码即可进行转发,一行代码是使用DEFINE_FORWARD_CALLBACK 宏生成一个callback的基类, 一行代码是用来表示这个接口的返回值,入参和出参,代码长相如下:

CASE宏定义,简化最终代码

1
2
3
4
5
6
7
8
#ifndef CASE
#define CASE(exp)                               \
    {                                           \
        if (exp) {                              \
            return ret;                         \
        }                                       \
    }
#endif

定义一个接口的转发代码

1
2
3
4
CASE((Dispatcher<FakeServerPrx::Callback, GET_FORWARDER(fakeRequestName), tars::TarsCurrentPtr>
::REQ<int64_t, std::string>   //入参列表,支持0-N个任意类型的参数
::RSP<const FakeResponse2&, const FakeResponse&>  //出参列表,0-N个任意类型的回包
::tryDispatch(current, iterPrx->second, req.header.funcName, GETSTR(getProfile), req.params)));

Dispatcher 的主要逻辑是把一个std::vector的请求参数列表解包成其对应类型的Tars结构体,然后日志记录并且向下进行转发,也使用了变参模板的特性.这里就不赘述了.

后面,我们可以根据.tars文件定义的RPC接口,来动态生成转发代码,从而实现Proxy服务全自动生成.对外暴露接口再也没有任何工作量了!

Notes

  • 我觉得写代码的一大乐趣,就是把复杂的问题简单化,复杂的事情只做一次.尽可能的减少对外暴露的复杂度,就本例来说,基本上算是做到了.
  • 以上代码使用了变参模板的递归,和tuple递归,这里的思想其实是和erlang这种函数式语言的for循环一样,使用这种看似很奇怪的写法,是因为模板里面没有变量,只有常量.
  • Dispatcher中使用了把变参模板暂存为一个tuple,然后再unpacking, 需要使用另一个函数来中转, 这种写法还是很奇怪,很hacking, 不过似乎也没有其他的写法了.参考这里.
  • 核心思想是:
    1. 把各种类型的Callback通过宏自动生成
    2. 找一个地方汇总所有的Callback的结果然后统一处理
赏

谢谢你请我吃糖

微信

扫一扫，分享到微信

微信分享二维码