C++ 设计：多个 TCP 客户端、boost asio 和观察者

Question

在我的系统中，我有一堆 TCP 客户端，我对如何设计它有点困惑 [我的大部分经验是在 C 中，因此不安全]。我正在使用 boost ASIO 来管理连接。这些是我拥有的组件

• TCPStream 类：boost asio 上的瘦包装器
• 一种 IPC 协议，它通过 TCP 实现协议： 基本上每条消息都以类型和长度字段开头 这样我们就可以从流中读取各个消息。
• 处理消息的连接类
• 监控连接的观察者类

为了简洁起见，我正在编写伪 C++ 代码。我想你会明白的

class TCPStream {
   boost::asio::socket socket_;
public:

   template <typename F>
   void connect (F f)
   {
       socket_.connect(f);
   }

   template <typename F>
   void read (F f)
   {
      socket_.read(f);
   }
};

class IpcProtocol : public TCPStream {
public:
    template <typename F
    void read (F f)
    {
        TCPStream::read(
              [f] (buffer, err) {

                while (msg = read_indvidual_message(buffer)) {
                      // **** this is a violation of how this pattern is 
                      // supposed to work. Ideally there should a callback 
                      // for individual message. Here the same callback
                      // is called for N no. of messages. But in our case  
                      // its the same callback everytime so this should be      
                      // fine - just avoids some function calls.
                      f(msg);
                };
              };
         )
    }
};

假设我有一堆 TCP 连接，并且有一个处理程序类 对于每个连接。让我们将其命名为 Connection1、Connection2 ...

class Connection {
    virtual int type() = 0;
};

class Connection1 : public Connection {

   shared_ptr<IpcProtocol> ipc_;

   int type ()
   {
       return 1;
   }

   void start ()
   {
       ipc_.connect([self = shared_from_this()](){ self->connected(); });

       ipc_.read(
            [self = shared_from_this()](msg, err) {

              if (!err)
                  self->process(msg);
              } else {
                  self->error();
              }   
            });
   }

   void connected ()
   {
       observer.notify_connected(shared_from_this());
   }

   void error ()
   {
       observer.notify_error(shared_from_this());
   }
};

这种模式以一种或另一种方式对所有连接重复。 消息由连接类本身处理。但它会让你知道 其他事件 [connect, error] 到观察者。原因——

1. 每次断开连接时重新启动连接
2. 一群人需要知道连接是否已建立，以便他们可以 将初始请求/配置发送到服务器。
3. 根据多个连接的连接状态需要做一些事情 eg：如果connection1和connection2都建立了，则启动connection3等。

我在那里添加了一个中间 Observer 类，以便观察者每次重新启动时都必须直接连接到连接。每次连接中断时，都会删除连接类并创建新的连接类。

 class Listeners {
public:
    virtual void notify_error(shared_ptr<Connection>) = 0;
    virtual void notify_connect(shared_ptr<Connection>) = 0;
    virtual void interested(int type) = 0;
};


class Observer {
   std::vector<Listeners *> listeners_;
public:

   void notify_connect(shared_ptr<Connection> connection)
   {
        for (listener : listeners_) {
            if (listener->interested(connection->type())) {
                listener->notify_error(connection);
            }
        }       
   }
};

现在这个作品的粗略原型。但我想知道如果这个类设计 有什么好处。有多个流服务器将不断产生状态并将其发送到我的模块以对状态进行硬件编程。这需要可扩展，因为将来会添加更多客户端。

线程

旧代码每个 TCP 连接有一个线程，这工作正常。在这里，我试图在同一个线程上处理多个连接。仍然会有多个线程调用 ioservice。所以观察者将在多个线程上运行。我计划为每个侦听器设置一个互斥体，这样侦听器就不会同时收到多个事件。

Answer 1

HTTP 通过 TCP 实现协议，因此 HTTP 服务器 asio examples 是您设计的一个很好的起点，尤其是：HTTP Server 2、HTTP Server 3 和 HTTP Server 4。

注意：连接生命周期可能是个问题，尤其是当您打算将类成员函数用作处理程序时，请在此处查看问题和答案：How to design proper release of a boost::asio socket or wrapper thereof。