c – 提升异步tcp客户端

C&C++ 前端之家
前端之家收集整理的这篇文章主要介绍了c – 提升异步tcp客户端，前端之家小编觉得挺不错的，现在分享给大家，也给大家做个参考。
我刚刚开始使用boost.
我正在使用异步套接字编写TCP客户端服务器.

任务如下：

>客户端向服务器发送一个号码
>客户端可以在收到服务器的答案之前发送另一个nubmer.
>服务器接收一个号码,用它做一些计算并将结果发送回客户端.
>可以将多个客户端连接到服务器.

现在工作如下

>从客户端向服务器发送号码
>服务器在当前线程中收到一个数字并在OnReceive处理程序中计算(我知道这很糟糕……但我应该如何开始一个新的线程来并行执行计算)
>服务器发回应答但客户端已断开连接

如何允许客户端从键盘输入数字并同时等待服务器的回答？

为什么我的客户不等待服务器的答案？

客户端代码：

using boost::asio::ip::tcp;

class TCPClient
{
    public:
        TCPClient(boost::asio::io_service& IO_Service,tcp::resolver::iterator EndPointIter);
        void Close();

    private:
        boost::asio::io_service& m_IOService;
        tcp::socket m_Socket;

        string m_SendBuffer;
        static const size_t m_BufLen = 100;
        char m_RecieveBuffer[m_BufLen*2];

        void OnConnect(const boost::system::error_code& ErrorCode,tcp::resolver::iterator EndPointIter);
        void OnReceive(const boost::system::error_code& ErrorCode);
        void OnSend(const boost::system::error_code& ErrorCode);
        void DoClose();
};

TCPClient::TCPClient(boost::asio::io_service& IO_Service,tcp::resolver::iterator EndPointIter)
: m_IOService(IO_Service),m_Socket(IO_Service),m_SendBuffer("")
{
    tcp::endpoint EndPoint = *EndPointIter;

    m_Socket.async_connect(EndPoint,boost::bind(&TCPClient::OnConnect,this,boost::asio::placeholders::error,++EndPointIter));
}

void TCPClient::Close()
{
    m_IOService.post(
        boost::bind(&TCPClient::DoClose,this));
}
void TCPClient::OnConnect(const boost::system::error_code& ErrorCode,tcp::resolver::iterator EndPointIter)
{
    cout << "OnConnect..." << endl;
    if (ErrorCode == 0)
    {
        cin >> m_SendBuffer;
        cout << "Entered: " << m_SendBuffer << endl;
        m_SendBuffer += "\0";

        m_Socket.async_send(boost::asio::buffer(m_SendBuffer.c_str(),m_SendBuffer.length()+1),boost::bind(&TCPClient::OnSend,boost::asio::placeholders::error));
    } 
    else if (EndPointIter != tcp::resolver::iterator())
    {
        m_Socket.close();
        tcp::endpoint EndPoint = *EndPointIter;

        m_Socket.async_connect(EndPoint,++EndPointIter));
    }
}

void TCPClient::OnReceive(const boost::system::error_code& ErrorCode)
{
    cout << "receiving..." << endl;
    if (ErrorCode == 0)
    {
        cout << m_RecieveBuffer << endl;

        m_Socket.async_receive(boost::asio::buffer(m_RecieveBuffer,m_BufLen),boost::bind(&TCPClient::OnReceive,boost::asio::placeholders::error));
    } 
    else 
    {
        cout << "ERROR! OnReceive..." << endl;
        DoClose();
    }
}

void TCPClient::OnSend(const boost::system::error_code& ErrorCode)
{
    cout << "sending..." << endl;
    if (!ErrorCode)
    {
        cout << "\""<< m_SendBuffer <<"\" has been sent" << endl;
        m_SendBuffer = "";

        m_Socket.async_receive(boost::asio::buffer(m_RecieveBuffer,boost::asio::placeholders::error));
    }
    else
    {
        cout << "OnSend closing" << endl;
        DoClose();
    }

}

void TCPClient::DoClose()
{
    m_Socket.close();
}

int main()
{
    try 
    {
        cout << "Client is starting..." << endl;
        boost::asio::io_service IO_Service;

        tcp::resolver Resolver(IO_Service);

        string port = "13";
        tcp::resolver::query Query("127.0.0.1",port);

        tcp::resolver::iterator EndPointIterator = Resolver.resolve(Query);

        TCPClient Client(IO_Service,EndPointIterator);

        cout << "Client is started!" << endl;

        cout << "Enter a query string " << endl;

        boost::thread ClientThread(boost::bind(&boost::asio::io_service::run,&IO_Service));

        Client.Close();
        ClientThread.join();
    } 
    catch (exception& e)
    {
        cerr << e.what() << endl;
    }

    cout << "\nClosing";
    getch();
}

这是控制台的输出

Client is starting...
Client is started!
OnConnect...
12
Entered: 12
sending...
"12" has been sent
receiving...
ERROR! OnReceive...

Closing

服务器部分

