seata是使用CompletableFuture来处理响应结果的。seata单独封装了MessageFuture类,用来包裹CompletableFuture:
public class MessageFuture {
private RpcMessage requestMessage;
private long timeout;
private long start = System.currentTimeMillis();
private transient CompletableFuture<Object> origin = new CompletableFuture<>();
}
在AbstractNettyRemoting抽象类中,seata定义了一个futures变量,用来存放MessageFuture信息:
protected final ConcurrentHashMap<Integer, MessageFuture> futures = new ConcurrentHashMap<>();
在AbstractNettyRemoting抽象类中,发送消息方法sendSyn会将创建MessageFuture实例,并将MessageFuture实例缓存futures中。
MessageFuture messageFuture = new MessageFuture();
messageFuture.setRequestMessage(rpcMessage);
messageFuture.setTimeout(timeoutMillis);
futures.put(rpcMessage.getId(), messageFuture);
channelWritableCheck(channel, rpcMessage.getBody());
String remoteAddr = ChannelUtil.getAddressFromChannel(channel);
doBeforeRpcHooks(remoteAddr, rpcMessage);
// 发送消息
channel.writeAndFlush(rpcMessage).addListener((ChannelFutureListener) future -> {
if (!future.isSuccess()) {
MessageFuture messageFuture1 = futures.remove(rpcMessage.getId());
if (messageFuture1 != null) {
messageFuture1.setResultMessage(future.cause());
}
destroyChannel(future.channel());
}
});
try {
// 获取响应结果,如果获取不到会一直等待,直到超时
Object result = messageFuture.get(timeoutMillis, TimeUnit.MILLISECONDS);
doAfterRpcHooks(remoteAddr, rpcMessage, result);
return result;
} catch (Exception exx) {
LOGGER.error("wait response error:{},ip:{},request:{}", exx.getMessage(), channel.remoteAddress(), rpcMessage.getBody());
if (exx instanceof TimeoutException) {
throw (TimeoutException) exx;
} else {
throw new RuntimeException(exx);
}
}
客户端在注册Processor时,会通过ClientOnResponseProcessor类,获取futures.
ClientOnResponseProcessor onResponseProcessor = new ClientOnResponseProcessor(mergeMsgMap, super.getFutures(), getTransactionMessageHandler());
这样,客户端在接收到服务端的消息时,会将响应结果set进MessageFuture中:
MessageFuture messageFuture = futures.remove(rpcMessage.getId());
if (messageFuture != null) {
// 设置响应结果,此时messageFuture.get(timeoutMillis, TimeUnit.MILLISECONDS);获取结果等待结束
messageFuture.setResultMessage(rpcMessage.getBody());
} else {
if (rpcMessage.getBody() instanceof AbstractResultMessage) {
if (transactionMessageHandler != null) {
transactionMessageHandler.onResponse((AbstractResultMessage) rpcMessage.getBody(), null);
}
}
}
标签:rpcMessage,seata,getBody,futures,MessageFuture,接收,messageFuture,客户端
From: https://www.cnblogs.com/demon001/p/17191218.html