网络版汉译英服务（muduo）

网络版汉译英服务（muduo）

项目源码：汉译英服务

muduo库

muduo 库是什么

Muduo 由陈硕大佬开发，是一个基于非阻塞 IO 和事件驱动的 C++高并发 TCP 网络编程库。 它是一款基于主从 Reactor 模型的网络库，其使用的线程模型是 one loop per thread, 所谓 one loop per thread 指的是：

• 一个线程只能有一个事件循环（EventLoop）， 用于响应计时器和 IO 事件
• 一个文件描述符只能由一个线程进行读写，换句话说就是一个 TCP 连接必须归属 于某个 EventLoop 管理

muduo 库常见接口介绍

muduo::net::EventLoop

class EventLoop : noncopyable 
{ 
    public:  
    //开始永久事件循环
    void loop(); 
    //停止循环
    void quit(); 
    TimerId runAt(Timestamp time, TimerCallback cb); 
    TimerId runAfter(double delay, TimerCallback cb); 
    TimerId runEvery(double interval, TimerCallback cb); 
    void cancel(TimerId timerId); 
    private: 
    std::atomic<bool> quit_; 
    std::unique_ptr<Poller> poller_; 
    mutable MutexLock mutex_; 
    std::vector<Functor> pendingFunctors_ GUARDED_BY(mutex_); 
}; 

muduo::net::TcpConnection

class TcpConnection : noncopyable, 
public 
    std::enable_shared_from_this<TcpConnection> 
{ 
    public:  
    TcpConnection(EventLoop* loop, 
                  const string& name, 
                  int sockfd, 
                  const InetAddress& localAddr, 
                  const InetAddress& peerAddr); 
    //判断是否在连接状态
    bool connected() const { return state_ == kConnected; } 
    bool disconnected() const { return state_ == kDisconnected; }  
    //发送数据
    void send(string&& message); // C++11 
    void send(const void* message, int len); 
    void send(const StringPiece& message); 
    // void send(Buffer&& message); // C++11 
    void send(Buffer* message); // this one will swap data 
    //关闭连接
    void shutdown(); // NOT thread safe, no simultaneous calling 

    void setContext(const boost::any& context) 
    { context_ = context; } 

    const boost::any& getContext() const 
    { return context_; } 

    boost::any* getMutableContext() 
    { return &context_; } 
	//设置回调函数
    void setConnectionCallback(const ConnectionCallback& cb) 
    { connectionCallback_ = cb; } 

    void setMessageCallback(const MessageCallback& cb) 
    { messageCallback_ = cb; } 
    private: 
    enum StateE { kDisconnected, kConnecting, kConnected, 
                 kDisconnecting }; 
    EventLoop* loop_; 
    ConnectionCallback connectionCallback_; 
    MessageCallback messageCallback_; 
    WriteCompleteCallback writeCompleteCallback_; 
    boost::any context_; 
}; 

muduo::net::TcpServer

typedef std::shared_ptr<TcpConnection> TcpConnectionPtr; 
typedef std::function<void (const TcpConnectionPtr&)> 
    ConnectionCallback; 
typedef std::function<void (const TcpConnectionPtr&, Buffer*, Timestamp)> MessageCallback; 

class InetAddress : public muduo::copyable 
{ 
public: 
    InetAddress(StringArg ip, uint16_t port, bool ipv6 = false); 
}; 

class TcpServer : noncopyable 
{ 
public: 
    //端口是否复用
    enum Option 
    { 
        kNoReusePort, 
        kReusePort, 
    }; 

    TcpServer(EventLoop* loop, 
              const InetAddress& listenAddr, 
              const string& nameArg, 
              Option option = kNoReusePort);
    //设置线程数量
    void setThreadNum(int numThreads); 
    //服务器启动函数
    void start(); 
    /// 当一个新连接建立成功的时候被调用 
    void setConnectionCallback(const ConnectionCallback& cb) 
    { connectionCallback_ = cb; } 
    /// 消息的业务处理回调函数---这是收到新连接消息的时候被调用的函数 
    void setMessageCallback(const MessageCallback& cb) 
    { messageCallback_ = cb; } 
}; 

muduo::net::TcpClient

class TcpClient : noncopyable 
{ 
    public: 
    // TcpClient(EventLoop* loop); 
    // TcpClient(EventLoop* loop, const string& host, uint16_t port); 
    TcpClient(EventLoop* loop, 
              const InetAddress& serverAddr, 
              const string& nameArg); 
    ~TcpClient(); // force out-line dtor, for std::unique_ptr members. 
    //连接服务器
    void connect(); 
    void disconnect();
    //关闭连接
    void stop(); 

    //获取客户端对应的通信连接 Connection 对象的接口，发起 connect 后，有可能还没有连接建立成功 
    TcpConnectionPtr connection() const 
    { 
        MutexLockGuard lock(mutex_); 
        return connection_; 
    } 
    /// 连接服务器成功时的回调函数 
    void setConnectionCallback(ConnectionCallback cb) 
    { connectionCallback_ = std::move(cb); } 
    /// 收到服务器发送的消息时的回调函数 
    void setMessageCallback(MessageCallback cb) 
    { messageCallback_ = std::move(cb); } 
    private: 
    EventLoop* loop_; 
    ConnectionCallback connectionCallback_; 
    MessageCallback messageCallback_; 
    WriteCompleteCallback writeCompleteCallback_; 
    TcpConnectionPtr connection_ GUARDED_BY(mutex_); 
}; 

