thrift的asio实现

sevencat

thrift是目前能找到的轻量级，高性能，非常灵活的一个rpc。
官方的实现是用libevent2+多线程做为网络接口，网上也有人有了asio的实现。

写了个简单的
#pragma once
#include "Api.h"
#include <iostream>
#include <stdexcept>
#include <sstream>
#include "rpc_conn.h"

using namespace apache::thrift;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;

class RpcJgServer : virtual public jgrpc::ApiIf
{
public:
    explicit RpcJgServer(const std::string& address, int port,std::size_t thread_pool_size);

    void run();

    void stop();

private:
    void start_accept();
    void handle_accept(RpcConnectionPtr conn,const asio::error_code& e);

    std::size_t thread_pool_size_;
    asio::io_service io_service_;

    asio::ip::tcp::acceptor acceptor_;

    boost::shared_ptr<TProcessor> processor_;
    boost::shared_ptr<TProtocolFactory> input_protocol_factory_;
    boost::shared_ptr<TProtocolFactory> output_protocol_factory_;
    boost::shared_ptr<jgrpc::ApiIf> evt_If;

public:
    std::set<RpcConnectionPtr> clients_;//已登录的客户端
    asio::detail::mutex mutex_;//是一把锁

public:
    void jgrpc::ApiIf::ping(void){};
};
#include "stdafx.h"
#include "rpc_jg.h"

RpcJgServer::RpcJgServer(const std::string& address, int port,std::size_t thread_pool_size)
               : thread_pool_size_(thread_pool_size),
               acceptor_(io_service_, asio::ip::tcp::endpoint(asio::ip::address_v4::from_string(address), port))
{
    boost::shared_ptr<TProtocolFactory> protocolFactory(new TBinaryProtocolFactory());
    input_protocol_factory_=protocolFactory;
    output_protocol_factory_=protocolFactory;
    evt_If.reset(this);
    processor_.reset(new jgrpc::ApiProcessor(evt_If));
    start_accept();
}

void RpcJgServer::start_accept()
{
    RpcConnectionPtr new_connection;
    new_connection.reset(new RpcConnection(io_service_, processor_,input_protocol_factory_, output_protocol_factory_));
    acceptor_.async_accept(new_connection->socket(),
        boost::bind(&RpcJgServer::handle_accept, this,
        new_connection,
        asio::placeholders::error));   
}
void RpcJgServer::run()
{
    std::vector<boost::shared_ptr<asio::thread> > threads;
    for (std::size_t i = 0; i < thread_pool_size_; ++i)
    {
        boost::shared_ptr<asio::thread> thread(new asio::thread(
            boost::bind(&asio::io_service::run, &io_service_)));
        threads.push_back(thread);
    }

    for (std::size_t i = 0; i < threads.size(); ++i) {
        threads->join();
    }
}

void RpcJgServer::stop()
{
    io_service_.stop();
}

void RpcJgServer::handle_accept(RpcConnectionPtr conn,const asio::error_code& e)
{
    if (!e)
    {
        //这里进行处理
        conn->start();
    }
    start_accept();   
}

1. #pragma once
3. using namespace apache::thrift;
4. using namespace apache::thrift::protocol;
5. using namespace apache::thrift::transport;
7. class RpcConnection  : public boost::enable_shared_from_this<RpcConnection>,private boost::noncopyable
8. {
9. public:
10.         explicit RpcConnection(asio::io_service& io_service,
11.                 boost::shared_ptr<TProcessor> processor,
12.                 boost::shared_ptr<TProtocolFactory> inputProtocolFactory,
13.                 boost::shared_ptr<TProtocolFactory> outputProtocolFactory);
15.         asio::ip::tcp::socket& socket()
16.         {
17.                 return socket_;
18.         }
19.         void start()
20.         {
21.                 asio::async_read(socket_, asio::buffer(length_),
22.                         strand_.wrap(
23.                         boost::bind(&RpcConnection::handle_read_length, shared_from_this(),
24.                         asio::placeholders::error)));
25.         }
27.         static RpcConnection *get_current_conn();
29. private:
30.         void handle_read_length(const asio::error_code& e);
31.         void handle_read_frame(int32_t sz, const asio::error_code& e);
32.         void handle_write(const asio::error_code& e);
33.         asio::io_service::strand strand_;
34.         asio::ip::tcp::socket socket_;
35.         boost::array<uint8_t, 4> length_;
36.         boost::shared_ptr<TProcessor> processor_;
37.         boost::shared_ptr<TProtocolFactory> inputProtocolFactory_;
38.         boost::shared_ptr<TProtocolFactory> outputProtocolFactory_;
39.         std::vector<uint8_t> frame_;
40.         asio::io_service& io_service_;
41.        
42. };
44. typedef boost::shared_ptr<RpcConnection> RpcConnectionPtr;

