如何防止基于ASIO的服务器终止

Question

我一直在阅读一些Boost ASIO教程.到目前为止,我的理解是整个发送和接收是一个只能迭代一次的循环.请看下面的简单代码:

client.cpp:

#include <boost/asio.hpp>
#include <boost/array.hpp>
#include <iostream>
#include <string>

boost::asio::io_service io_service;
boost::asio::ip::tcp::resolver resolver(io_service);
boost::asio::ip::tcp::socket sock(io_service);
boost::array<char, 4096> buffer;

void read_handler(const boost::system::error_code &ec, std::size_t bytes_transferred)
{
  if (!ec)
  {
    std::cout << std::string(buffer.data(), bytes_transferred) << std::endl;
    sock.async_read_some(boost::asio::buffer(buffer), read_handler);
  }
}

void connect_handler(const boost::system::error_code &ec)
{
  if (!ec)
  {        
    sock.async_read_some(boost::asio::buffer(buffer), read_handler);
  }
}

void resolve_handler(const boost::system::error_code &ec, boost::asio::ip::tcp::resolver::iterator it)
{
  if (!ec)
  {
    sock.async_connect(*it, connect_handler);
  }
}

int main()
{
  boost::asio::ip::tcp::resolver::query query("localhost", "2013");
  resolver.async_resolve(query, resolve_handler);
  io_service.run();
}

程序resolves地址,connects服务器和reads数据,最后ends没有数据时.我的问题:我怎样才能继续这个循环？我的意思是,如何在应用程序的整个生命周期内保持客户端和服务器之间的连接,以便服务器在有内容发送时发送数据？我试图打破这个圈子,但是所有接缝都被困在io_service.run() 同一个问题中,如果我的服务器也是如此:

server.cpp:

#include <boost/asio.hpp>
#include <string>

boost::asio::io_service io_service;
boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::tcp::v4(), 2013);
boost::asio::ip::tcp::acceptor acceptor(io_service, endpoint);
boost::asio::ip::tcp::socket sock(io_service);
std::string data = "Hello, world!";

void write_handler(const boost::system::error_code &ec, std::size_t bytes_transferred)
{
}

void accept_handler(const boost::system::error_code &ec)
{
  if (!ec)
  {
    boost::asio::async_write(sock, boost::asio::buffer(data), write_handler);
  }
}

int main()
{
  acceptor.listen();
  acceptor.async_accept(sock, accept_handler);
  io_service.run();
}

这只是一个例子.在实际应用程序中,我可能希望保持套接字打开并将其重用于其他数据交换(读取和写入).我怎么能这样做

我非常重视你的评论.如果你有一些解决这个问题的简单解决方案的参考,我很感激,如果你提到它.谢谢

更新(服务器示例代码)

根据下面给出的答案(更新2),我编写了服务器代码.请注意,代码已经简化(能够编译和运行).另请注意,io_service永远不会返回,因为它始终在等待新的连接.这就是io_service.run永不回归和永远运行的方式.每当你想要io_service.run返回时,只需让接受者不再接受.请以我目前不记得的许多方式之一来做这件事.(说真的,我们如何以干净的方式做到这一点？:))

请享用:

#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <string>
#include <iostream>
#include <vector>
#include <time.h>
boost::asio::io_service io_service;
boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::tcp::v4(), 2013);
boost::asio::ip::tcp::acceptor acceptor(io_service, endpoint);
//boost::asio::ip::tcp::socket sock(io_service);
std::string data = "Hello, world!";

class Observer;
std::vector<Observer*> observers;

class Observer
{
public:
    Observer(boost::asio::ip::tcp::socket *socket_):socket_obs(socket_){}
    void notify(std::string data)
    {
        std::cout << "notify called data[" << data << "]" << std::endl;
        boost::asio::async_write(*socket_obs, boost::asio::buffer(data) , boost::bind(&Observer::write_handler, this,boost::asio::placeholders::error));
    }
    void write_handler(const boost::system::error_code &ec)
    {
         if (!ec) //no error: done, just wait for the next notification
             return;
         socket_obs->close(); //client will get error and exit its read_handler
         observers.erase(std::find(observers.begin(), observers.end(),this));
         std::cout << "Observer::write_handler  returns as nothing was written" << std::endl;
    }
private:
        boost::asio::ip::tcp::socket *socket_obs;
};


class server
{
public:
     void CreatSocketAndAccept()
     {
          socket_ = new boost::asio::ip::tcp::socket(io_service);
          observers.push_back(new Observer(socket_));
          acceptor.async_accept(*socket_,boost::bind(&server::handle_accept, this,boost::asio::placeholders::error));
     }
      server(boost::asio::io_service& io_service)
      {
          acceptor.listen();
          CreatSocketAndAccept();
      }

