C++ 设计 : Multiple TCP clients, 提升 asio 和观察者

Question

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

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

为了简洁起见，我正在编写伪 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);
            }
        }       
   }
};

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

线程

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

Answer 1

HTTP 通过 TCP 实现协议(protocol)，因此 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 .