复制代码

1. #include "stdafx.h"
2. #include "rpc_conn.h"
3. #include <boost/thread/tss.hpp>
5. static boost::thread_specific_ptr<RpcConnection> tls_sessiondata;
6. static boost::shared_ptr<boost::detail::tss_cleanup_function> tls_cleanup;
8. RpcConnection *RpcConnection::get_current_conn()
9. {
10.         RpcConnection *cursession=tls_sessiondata.get();
11.         return cursession;
12. }
14. RpcConnection::RpcConnection(asio::io_service& io_service,
15.                                            boost::shared_ptr<TProcessor> processor,
16.                                            boost::shared_ptr<TProtocolFactory> inputProtocolFactory,
17.                                            boost::shared_ptr<TProtocolFactory> outputProtocolFactory
18.                                            )
19.                                            : strand_(io_service),
20.                                            socket_(io_service),
21.                                            processor_(processor),
22.                                            inputProtocolFactory_(inputProtocolFactory),
23.                                            outputProtocolFactory_(outputProtocolFactory),
24.                                            io_service_(io_service)
25. {
26. }
28. void RpcConnection::handle_read_length(const asio::error_code& e)
29. {
30.         if (!e)
31.         {
32.                 int32_t sz =
33.                         ((length_[0] & 0xff) << 24) |
34.                         ((length_[1] & 0xff) << 16) |
35.                         ((length_[2] & 0xff) <<  8) |
36.                         ((length_[3] & 0xff));
38.                 if (sz < 0)
39.                 {
40.                         std::cerr << "Frame length is negative. Are you sure your client uses a TFramedTransport?" << std::endl;
41.                         delete this;
42.                 }
43.                 else if (sz == 0)
44.                 {
45.                         std::cerr << "Frame length is zero, weird" << std::endl;
46.                         delete this;
47.                 }
48.                 else
49.                 {
50.                         frame_.resize(sz);
51.                         asio::async_read(socket_,
52.                                 asio::buffer(frame_, sz),
53.                                 strand_.wrap(
54.                                 boost::bind(&RpcConnection::handle_read_frame, shared_from_this(), sz,asio::placeholders::error)));
55.                 }
56.         }
57. }
59. //放到tls里面
60. void RpcConnection::handle_read_frame(int32_t sz, const asio::error_code& e)
61. {
62.         if (!e)
63.         {
64.                 boost::detail::set_tss_data(&tls_sessiondata,tls_cleanup,this,false);
65.                 boost::shared_ptr<TMemoryBuffer> inputTransport(new TMemoryBuffer(&frame_[0], sz));
66.                 boost::shared_ptr<TProtocol> inputProtocol = inputProtocolFactory_->getProtocol(inputTransport);
68.                 boost::shared_ptr<TMemoryBuffer> outputTransport(new TMemoryBuffer);
69.                 boost::shared_ptr<TFramedTransport> outputFramedTransport(new TFramedTransport(outputTransport));
70.                 boost::shared_ptr<TProtocol> outputProtocol = outputProtocolFactory_->getProtocol(outputFramedTransport);
72.                 processor_->process(inputProtocol, outputProtocol,NULL);
74.                 uint32_t outframeSize;
75.                 uint8_t *outframe;
77.                 outputTransport->getBuffer(&outframe, &outframeSize);
79.                 asio::async_write(socket_,
80.                         asio::buffer(outframe, outframeSize),
81.                         strand_.wrap(boost::bind(&RpcConnection::handle_write, shared_from_this(), asio::placeholders::error)));
83.         }
84. }
86. void RpcConnection::handle_write(const asio::error_code& e)
87. {
88.         if (!e)
89.         {
90.                 asio::async_read(socket_, asio::buffer(length_),
91.                         strand_.wrap(boost::bind(&RpcConnection::handle_read_length, shared_from_this(), asio::placeholders::error)));
92.         }
93. }

复制代码

sevencat

没有实现servercontext接口，因为在这个框架下你可以直接通过tls直接获取connection。
本代码没有经过哪怕简单的测试。

sevencat

注：TFramedTransport这个东东其实就是在传输前加上一个长度。所以代码中读的时候先读一个整数，写的时候就直接调用的
boost::shared_ptr<TFramedTransport> outputFramedTransport(new TFramedTransport(outputTransport));