使用 boost::thread 开始/停止记录数据（第二次更新）

Posted 2023-02-21

【中文标题】使用 boost::thread 开始/停止记录数据（第二次更新）

【英文标题】：Using boost::thread to start/stop logging data (2nd update)

【发布时间】：2014-03-10 16:14:38

【问题描述】：

我目前正在尝试使用 boost::thread 和一个复选框来记录实时数据。当我选中该框时，日志记录线程开始。当我取消选中时，日志记录线程停止。当我反复且非常快速地检查/取消检查（程序崩溃、某些文件未记录等）时，就会出现问题。如何编写一个可靠的线程安全程序，在反复快速检查/取消检查时不会出现这些问题？我也不想使用 join() 因为这会暂时停止来自主线程的数据输入。在辅助线程中，我打开一个日志文件，从套接字读取到缓冲区，将其复制到另一个缓冲区，然后将此缓冲区写入日志文件。我在想也许我应该使用互斥锁进行读/写。如果是这样，我应该使用什么特定的锁？下面是一段代码sn-p：

//Main thread
 if(m_loggingCheckBox->isChecked()) 

...

if(m_ThreadLogData.InitializeReadThread(socketInfo))//opens the socket. 
//If socket is opened and can be read, start thread.
 m_ThreadLogData.StartReadThread();
 else
 std::cout << "Did not initialize thread\n";
 
 else if(!m_loggingCheckBox->isChecked())
 

m_ThreadLogData.StopReadThread();



void ThreadLogData::StartReadThread()
 
 //std::cout << "Thread started." << std::endl;
 m_stopLogThread = false;
 m_threadSendData = boost::thread(&ThreadLogData::LogData,this);
 

void ThreadLogData::StopReadThread()
 
 m_stopLogThread = true;
 m_ReadDataSocket.close_socket(); // close the socket

if(ofstreamLogFile.is_open())
 
 ofstreamLogFile.flush(); //flush the log file before closing it.
 ofstreamLogFile.close(); // close the log file
 
 m_threadSendData.interrupt(); // interrupt the thread
 //m_threadSendData.join(); // join the thread. Commented out since this
 temporarily stops data input.



//secondary thread
 bool ThreadLogData::LogData()
 

unsigned short int buffer[1024];
 bool bufferflag;
 unsigned int iSizeOfBuffer = 1024;
 int iSizeOfBufferRead = 0;
 int lTimeout = 5;

if(!ofstreamLogFile.is_open())
 
 ofstreamLogFile.open(directory_string().c_str(), ios::out);

if(!ofstreamLogFile.is_open())
 
 return 0;
 
 

while(!m_stopLogThread)
 
 try 
 int ret = m_ReadDataSocket.read_sock(&m_msgBuffer.m_buffer
 [0],iSizeOfBuffer,lTimeout,&iSizeOfBufferRead);

memcpy(&buffer[0],m_msgBuffer.m_buffer,iSizeOfBufferRead);
 bufferflag = m_Buffer.setBuffer(buffer);
 if(!bufferflag) return false;
 object = &m_Buffer;

unsigned int data = object->getData();

ofstreamLogFile << data << std::endl;

boost::this_thread::interruption_point();

 catch (boost::thread_interrupted& interruption) 
 std::cout << "ThreadLogData::LogData(): Caught Interruption thread." << std::endl;
 StopReadThread();
  catch (...) 
 std::cout << "ThreadLogData::LogData(): Caught Something." << std::endl;
 StopReadThread();
 

 // end while()

【问题讨论】：

请编辑您的问题，而不是发布重复的问题。我只是在一个版本上浪费了时间，却不知道你取代了它。两次。

【参考方案1】：

我喜欢将 Boost Asio 用于异步处理

#include <iostream>
#include <fstream>

#include <boost/asio.hpp>
#include <boost/asio/signal_set.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/bind.hpp>
#include <boost/optional.hpp>

#include <thread>

using boost::asio::ip::tcp;
namespace asio = boost::asio;

struct program

    asio::io_service       _ioservice;
    asio::deadline_timer   _timer;
    asio::signal_set       _signals;
    std::array<char, 1024> _buffer;
    tcp::socket            _client;
    tcp::resolver          _resolver;
    std::ofstream          _logfile;
    std::thread            _thread;

    program() 
        : _timer(_ioservice),
        _signals(_ioservice),
        _client(_ioservice),
        _resolver(_ioservice)
    
        do_connect(_resolver.resolve( "localhost", "6767" ));
        do_toggle_logging_cycle();

        _signals.add(SIGINT);
        _signals.async_wait([this](boost::system::error_code ec, int)  if (!ec) close(); );

        _thread = std::thread(boost::bind(&asio::io_service::run, boost::ref(_ioservice)));
    

    ~program()
    
        if (_thread.joinable())
            _thread.join();
    

    void close() 
        _ioservice.post([this]()  
            _signals.cancel();
            _timer.cancel();
            _client.close(); 
        );
    

private:

  void do_toggle_logging_cycle(boost::system::error_code ec = )
  
      if (ec != boost::asio::error::operation_aborted)
      
          if (_logfile.is_open())
          
              _logfile.close();
              _logfile.clear();
           else
          
              _logfile.open("/tmp/output.log");
          

          _timer.expires_from_now(boost::posix_time::seconds(2));
          _timer.async_wait(boost::bind(&program::do_toggle_logging_cycle, this, boost::asio::placeholders::error()));
       else 
      
          std::cerr << "\nDone, goobye\n";
      
  

  void do_connect(tcp::resolver::iterator endpoint_iterator) 

      boost::asio::async_connect(
          _client, endpoint_iterator,
          [this](boost::system::error_code ec, tcp::resolver::iterator) 
                if (!ec) do_read();
                else     close();
          );
  

  void do_read() 
    boost::asio::async_read(
        _client, asio::buffer(_buffer.data(), _buffer.size()),
        [this](boost::system::error_code ec, std::size_t length) 
            if (!ec) 
              if (_logfile.is_open())
              
                    _logfile.write(_buffer.data(), length);
              
              do_read();
             else 
                close();
            
    );
  

;

int main()

    
        program p; // does socket reading and (optional) logging on a separate thread

        std::cout << "\nMain thread going to sleep for 15 seconds...\n";
        std::this_thread::sleep_for(std::chrono::seconds(15));

        p.close(); // if the user doesn't press ^C, let's take the initiative
        std::cout << "\nDestruction of program...\n";
    
    std::cout << "\nMain thread ends\n";
;

程序连接到localhost的6767端口，异步读取数据。

如果日志记录处于活动状态 (_logfile.is_open())，则所有接收到的数据都将写入 /tmp/output.log。

现在

读/写在单独的线程上，但所有操作都使用_ioservice 序列化（参见例如close() 中的post） 用户可以使用 Ctrl+C 中止套接字读取循环 每 2 秒，日志记录将被（停用）激活（请参阅 do_toggle_logging_cycle）

主线程在取消程序之前只休眠 15 秒（类似于用户按 Ctrl-C）。