onResponse处理回复

客户端发送请求的功能进行请求描述对服务器响应的处理机制并对返回信息

进行对应接收

意义针对客户端的每一条请求进行管理以便于对请求对应的响应做出合适操作。

对于客户端而言其通常是主动发起请求服务的一方。

然而在多线程网络通信中针对多个请求进行响应时可能会存在时序问题导致无法保证一个线程发送的请求后接收到的响应就是针对这条请求的响应这是非常危险的情况。

异步IO挑战类似于Muduo库这种异步IO网络通信库通常IO操作都是异步的即发送数据是将数据放入发送缓冲区而何时发送由底层网络库协调并且不提供recv接口而是连接触发可读事件后IO读取数据完成调用处理回调进行数据处理因此在发送请求后无法直接等待该条请求的响应。

解决方案

创建请求管理模块通过给每个请求设定一个请求ID来解决上述问题。

服务端响应时会标识出响应针对的是哪个请求即响应信息包含请求ID。

客户端无论收到

哪条请求的响应都将数据存储入hash_map中以请求ID作为映射并提供获取指定请求ID响应的阻塞接口。

这样只要知道自己的请求ID就能准确获取到想要的响应。

进一步优化可以将每个请求封装描述添加异步future控制或设置回调函数的方式不仅支持阻塞获取响应也能实现异步获取响应及

namespace

std::futureBaseMessage::ptr;struct

ptrstd::shared_ptrRequestDescribe;//智能指针

管理};//请求

描述进行CURD//增RequestDescribe::ptr

newDescribe(const

_mutex;std::unordered_mapstd::string,

RequestDescribe::ptr

std::shared_ptrRequestDescribe;BaseMessage::ptr

request;RType

rtype;std::promiseBaseMessage::ptr

callback;};

getDescribe(rid);//根据先获取msg-rid()

(rdp.get()

{rdp-response.set_value(msg);//调用

(rdp-rtype

{ELOG(请求类型未知);}delDescribe(rid);

onResponse处理回复

onResponse方法是对收到的消息进行处理的入口点。

当收到服务器的响应时该方法会被调用来匹配相应的请求描述(RequestDescribe)并通过请求类型(RType)来决定如何处理响应

异步请求(RType::REQ_ASYNC)则通过设置std::promise的值(response.set_value(msg))来完成对应的std::future使得调用者可以通过未来对象获取响应。

如果是带有回调的请求(RType::REQ_CALLBACK)则直接调用注册的回调函数(rdp-callback(msg))来处理响应。

如果请求类型未知则记录错误日志。

promise

nullptr){ELOG(构造请求描述对象失败);return

false;}conn-send(req);async_rsp

rdp-response.get_future();return

true;

nullptr){ELOG(构造请求描述对象失败);return

true;

基本参数所有三个send方法都接受两个相同的基本参数一个指向BaseConnection的智能指针conn表示网络连接和一个指向BaseMessage的智能指针req表示要发送的消息。

错误处理每个send方法在无法成功创建请求描述对象时都会记录错误日志并返回false指示操作失败。

消息发送无论哪种方式最终都是通过调用conn-send(req)来执行实际的消息发送。

send方法有三个重载版本用于通过网络连接conn发送请求消息req到服务器

async_rsp参数用于异步发送请求并返回一个std::future对象以便于后续获取响应。

不阻塞第二个send方法是同步的它

等待直到接收到服务器的响应并将结果赋值给BaseMessage::ptr

rsp。

第三个send方法允许用户在发送请求时提供一个回调函数const

RequestCallback

第一种方法需调用者主动通过future.get()获取结果可能导致阻塞。

第三种方法响应到达时自动调用回调函数处理结果无需主动获取。

编程模型

第一种更接近同步编程风格通过异步手段避免长时间阻塞。

第三种是典型的异步编程模型更适合处理并发任务和事件驱动架构。

灵活性与复杂性

第一种方法直观但可能引入复杂的依赖关系管理。

第三种方法灵活尤其适合链式异步操作但可能导致“回调地狱”增加代码维护难度。

id,请求

描述进行CURD//增RequestDescribe::ptr

newDescribe(const

RequestCallback()){std::unique_lockstd::mutex

lock(_mutex);//加锁RequestDescribe::ptr

std::make_sharedRequestDescribe();rd-request

req;rd-rtype

cb;}_request_desc.insert(std::make_pair(req-GetId(),

rd));//将id

rid){std::unique_lockstd::mutex

lock(_mutex);auto

rid){std::unique_lockstd::mutex

lock(_mutex);_request_desc.erase(rid);}

完整代码

std::futureBaseMessage::ptr;//异步处理信息struct

RequestDescribe

std::shared_ptrRequestDescribe;BaseMessage::ptr

request;RType

rtype;std::promiseBaseMessage::ptr

callback;};//请求

msg-GetId();RequestDescribe::ptr

rdp

{rdp-response.set_value(msg);}else

(rdp-rtype

{ELOG(请求类型未知);}delDescribe(rid);}

//对收到

async_rsp){RequestDescribe::ptr

rdp

nullptr){ELOG(构造请求描述对象失败);return

false;}conn-send(req);async_rsp