class Session
{
    public:
        Session(boost::asio::io_service& io_service)
            : socket_(io_service)
        {
            dataRx[0] = '\0';
            dataTx[0] = '\0';
        }

        tcp::socket& socket()
        {
            return socket_;
        }

        void start()
        {
            socket_.async_read_some(boost::asio::buffer(dataRx,max_length),boost::bind(&Session::handle_read,boost::asio::placeholders::bytes_transferred));
        }

        void handle_read(const boost::system::error_code& error,size_t bytes_transferred)
        {
            cout << "reading..." << endl;
            cout << "Data: " << dataRx << endl;

            if (!error)
            {
                if (!isValidData())
                {
                    cout << "Bad data!" << endl;
                    sprintf(dataTx,"Bad data!\0");
                    dataRx[0] = '\0';
                }
                else
                {
                    sprintf(dataTx,getFactorization().c_str());
                    dataRx[0] = '\0';
                }

                boost::asio::async_write(socket_,boost::asio::buffer(dataTx,max_length*2),boost::bind(&Session::handle_write,boost::asio::placeholders::error));
            }
            else
            {
                delete this;
            }
        }

        void handle_write(const boost::system::error_code& error)
        {
            cout << "writing..." << endl;
            if (!error)
            {
                cout << "dataTx sent: " << dataTx << endl;
                dataTx[0] = '\0';

                socket_.async_read_some(boost::asio::buffer(dataRx,boost::asio::placeholders::bytes_transferred));
            }
            else
            {
                delete this;
            }
        }

        string getFactorization() const
        {
            //Do something
        }

        bool isValidData()
        {
            locale loc; 
            for (int i = 0; i < strlen(dataRx); i++)
                if (!isdigit(dataRx[i],loc))
                    return false;

            return true;
        }

    private:
        tcp::socket socket_;
        static const size_t max_length = 100;
        char dataRx[max_length];
        char dataTx[max_length*2];
};

class Server
{
    public:
        Server(boost::asio::io_service& io_service,short port)
            : io_service_(io_service),acceptor_(io_service,tcp::endpoint(tcp::v4(),port))
        {
            Session* new_session = new Session(io_service_);
            acceptor_.async_accept(new_session->socket(),boost::bind(&Server::handle_accept,new_session,boost::asio::placeholders::error));
        }

        void handle_accept(Session* new_session,const boost::system::error_code& error)
        {
            if (!error)
            {
                new_session->start();
                new_session = new Session(io_service_);
                acceptor_.async_accept(new_session->socket(),boost::asio::placeholders::error));
            }
            else
            {
                delete new_session;
            }
        }

    private:
        boost::asio::io_service& io_service_;
        tcp::acceptor acceptor_;
};

int main(int argc,char* argv[])
{
    cout << "Server is runing..." << endl;
    try
    {
        boost::asio::io_service io_service;

        int port = 13;
        Server s(io_service,port);
        cout << "Server is run!" << endl;
        io_service.run();
    }
    catch (boost::system::error_code& e)
    {
        std::cerr << e << "\n";
    }
    catch (std::exception& e)
    {
        std::cerr << "Exception: " << e.what() << "\n";
    }

    return 0;
}

服务器的输出

Server is runing...
Server is run!
reading...
Data: 12
writing...
dataTx sent: 13    //just send back received ++number
reading...
Data:

非常感谢您的帮助

========

添加

好的,我明白了.但检查ErrorCode == boost :: asio :: error :: eof不起作用……我做错了什么？

else if (ErrorCode == boost::asio::error::eof)
{
    cout << "boost::asio::error::eof in OnReceive!" << endl;
}
else 
{
    cout << "ERROR! OnReceive..." << ErrorCode << endl;
    DoClose();
}

打印输出错误！ OnReceive …系统：10009似乎是我的比较是不正确的

========

添加

我找到了根本原因.我已声明使用async_receive(而不是async_read_some)并将main中的行转换为

ClientThread.join();
Client.Close();

现在它工作正常！