      void handle_accept(const boost::system::error_code& e)
      {
          CreatSocketAndAccept();
      }
private:
  boost::asio::ip::tcp::socket *socket_;
};

class Agent
{
public:
    void update(std::string data)
    {
        if(!observers.empty())
        {
//          std::cout << "calling notify data[" << data << "]" << std::endl;
            observers[0]->notify(data);
        }
    }

};
Agent agent;
void AgentSim()
{
    int i = 0;
    sleep(10);//wait for me to start client
    while(i++ < 10)
    {
        std::ostringstream out("");
        out << data << i ;
//      std::cout << "calling update data[" << out.str() << "]" << std::endl;
        agent.update(out.str());
        sleep(1);
    }
}
void run()
{
    io_service.run();
    std::cout << "io_service returned" << std::endl;
}
int main()
{
    server server_(io_service);

    boost::thread thread_1(AgentSim);
    boost::thread thread_2(run);
    thread_2.join();
    thread_1.join();
}

Answer 1

您可以简化基于asio的图的逻辑,如下所示:调用async_X函数的每个函数都提供一个处理程序.这有点像状态机状态之间的转换,其中处理程序是状态,异步调用是状态之间的转换.只是在不调用async_*函数的情况下退出处理程序就像转换到结束状态一样.程序"执行"(发送数据,接收数据,连接套接字等)的所有内容都在转换期间发生.

如果你这样看,你的客户端看起来像这样(只有"好路径",即没有错误):

<start>         --(resolve)----> resolve_handler
resolve_handler --(connect)----> connect_handler
connect_handler --(read data)--> read_handler
read_handler    --(read data)--> read_handler

你的服务器是这样的:

<start>         --(accept)-----> accept handler
accept_handler  --(write data)-> write_handler
write_handler   ---------------> <end>

由于您write_handler没有做任何事情,它会转换到最终状态,即ioservice::run返回.现在的问题是,在将数据写入套接字后,您想要做什么？根据这一点,您必须定义相应的转换,即执行您想要执行的操作的异步调用.

更新: 从您的评论中我看到您要等待下一个数据准备就绪,即下一个滴答.然后转换看起来像这样:

write_handler   --(wait for tick/data)--> dataready
dataready       --(write data)----------> write_handler

你看,这引入了一个新的状态(处理程序),我称之为dataready,你也可以调用它tick_handler或其他东西.转回write_handler很容易:

void dataready()
{
  // get the new data...
  async_write(sock, buffer(data), write_handler);
}

某些计时器的过渡write_handler可以很简单async_wait.如果数据来自"外部"并且您不确切知道它们何时准备就绪,请等待一段时间,检查数据是否存在,如果不存在,请等待一段时间:

write_handler    --(wait some time)--> checkForData
checkForData:no  --(wait some time)--> checkForData
checkForData:yes --(write data)------> write_handler

或者,在(伪)代码中:

void write_handler(const error_code &ec, size_t bytes_transferred)
{
  //...
  async_wait(ticklenght, checkForData);
}

void checkForData(/*insert wait handler signature here*/)
{
  if (dataIsReady())
  {
    async_write(sock, buffer(data), write_handler);
  }
  else
  {
    async_wait(shortTime, checkForData):
  }
}

更新2: 根据您的评论,您已经有一个代理人以某种方式进行时间管理(每次打勾调用更新).以下是我将如何解决这个问题:

1. 让代理有一个观察者列表,当更新调用中有新数据时,它会得到通知.
2. 让每个观察者处理一个客户端连接(套接字).
3. 让服务器等待接入连接,从它们创建观察者并将其注册到代理.

我对ASIO的确切语法不是很坚定,所以这将是handwavy伪代码:

服务器:

void Server::accept_handler()
{
    obs = new Observer(socket);
    agent.register(obs);
    new socket; //observer takes care of the old one
    async_accept(..., accept_handler);
}

代理:

void Agent::update()
{
    if (newDataAvailable())
    {
        for (auto& obs : observers)
        {
             obs->notify(data);
        }
    }
}

观察报:

void Observer::notify(data)
{
    async_write(sock, data, write_handler);
}

void Observer::write_handler(error_code ec, ...)
{
     if (!ec) //no error: done, just wait for the next notification
         return;
     //on error: close the connection and unregister
     agent.unregister(this);
     socket.close(); //client will get error and exit its read_handler
}