/* 
需要注意的是，因为 muduo 库不管是服务端还是客户端都是异步操作， 
对于客户端来说如果我们在连接还没有完全建立成功的时候发送数据，这是不被允
许的。 
因此我们可以使用内置的 CountDownLatch 类进行同步控制 
*/ 
class CountDownLatch : noncopyable 
{ 
    public: 
    explicit CountDownLatch(int count);
    //配合client的connect()函数使用，在连接还没有建立好时进行wait
    //直到client的连接回调函数被执行时，使用countDown()函数解除等待状态
    void wait(){ 
        MutexLockGuard lock(mutex_); 
        while (count_ > 0) 
        { 
            condition_.wait();  } 
    } 
    void countDown(){ 
        MutexLockGuard lock(mutex_); 
        --count_; 
        if (count_ == 0) 
        { 
            condition_.notifyAll(); 
        } 
    } 
    int getCount() const; 
    private: 
    mutable MutexLock mutex_; 
    Condition condition_ GUARDED_BY(mutex_); 
    int count_ GUARDED_BY(mutex_); 
}; 

muduo::net::Buffer

class Buffer : public muduo::copyable 
{ 
    public: 
    static const size_t kCheapPrepend = 8; 
    static const size_t kInitialSize = 1024; 
    explicit Buffer(size_t initialSize = kInitialSize) 
        : buffer_(kCheapPrepend + initialSize), 
    readerIndex_(kCheapPrepend), 
    writerIndex_(kCheapPrepend); 
        //反序列化
        void retrieve(size_t len) 
        void retrieveInt64() 
        void retrieveInt32() 
        void retrieveInt16() 
        void retrieveInt8() 
        string retrieveAllAsString() 
        string retrieveAsString(size_t len)  
        private: 
    std::vector<char> buffer_; 
    size_t readerIndex_; 
    size_t writerIndex_; 
    static const char kCRLF[]; 
};

汉译英服务

服务端

#include "include/muduo/net/TcpServer.h"
#include "include/muduo/net/TcpConnection.h"
#include "include/muduo/net/EventLoop.h"

#include <iostream>
#include <functional>
#include <unordered_map>

class TranslateServer
{
    public:
    TranslateServer(uint16_t port)
        : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port),
                  "TranslateServer", muduo::net::TcpServer::kReusePort)
        {
            // 将成员函数设置成server的回调函数
            _server.setConnectionCallback(std::bind(&TranslateServer::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&TranslateServer::onMessage, this, std::placeholders::_1,
                                                 std::placeholders::_2, std::placeholders::_3));
        }
    void Start()
    {
        _server.start();
        _baseloop.loop(); 
    }

    private:
    void onConnection(const muduo::net::TcpConnectionPtr &conn)
    {
        // 建立新连接的回调函数
        if (conn->connected())
        {
            std::cout << "新连接建立成功！" << std::endl;
        }
        else
        {
            std::cout << "新连接退出" << std::endl;
        }
    }
    std::string translate(const std::string& req)
    {
        static std::unordered_map<std::string,std::string> dictMap = {
            {"hello","你好"},
            {"Hello","你好"},
            {"吃了吗","肉夹馍"}
        };
        auto it = dictMap.find(req);
        if(it == dictMap.end())
            return "没找到";
        else
            return it->second;
    }
    void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buffer, muduo::Timestamp)
    {
        // 处理请求的回调函数
        //1.从buffer中拿到请求
        std::string req = buffer->retrieveAllAsString();
        //2.对请求进行处理
        std::string resp = translate(req);
        //3.发送响应
        conn->send(resp);
    }

    private:
    muduo::net::TcpServer _server;
    muduo::net::EventLoop _baseloop;
};

int main()
{
    TranslateServer server(8088);
    server.Start();
    return 0;
}

客户端

#include "include/muduo/net/TcpServer.h"
#include "include/muduo/net/TcpClient.h"
#include "include/muduo/net/TcpConnection.h"
#include "include/muduo/base/CountDownLatch.h"
#include "include/muduo/net/EventLoopThread.h"

#include <iostream>
#include <functional>


class TranslateClient
{
public:
    TranslateClient(const std::string &server_ip, int server_port)
    : _latch(1)
    , _client(_loopthread.startLoop(), muduo::net::InetAddress(server_ip,server_port), "TranslateClient")
    {
        _client.setConnectionCallback(std::bind(&TranslateClient::onConnection, this, std::placeholders::_1));
        _client.setMessageCallback(std::bind(&TranslateClient::onMessage, this, std::placeholders::_1,
                                             std::placeholders::_2, std::placeholders::_3));
    }
    // 阻塞式等待连接成功
    void connect()
    {
        _client.connect();
        _latch.wait();
    }
    bool send(const std::string &msg)
    {
        //连接状态正常再发送
        if(_conn->connected())
        {
            _conn->send(msg);
            return true;
        }
        return false;
    }

private:
    // 建立连接成功后，唤醒上面的阻塞
    void onConnection(const muduo::net::TcpConnectionPtr &conn)
    {
        if(conn->connected())
        {
            _latch.countDown();
            _conn = conn;
        }
        else
        {
            //连接关闭时的操作
            _conn.reset();
        }
    }
    void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buffer, muduo::Timestamp)
    {
        std::cout<<"翻译结果："<<buffer->retrieveAllAsString()<<std::endl;
    }
    muduo::CountDownLatch _latch;
    muduo::net::EventLoopThread _loopthread;
    muduo::net::TcpClient _client;
    muduo::net::TcpConnectionPtr _conn;   
};

int main()
{
    TranslateClient client("127.0.0.1", 8088);
    client.connect();

    while (1)
    {
        std::string msg;
        std::cin >> msg;
        client.send(msg);
    }
    return 0;
}