现在我正在尝试同时从/向套接字读取和写入数据(因为客户端应该能够在收到服务器的回复之前发送其他请求.

在OnConnect函数中,我创建了boost线程：

boost::thread addMsgThread(boost::bind(&TCPClient::addMsgLoop,this));
boost::thread receivingThread(boost::bind(&TCPClient::startReceiving,this));
boost::thread sendingThread(boost::bind(&TCPClient::startSending,this));

与实施

void TCPClient::startReceiving()
{
    cout << "receiving..." << endl;
    m_RecieveBuffer[0] = '\0';
    m_Socket.async_receive(boost::asio::buffer(m_RecieveBuffer,boost::bind(&TCPClient::receivingLoop,boost::asio::placeholders::error)); //runtime error here
    cout << "m_RecieveBuffer = " << m_RecieveBuffer << endl;
}

void TCPClient::receivingLoop(const boost::system::error_code& ErrorCode)
{
    cout << "receiving..." << endl;
    if (ErrorCode == 0)
    {
        cout << "m_RecieveBuffer = " << m_RecieveBuffer << endl;

        m_Socket.async_receive(boost::asio::buffer(m_RecieveBuffer,boost::asio::placeholders::error));
    }
    else 
    {
        cout << "ERROR! receivingLoop..." << ErrorCode << endl;
        DoClose();
    }
}

void TCPClient::addMsgLoop()
{
    while (true)
    {
        string tmp;
        cin >> tmp;

        cout << "Entered: " << tmp << endl;
        tmp += "\0";

        try
        {
            msgQueue.push(tmp);
        }
        catch(exception &e)
        {
            cerr << "Canno add msg to send queue... " << e.what() << endl;
        }
    }
}

接收和发送线程的问题是相同的：运行时错误(在boost库中的某处写入访问冲突).

void TCPClient::startReceiving()
{
     ...
     m_Socket.async_receive(); //runtime error here
}

在后续版本中一切正常(但我不知道如何在回答之前实现多个发送).
任何人都可以告诉我如何解决问题或如何通过另一种方式实现这个问题？可能是汇集可以帮助,但我现在肯定这是好方法.

解决方法

顾名思义,boost::asio::ip::tcp::socket::async_read_some不保证能够读取完整的数据.当客户端完成写入时,它将error对象设置为boost :: asio :: error :: eof.

你得到的错误是因为：

服务器部分

if (!error)
        {
            ...
        }
        else
        {
            delete this;
        }

在else块中,您假设这是一个错误情况并关闭连接.这并非总是如此.在其他之前你需要检查错误== boost :: asio :: error :: eof.

除了读取处理程序中的这个,你应该继续收集缓冲区中读取的内容,直到你遇到error == boost :: asio :: error :: eof.只有这样,您才应验证读取数据并写回客户端.

请参阅examples部分中的HTTP服务器1,2,3实现.

更新：回答更新的问题

您有更新的代码的线程同步问题.

> msgQueue可以从两个或多个线程同时访问,无需任何锁定.
>可以同时调用同一个套接字上的读写.

如果我理解你的问题,你想：

>获取用户输入并将其发送到服务器.
>同时保持接收服务器的响应.

您可以使用两个boost::asio::io_service::strands执行这两项任务.使用Asio时,strands是同步任务的方式. Asio确保在一个链中发布的任务是同步执行的.

>在strand1中发布一个类似于以下内容的发送任务：read_user_input – > send_to_server – > handle_send – > read_user_input
>在strand2中发布一个看起来像的读取任务：read_some – > handle_read – > read_some

这将确保不会从两个线程同时访问msgQueue.使用两个套接字对服务器进行读写操作,以确保不会在同一个套接字上调用同时读写.

猜你在找的C&C++相关文章

C++11新特性的一些用法举例②
/** C+⬑ * 默认成员函数 原来C++类中，有6个默认成员函数： 构造函数 析...
C++特殊类的设计与单例模式
#pragma once // 1. 设计一个不能被拷贝的类/* 解析:拷贝只会放生在两个场景中：拷贝构造函...
C++11的类型转换
C类型转换 C语言:显式和隐式类型转换 隐式类型转化：编译器在编译阶段自动进行，能转就转，...
C++异常的基本概念与用法
//异常的概念/*抛出异常后必须要捕获,否则终止程序(到最外层后会交给main管理,main的行为就...
C++的智能指针
#pragma once /*Smart pointer 智能指针;灵巧指针 智能指针三大件//1.RAII//2.像指针一样使...
C++11标准库 原子变量 <atomic> 梳理
目录&lt;atomic&gt;原子操作的概念CAS实现原理CAS操作的伪代码：使用CAS完成变量的...
C++11新特性的一些用法举例①
//字符串字面量/*常用:1.原始字符串字面量 括号内保持原样输出 没有转义字符,如n不再是换行...
C++11标准库 条件变量 <condition_variable> 梳理
目录&lt;condition_variable&gt;condition_variable类类方法生产者消费者模型 -- ...
C++11智能指针 unique_ptr、shared_ptr/weak_ptr、make_shared、循环引用、定制删除器 (万字长文)
目录智能指针场景引入 - 为什么需要智能指针？内存泄漏什么是内存泄漏内存泄漏的危害内存泄...
C++多态
/** 多态/动态调用 * * * 1.虚函数/虚拟函数 * 语法:在函数前面加上virtual * * 1.1虚函数...
PHPJavaJava SEPythonC#C&C++RubyVBasp.NetGoPerlnettyDjangoDelphiJsp.NET CoreSpringFlaskSpringbootSpringMVCLuaLaravelMybatisAspGroovyThinkPHPYiiswoole
文件时间pythonm相等性PHP Warning时间问题问题解决pcntl_signal采样点wav模块动态文本调用频率限制对外暴露多个访问请求更新数据表模型结构type()方法比较速度手写体sobel算子保存模型Image类nn.Conv2dpytorch1.0kaggleDCGAN交并比range()用法打印模型反卷积卷积