rdp-response.get_future();return

true;}//同步操作bool

nullptr){ELOG(构造请求描述对象失败);return

true;}//private://

RequestCallback()){std::unique_lockstd::mutex

lock(_mutex);RequestDescribe::ptr

std::make_sharedRequestDescribe();rd-request

req;rd-rtype

cb;}_request_desc.insert(std::make_pair(req-GetId(),

rd));//将id

rid){std::unique_lockstd::mutex

lock(_mutex);auto

rid){std::unique_lockstd::mutex

lock(_mutex);_request_desc.erase(rid);}private:std::mutex

_mutex;std::unordered_mapstd::string,

RequestDescribe::ptr

意义向用户提供进行RPC调用的模块。

这个模块相对简单主要功能是向外提供几个RPC调用接口内部实现向服务端发送请求并等待获取结果。

调用方式

同步调用发起调用后等到收到响应结果后返回。

异步调用发起调用后立即返回可以在需要的时候获取结果。

回调调用发起调用的同时设置结果的处理回调收到响应后自动对结果进行回调处理。

❗❗❗❗

requestor中的处理是针对BaseMessage进行处理的

#pragma

requestor中的处理是针对BaseMessage进行处理的//

RpcResponse里边的result进行的//1.bool

call(const

MessageFactory::createRpcRequest();req_msg-SetId(UUID::uuid());req_msg-setMethod(method);req_msg-setParams(params);req_msg-SetMType(MType::REQ_RPC);BaseMessage::ptr

rsp_msg;//

std::dynamic_pointer_castBaseMessage(req_msg),

rsp_msg);if

std::dynamic_pointer_castRpcResponse(rsp_msg);if

(!rpc_rsp_msg){ELOG(rpc响应向下类型转换失败);return

false;}if

RCode::RCODE_OK){ELOG(rpc请求出错%s,

errReason(rpc_rsp_msg-rcode()));return

false;}result

rpc_rsp_msg-result();DLOG(结果设置完毕);return

true;}////2.bool

向服务器发送异步回调请求设置回调函数回调函数中会传入一个promise对象在回调函数中去堆promise设置数据auto

req_msg

MessageFactory::createRpcRequest();req_msg-SetId(UUID::uuid());req_msg-SetMType(MType::REQ_RPC);req_msg-setMethod(method);req_msg-setParams(params);auto

json_promise

std::make_sharedstd::promiseJson::Value();result

json_promise-get_future();Requestor::RequestCallback

std::bind(RpcCaller::Callback,this,

json_promise,

std::dynamic_pointer_castBaseMessage(req_msg),

cb);if

MessageFactory::createRpcRequest();req_msg-SetId(UUID::uuid());req_msg-SetMType(MType::REQ_RPC);req_msg-setMethod(method);req_msg-setParams(params);Requestor::RequestCallback

req_cb

std::bind(RpcCaller::Callback1,this,

cb,

std::dynamic_pointer_castBaseMessage(req_msg),

req_cb);if

std::dynamic_pointer_castRpcResponse(msg);if

(!rpc_rsp_msg){ELOG(rpc响应向下类型转换失败);return;}if

(rpc_rsp_msg-rcode()

RCode::RCODE_OK){ELOG(rpc回调请求出错%s,

errReason(rpc_rsp_msg-rcode()));return;}cb(rpc_rsp_msg-result());}void

Callback(std::shared_ptrstd::promiseJson::Value

result,

std::dynamic_pointer_castRpcResponse(msg);if

(!rpc_rsp_msg){ELOG(rpc响应向下类型转换失败);return;}if

(rpc_rsp_msg-rcode()

RCode::RCODE_OK){ELOG(rpc异步请求出错%s,

errReason(rpc_rsp_msg-rcode()));return;}result-set_value(rpc_rsp_msg-result());}private:Requestor::ptr

_requestor;};}

三个重载的call方法这些方法提供了不同的方式来发起RPC调用并处理从服务器返回的响应。

每个call方法根据其参数和使用场景的不同具有特定的功能和适用性

同步调用

创建并配置一个RpcRequest消息。

使用_requestor-send()发送请求并阻塞等待响应。

接收到响应后解析结果并检查是否有错误发生。

将结果设置到输出参数result中

应用场景适用于需要立即获取结果且可以接受当前线程被阻塞的情况。

异步调用

目的异步地发起一个RPC请求不阻塞当前线程允许后续通过std::future机制获取结果。

实现

创建并配置一个RpcRequest消息。

绑定一个回调函数Callback用于在接收到响应时设置std::promise的值。

使用_requestor-send()发送请求并立即返回非阻塞。

结果

应用场景适合于那些希望避免阻塞主线程但仍然需要明确获取结果的场景。

回调调用

目的异步发起RPC请求并在接收到响应时自动调用用户提供的回调函数进行处理。

实现

创建并配置一个RpcRequest消息。

绑定一个回调函数Callback1该回调会在接收到响应时被调用并进一步调用用户提供的回调函数处理结果。

使用_requestor-send()发送请求并立即返回非阻塞。

就回调