我正在尝试用 c++ 设计一个软件，它将使用 **UDP 协议(protocol)** 发送请求字节(遵循标准的 **应用程序级别** 协议(protocol)，其要填充的字段将从文本文件中获取)。
现在这个客户端必须能够以非常高的速率发送这些请求..高达每秒 **2000 个事务**并且如果它在指定的超时内到达也应该接收响应否则不接收它
我将对所有套接字使用 boost 库，但我不确定它的设计是否适用于如此高速的应用程序 :(
我想我必须使用高度多线程的应用程序(将再次使用 Boost)。我对吗 ？我是否必须为每个请求创建一个单独的线程？但我认为只有一个线程必须等待接收响应，否则如果许多线程正在等待响应，我们如何区分我们已获得响应的线程请求!
希望这个问题很清楚。我只需要一些关于我可能面临的设计要点和疑似问题的帮助。
最佳答案
我目前正在使用自己的网络客户端，所以也许我可以提供一些建议和一些资源供您查看。在这方面有很多经验丰富的人，希望他们能插话 :)
首先，您需要提升。一旦你习惯了它们是如何联系在一起的， boost::asio 是编写网络代码的绝佳工具包。本质上，您创建了一个 io_service并调用run执行直到所有工作完成，或 runOne执行单个 IO 操作。就他们自己而言，这并不是很有帮助。当您运行 runOne 时，电源就会产生在它自己的循环中:
boost::asio::io_service myIOService;
while(true)
{
myIOService.runOne();
}
，或运行 run在一个(或多个)线程上运行:
boost::thread t(boost::bind(&boost::asio::io_service::run, &myIOService));
但是，值得注意的是run一旦没有工作要做就返回(这样你就可以和那个线程说再见了)。正如我在 Stackoverflow 上发现的那样，诀窍是确保它始终有事可做。解决方案在boost::asio::io_service::work :
boost::asio::io_service::work myWork(myIOService); // just some abstract "work"
上面一行确保你的线程在没有任何事情发生时不会停止。我认为这是一种保持活力的方式:)
在某些时候，您会想要创建一个套接字并将其连接到某处。我创建了一个通用的 Socket 类(并从中派生了一个文本套接字以创建缓冲输入)。我还想要一个非常像 C# 的基于事件的系统。我在下面为您概述了这些内容:
第一步，我们需要一种传递参数的通用方法，因此，EventArgs :
eventArgs.h
class EventArgs : boost::noncopyable
{
private:
public:
EventArgs();
virtual ~EventArgs() = 0;
}; // eo class EventArgs:
现在，我们需要一个人们可以订阅/取消订阅的事件类:
事件.h
// STL
#include <functional>
#include <stack>
// Boost
#include <boost/bind.hpp>
#include <boost/thread/mutex.hpp>
// class Event
class Event : boost::noncopyable
{
public:
typedef std::function<void(const EventArgs&)> DelegateType;
typedef boost::shared_ptr<DelegateType> DelegateDecl;
private:
boost::mutex m_Mutex;
typedef std::set<DelegateDecl> DelegateSet;
typedef std::stack<DelegateDecl> DelegateStack;
typedef DelegateSet::const_iterator DelegateSet_cit;
DelegateSet m_Delegates;
DelegateStack m_ToRemove;
public:
Event()
{
}; // eo ctor
Event(Event&& _rhs) : m_Delegates(std::move(_rhs.m_Delegates))
{
}; // eo mtor
~Event()
{
}; // eo dtor
// static methods
static DelegateDecl bindDelegate(DelegateType _f)
{
DelegateDecl ret(new DelegateType(_f));
return ret;
}; // eo bindDelegate
// methods
void raise(const EventArgs& _args)
{
boost::mutex::scoped_lock lock(m_Mutex);
// get rid of any we have to remove
while(m_ToRemove.size())
{
m_Delegates.erase(m_Delegates.find(m_ToRemove.top()));
m_ToRemove.pop();
};
if(m_Delegates.size())
std::for_each(m_Delegates.begin(),
m_Delegates.end(),
[&_args](const DelegateDecl& _decl) { (*_decl)(_args); });
}; // eo raise
DelegateDecl addListener(DelegateDecl _decl)
{
boost::mutex::scoped_lock lock(m_Mutex);
m_Delegates.insert(_decl);
return _decl;
}; // eo addListener
DelegateDecl addListener(DelegateType _f)
{
DelegateDecl ret(bindDelegate(_f));
return addListener(ret);
}; // eo addListener
void removeListener(const DelegateDecl _decl)
{
boost::mutex::scoped_lock lock(m_Mutex);
DelegateSet_cit cit(m_Delegates.find(_decl));
if(cit != m_Delegates.end())
m_ToRemove.push(_decl);
}; // eo removeListener
// operators
// Only use operator += if you don't which to manually detach using removeListener
Event& operator += (DelegateType _f)
{
addListener(_f);
return *this;
}; // eo op +=
}; // eo class Event
然后，是时候创建一个套接字类了。下面是标题:
套接字.h
(一些注释:ByteVector 是 typedef std::vector<unsigned char> )
#pragma once
#include "event.h"
// boost
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/buffer.hpp>
// class Socket
class MORSE_API Socket : boost::noncopyable
{
protected:
typedef boost::shared_ptr<boost::asio::ip::tcp::socket> SocketPtr;
private:
ByteVector m_Buffer; // will be used to read in
SocketPtr m_SocketPtr;
boost::asio::ip::tcp::endpoint m_RemoteEndPoint;
bool m_bConnected;
// reader
void _handleConnect(const boost::system::error_code& _errorCode, boost::asio::ip::tcp::resolver_iterator _rit);
void _handleRead(const boost::system::error_code& _errorCode, std::size_t read);
protected:
SocketPtr socket() { return m_SocketPtr; };
public:
Socket(ByteVector_sz _bufSize = 512);
virtual ~Socket();
// properties
bool isConnected() const { return m_bConnected; };
const boost::asio::ip::tcp::endpoint& remoteEndPoint() const {return m_RemoteEndPoint; };
// methods
void connect(boost::asio::ip::tcp::resolver_iterator _rit);
void connect(const String& _host, const Port _port);
void close();
// Events
Event onRead;
Event onResolve;
Event onConnect;
Event onClose;
}; // eo class Socket
然后，现在是实现。您会注意到它调用另一个类来执行 DNS 解析。之后我会证明这一点。还有一些EventArg -我省略的衍生物。当套接字事件发生时，它们只是作为 EventArg 参数传递。
套接字.cpp
#include "socket.h"
// boost
#include <boost/asio/placeholders.hpp>
namespace morse
{
namespace net
{
// ctor
Socket::Socket(ByteVector_sz _bufSize /* = 512 */) : m_bConnected(false)
{
m_Buffer.resize(_bufSize);
}; // eo ctor
// dtor
Socket::~Socket()
{
}; // eo dtor
// _handleRead
void Socket::_handleRead(const boost::system::error_code& _errorCode,
std::size_t _read)
{
if(!_errorCode)
{
if(_read)
{
onRead.raise(SocketReadEventArgs(*this, m_Buffer, _read));
// read again
m_SocketPtr->async_read_some(boost::asio::buffer(m_Buffer), boost::bind(&Socket::_handleRead, this, _1, _2));
};
}
else
close();
}; // eo _handleRead
// _handleConnect
void Socket::_handleConnect(const boost::system::error_code& _errorCode,
boost::asio::ip::tcp::resolver_iterator _rit)
{
m_bConnected = !_errorCode;
bool _raise(false);
if(!_errorCode)
{
m_RemoteEndPoint = *_rit;
_raise = true;
m_SocketPtr->async_read_some(boost::asio::buffer(m_Buffer), boost::bind(&Socket::_handleRead, this, _1, _2));
}
else if(++_rit != boost::asio::ip::tcp::resolver::iterator())
{
m_SocketPtr->close();
m_SocketPtr->async_connect(*_rit, boost::bind(&Socket::_handleConnect, this, boost::asio::placeholders::error, _rit));
}
else
_raise = true; // raise complete failure
if(_raise)
onConnect.raise(SocketConnectEventArgs(*this, _errorCode));
}; // eo _handleConnect
// connect
void Socket::connect(boost::asio::ip::tcp::resolver_iterator _rit)
{
boost::asio::ip::tcp::endpoint ep(*_rit);
m_SocketPtr.reset(new boost::asio::ip::tcp::socket(Root::instance().ioService()));
m_SocketPtr->async_connect(ep, boost::bind(&Socket::_handleConnect, this, boost::asio::placeholders::error, _rit));
};
void Socket::connect(const String& _host, Port _port)
{
// Anon function for resolution of the host-name and asynchronous calling of the above
auto anonResolve = [this](const boost::system::error_code& _errorCode,
boost::asio::ip::tcp::resolver_iterator _epIt)
{
// raise event
onResolve.raise(SocketResolveEventArgs(*this, !_errorCode ? (*_epIt).host_name() : String(""), _errorCode));
// perform connect, calling back to anonymous function
if(!_errorCode)
this->connect(_epIt);
};
// Resolve the host calling back to anonymous function
Root::instance().resolveHost(_host, _port, anonResolve);
}; // eo connect
void Socket::close()
{
if(m_bConnected)
{
onClose.raise(SocketCloseEventArgs(*this));
m_SocketPtr->close();
m_bConnected = false;
};
} // eo close
正如我所说的 DNS 解析，行 Root::instance().resolveHost(_host, _port, anonResolve);调用它来执行异步 DNS:
// resolve a host asynchronously
template<typename ResolveHandler>
void resolveHost(const String& _host, Port _port, ResolveHandler _handler)
{
boost::asio::ip::tcp::endpoint ret;
boost::asio::ip::tcp::resolver::query query(_host, boost::lexical_cast<std::string>(_port));
m_Resolver.async_resolve(query, _handler);
}; // eo resolveHost
最后，我需要一个基于文本的套接字，它在每次接收到我的行时引发一个事件(然后进行处理)。这次我将省略头文件，只显示实现文件。不用说它声明了一个 Event称为 onLine每次完整接收一行时它都会触发:
// boost
#include <boost/asio/buffer.hpp>
#include <boost/asio/write.hpp>
#include <boost/asio/placeholders.hpp>
namespace morse
{
namespace net
{
String TextSocket::m_DefaultEOL("\r\n");
// ctor
TextSocket::TextSocket() : m_EOL(m_DefaultEOL)
{
onRead += boost::bind(&TextSocket::readHandler, this, _1);
}; // eo ctor
// dtor
TextSocket::~TextSocket()
{
}; // eo dtor
// readHandler
void TextSocket::readHandler(const EventArgs& _args)
{
auto& args(static_cast<const SocketReadEventArgs&>(_args));
m_LineBuffer.append(args.buffer().begin(), args.buffer().begin() + args.bytesRead());
String::size_type pos;
while((pos = m_LineBuffer.find(eol())) != String::npos)
{
onLine.raise(SocketLineEventArgs(*this, m_LineBuffer.substr(0, pos)));
m_LineBuffer = m_LineBuffer.substr(pos + eol().length());
};
}; // eo readHandler
// writeHandler
void TextSocket::writeHandler(const boost::system::error_code& _errorCode, std::size_t _written)
{
if(!_errorCode)
{
m_Queue.pop_front();
if(!m_Queue.empty()) // more to do?
boost::asio::async_write(*socket().get(), boost::asio::buffer(m_Queue.front(), m_Queue.front().length()), boost::bind(&TextSocket::writeHandler, this, _1, _2));
}
else
close();
}; // eo writeHandler
void TextSocket::sendLine(String _line)
{
Root::instance().ioService().post(boost::bind(&TextSocket::_sendLine, this, _line));
}; // eo sendLine
// _sendLine
void TextSocket::_sendLine(String _line)
{
// copy'n'queue
_line.append(m_EOL);
m_Queue.push_back(_line);
if(m_Queue.size() == 1) // previously an empty queue, must start write!
boost::asio::async_write(*socket().get(), boost::asio::buffer(m_Queue.front(), m_Queue.front().length()), boost::bind(&TextSocket::writeHandler, this, _1, _2));
}; // eo sendLine
关于上面的类需要注意的一些事情...它使用 boost::asio::post发送线路。这允许它全部发生在 ASIO 以线程安全的方式管理的线程上，并允许我们排队等待发送的行。这使得它具有很强的可扩展性。
我确信还有更多问题，也许我的代码没有帮助。我花了几天时间将它们拼凑起来并理解它，但我怀疑它是否真的有用。希望一些更好的头脑会扫过它然后说“天哪，这
关于c++ - 网络客户端模拟器设计，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/4767678/