基础知识

std::thread

std::thread是C++11标准库中的一个类，用于创建并发执行的线程。它的详细用法如下：

1. 头文件

#include <thread>

2. 创建线程

std::thread t(func, args...);

其中，func是线程要执行的函数，args是传递给函数的参数。创建线程后，线程会立即开始执行func函数。

3. 等待线程结束

t.join();

join()函数会阻塞当前线程，直到t线程执行完毕。

4. 分离线程

t.detach();

detach()函数会将t线程与当前线程分离，使得t线程在后台继续执行，当前线程不再等待t线程执行完毕。

5. 判断线程是否可执行

if (t.joinable()) {// ...
}

joinable()函数用于判断线程是否可执行，如果线程已经被join()或detach()，则返回false。

6. 获取线程ID

std::thread::id id = t.get_id();

get_id()函数用于获取线程的ID。

7. 线程休眠

std::this_thread::sleep_for(std::chrono::milliseconds(ms));

sleep_for()函数用于让当前线程休眠一段时间，ms是休眠的毫秒数。

8. 线程同步

C++11标准库中提供了多种线程同步的机制，如互斥锁、条件变量、原子操作等。这些机制可以用于保证多个线程之间的数据同步和互斥访问。

实战

boost domain socket server

#include <cstdio>
#include <iostream>
#include <boost/array.hpp>
#include <boost/bind/bind.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/asio.hpp>#if defined(BOOST_ASIO_HAS_LOCAL_SOCKETS)namespace CVpn {
using boost::asio::local::stream_protocol;class UnixSocketSession : public boost::enable_shared_from_this<UnixSocketSession>
{
public:UnixSocketSession(boost::asio::io_context &ioContext,std::function<std::optional<std::string>(std::string &)> msgProc) : socket(ioContext), msgProc(msgProc){}stream_protocol::socket &GetSocket(){return socket;}void Start(){socket.async_read_some(boost::asio::buffer(data), boost::bind(&UnixSocketSession::HandleRead,shared_from_this(), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));}void HandleRead(const boost::system::error_code &error, size_t bytes_transferred){if (!error) {// 处理消息std::cout << "UnixSocketSession: HandleRead: ok, data=" << data.data() << std::endl;std::string message(data.data(), bytes_transferred);auto replay = msgProc(message);if (replay.has_value()) { // 返回值非空表示有消息要回复给客户端boost::asio::async_write(socket, boost::asio::buffer(replay.value()),boost::bind(&UnixSocketSession::HandleWrite, shared_from_this(), boost::asio::placeholders::error,boost::asio::placeholders::bytes_transferred));}} else {std::cout << "UnixSocketSession: HandleRead: fail, error:" << error.message() << std::endl;}}void HandleWrite(const boost::system::error_code &error, size_t bytes_transferred){if (!error){std::cout << "UnixSocketSession: HandleWrite: replay ok, bytes=" << bytes_transferred << std::endl;socket.async_read_some(boost::asio::buffer(data),boost::bind(&UnixSocketSession::HandleRead,shared_from_this(),boost::asio::placeholders::error,boost::asio::placeholders::bytes_transferred));} else {std::cout << "UnixSocketSession: HandleWrite: fail, error:" << error.message() << std::endl;}}private:stream_protocol::socket socket; // 用于跟客户端通信的socketstd::function<std::optional<std::string>(std::string &)> msgProc; // 收到客户端消息后的消息处理函数boost::array<char, 1024> data; // 用于保存从客户端接收的数据
};class UnixServer
{
public:UnixServer(boost::asio::io_context &ioContext, const std::string &filePath,std::function<std::optional<std::string>(std::string &)> handler) :ioc(ioContext),acceptor(ioContext, stream_protocol::endpoint(filePath)),workingStatus(STOPPING),msgProc(handler){}void Start(){workingStatus = RUNNING;boost::shared_ptr<UnixSocketSession> newSession(new UnixSocketSession(ioc, msgProc));acceptor.async_accept(newSession->GetSocket(),boost::bind(&UnixServer::HandleAccept, this, newSession, boost::asio::placeholders::error));}void Stop(){workingStatus = STOPPING;acceptor.close();}// 测试接口void TestSetServerStopping(){workingStatus = STOPPING;}void HandleAccept(boost::shared_ptr<UnixSocketSession> newSession,const boost::system::error_code &error){if (!error) {std::cout << "UnixServer: HandleAccept: ok" << std::endl;newSession->Start();} else {std::cout << "UnixServer: HandleAccept: fail, error:" << error.message() << std::endl;}if (workingStatus == RUNNING) {// reset后会将之前的newSession对象的共享计数减一，等该session不再运行后就会自动回收相关资源newSession.reset(new UnixSocketSession(ioc, msgProc));acceptor.async_accept(newSession->GetSocket(),boost::bind(&UnixServer::HandleAccept, this, newSession, boost::asio::placeholders::error));}}private:boost::asio::io_context &ioc;stream_protocol::acceptor acceptor;enum {RUNNING, STOPPING,} workingStatus;// 收到客户端消息后回调用户传入的消息处理函数，如果返回值非空表示有消息要回复给客户端std::function<std::optional<std::string>(std::string &)> msgProc;
};
} // namespace CVpn
#else // defined(BOOST_ASIO_HAS_LOCAL_SOCKETS)
# error Local sockets not available on this platform.
#endif // defined(BOOST_ASIO_HAS_LOCAL_SOCKETS)