Асинхронная обработка в C++

Question

Сервер, который будет работать вечно и обрабатывает запросы, нуждается в асинхронной части кода, которая будет выполнять некоторые запросы к базе данных и обновляться только тогда, когда в нее будут внесены новые изменения. Сервер должен запускаться вечно, и эта функция для выполнения функции db снова и снова должна выполняться асинхронно, так что сервер не будет препятствовать из-за обновления один раз в каждые «x» минуты.

Как лучше всего это можно обрабатывать асинхронно в C++? Как я могу настроить эту функцию самостоятельно для запуска на демонах, чтобы она вообще не блокировала сервер?

Answer 1

я настоятельно рекомендую использовать Boost's ASIO библиотеку

Вы должны были бы класс, чтобы принимать новые запросы, а другой периодически проверять наличие обновлений. Оба могут выполнять свою работу асинхронно и использовать один и тот же boost :: asio :: io_service для планирования работы.

Установка будет

• сеть асинхронного boost::asio::ip::tcp::acceptor прослушивания для новых запросов.
• A boost::asio::deadline_time выполнить асинхронное ожидание проверить наличие обновлений для базы данных.

Псевдо-код для того, что я понимаю, что вы описываете ниже:

#include <iostream> 
#include <boost/asio.hpp> 
#include <boost/bind.hpp> 
#include <boost/shared_ptr.hpp> 
#include <string> 

class DatabaseUpdateChecker{ 
    public: 
    DatabaseUpdateChecker(boost::asio::io_service& io, const int& sleepTimeSeconds) 
    :timer_(io,boost::posix_time::seconds(sleepTimeSeconds)),sleepSeconds_(sleepTimeSeconds){ 
     this->timer_.async_wait(boost::bind(&DatabaseUpdateChecker::doDBUpdateCheck,this,boost::asio::placeholders::error)); 
    }; 

    protected: 
    void doDBUpdateCheck(const boost::system::error_code& error){ 
     if(!error){ 
      std::cout << " Checking Database for updates" << std::endl; 
      //Reschdule ourself 
      this->timer_.expires_at(timer_.expires_at() + boost::posix_time::seconds(this->sleepSeconds_)); 
      this->timer_.async_wait(boost::bind(&DatabaseUpdateChecker::doDBUpdateCheck,this,boost::asio::placeholders::error)); 
     } 
    }; 
    private: 
    boost::asio::deadline_timer timer_; 
    int sleepSeconds_; 
}; 

typedef boost::shared_ptr<boost::asio::ip::tcp::socket> TcpSocketPtr; 

class NetworkRequest{ 
    public: 
    NetworkRequest(boost::asio::io_service& io, const int& port) 
    :acceptor_(io,boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(),port)){ 
     this->start_accept(); 
    }; 
    protected: 
    void start_accept(){ 
     TcpSocketPtr socketPtr(new boost::asio::ip::tcp::socket(acceptor_.get_io_service())); 
     std::cout << "About to accept new connection" << std::endl; 
     acceptor_.async_accept(*socketPtr,boost::bind(&NetworkRequest::handle_accept,this,socketPtr,boost::asio::placeholders::error)); 
    }; 
    void handle_accept(TcpSocketPtr socketPtr,const boost::system::error_code& error){ 
     std::cout << "Accepted new network connection" << std::endl; 
     if(!error){ 
      std::string response("This is a response\n"); 
      boost::asio::async_write(*socketPtr,boost::asio::buffer(response), 
       boost::bind(&NetworkRequest::handle_write,this,boost::asio::placeholders::error,boost::asio::placeholders::bytes_transferred)); 
     } 
     //Start listeing for a new connection 
     this->start_accept(); 
    } 
    void handle_write(const boost::system::error_code& error,size_t size){ 
     if(!error){ 
      std::cout << "Wrote out " << size << " bytes to the network connection" << std::endl; 
     } 

    } 
    private: 
    boost::asio::ip::tcp::acceptor acceptor_; 
}; 

int main(int argc, char *argv[]) { 
    static const int DB_TIMER_SECONDS=5; 
    static const int LISTENING_TCP_PORT=4444; 

    std::cout << "About to start" << std::endl; 
    boost::asio::io_service io; 

    DatabaseUpdateChecker dbChecker(io,DB_TIMER_SECONDS); 
    NetworkRequest networkRequestAcceptor(io,LISTENING_TCP_PORT); 

    io.run(); 

    std::cout << "This won't be printed" << std::endl; 
    return 0; 
} 

Компиляция выше, и работает он покажет, что база данных Update Checker будет проверять наличие обновлений каждые 5 секунд во время прослушивания для соединений на TCP-порт 4444. Чтобы увидеть, как код принимает новое соединение, вы можете использовать telnet/netcat/ваш любимый инструмент сетевого клиента ....

telnet 127.0.0.1 4444 
Trying 127.0.0.1... 
Connected to localhost. 
Escape character is '^]'. 
This is a response 
Connection closed by foreign host. 

Если вы обнаружили, что обработка обновлений и/или запросов занимает значительное количество времени, то я хотел бы посмотреть в резьбе приложения и запуская каждую задачу в своем собственном потоке. io_service запланирует, какую работу он должен выполнить, а не завершить, пока не будет больше работы. Хитрость заключается в том, чтобы занятия, выполняющие работу, перенесли на себя, когда они будут выполнены.

Конечно, вы должны учитывать комментарии других участников по вашему вопросу. Я не знаю, как интерфейс CORBA может усложнить это, но я бы подумал, что boost :: asio как асинхронная библиотека C++ будет хорошим решением и достаточно гибким для того, что вы описываете.

Answer 2

Похоже, что это означает, что, хотя система непрерывно обрабатывает сетевые запросы, она асинхронно связывается с БД.

Это означает, что когда ему нужно поговорить с БД, он отправляет запрос, но не ждет ответа.

Когда он получает ответ от БД, он обрабатывает его.

Асинхронная часть может быть реализована посредством отдельного потока, который ведет переговоры с БД, и когда он получает ответ, он помещает даже в очередь сервера для обработки.

Или сервер может прослушивать множество сокетов для данных, а один из них может быть подключением к базе данных, где он получает ответы от БД.

Answer 3

Таким образом, в основном (если я вас правильно понимаю) вам необходимо периодически опробовать базу данных, чтобы узнать, не изменилось ли оно. И когда он изменился, вам нужно уведомить обработчик запросов на основе CORBA, что произошли изменения.

Что бы я сделал, это добавить новый тип запроса на ваш сервер CORBA. Запрос будет «Обновление базы данных». Затем вы можете написать небольшую, совершенно другую программу, единственной задачей которой является опрос базы данных и отправка запроса CORBA при обновлении базы данных.

Таким образом, вы можете сбросить сообщение обновления базы данных в основной поток запросов для сервера CORBA.

Нет нитей, никаких асинхронных ничего. Всего два процесса, каждый из которых делает свое дело.