当前位置： 首页 > ai >正文

Netty中future和promise用法和区别

ai 2025/7/26 18:39:10

定义与概念

  • Future：表示一个异步操作的结果。它是只读的，意味着你只能查看操作是否完成、是否成功、获取结果或者异常等信息，但不能主动设置操作的结果。
  • Promise：是 Future 的可写扩展。它不仅可以像 Future 一样查看操作结果，还能主动设置操作的成功、失败或者取消状态，并且通知所有的监听器。

用法示例

Future 的用法

Future 通常用于获取异步操作的结果，并且可以添加监听器来处理操作完成后的逻辑。以下是一个简单的示例，展示了如何使用 Future 来处理 DNS 解析结果：

dnsNameResolver.resolve(host).addListener(new FutureListener<InetAddress>() {@Overridepublic void operationComplete(Future<InetAddress> future) throws Exception {if (future.isSuccess()) {InetAddress hostAddress = future.get();// 处理解析成功的结果} else {// 处理解析失败的情况}}
});

在这个示例中，dnsNameResolver.resolve(host) 方法返回一个 Future<InetAddress> 对象，我们通过添加 FutureListener 来监听解析操作的完成状态。当操作完成后，会调用 operationComplete 方法，我们可以在这个方法中处理解析结果。

Promise 的用法

Promise 主要用于主动设置异步操作的结果，并且可以通知所有的监听器。以下是一个示例，展示了如何使用 Promise 来处理 OCSP 查询结果：

final Promise<OCSPResp> responsePromise = eventLoop.newPromise();// 异步操作
dnsNameResolver.resolve(host).addListener(new FutureListener<InetAddress>() {@Overridepublic void operationComplete(Future<InetAddress> future) throws Exception {if (future.isSuccess()) {// 处理解析成功的结果InetAddress hostAddress = future.get();final ChannelFuture channelFuture = bootstrap.connect(hostAddress, port);channelFuture.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) {if (future.isSuccess()) {// 处理连接成功的结果responsePromise.trySuccess(result); // 设置操作成功的结果} else {responsePromise.tryFailure(new IllegalStateException("Connection to OCSP Responder Failed", future.cause())); // 设置操作失败的结果}}});} else {responsePromise.tryFailure(future.cause()); // 设置操作失败的结果}}
});// 添加监听器来处理操作结果
responsePromise.addListener(new FutureListener<OCSPResp>() {@Overridepublic void operationComplete(Future<OCSPResp> future) throws Exception {if (future.isSuccess()) {OCSPResp resp = future.get();// 处理操作成功的结果} else {// 处理操作失败的情况}}
});

在这个示例中，我们首先创建了一个 Promise<OCSPResp> 对象 responsePromise，然后在异步操作完成后，根据操作结果调用 trySuccess 或 tryFailure 方法来设置 Promise 的状态。最后，我们添加了一个 FutureListener 来监听 Promise 的完成状态，并处理操作结果。

区别总结

  • 可写性：
    • Future 是只读的，只能查看异步操作的结果，不能主动设置操作的状态。
    • Promise 是可写的，可以主动设置操作的成功、失败或者取消状态。
  • 用途：
    • Future 主要用于获取异步操作的结果，并且可以添加监听器来处理操作完成后的逻辑。
    • Promise 主要用于在异步操作完成后，主动设置操作的结果，并且通知所有的监听器。
  • 方法差异：
    • Future 提供了一些方法来查看操作的状态，如 isDone()、isSuccess()、cause() 等。
    • Promise 除了继承了 Future 的方法外，还提供了一些方法来设置操作的结果，如 setSuccess()、trySuccess()、setFailure()、tryFailure() 等。

代码中的体现

在提供的代码片段中，InflightNameResolver 类的 resolve 方法使用了 Promise 来处理 DNS 解析结果：

private <U> Promise<U> resolve(final ConcurrentMap<String, Promise<U>> resolveMap,final String inetHost, final Promise<U> promise, boolean resolveAll) {// ...if (resolveAll) {@SuppressWarnings("unchecked")final Promise<List<T>> castPromise = (Promise<List<T>>) promise; // U is List<T>delegate.resolveAll(inetHost, castPromise);} else {@SuppressWarnings("unchecked")final Promise<T> castPromise = (Promise<T>) promise; // U is Tdelegate.resolve(inetHost, castPromise);}// ...return promise;
}

在这个方法中，我们可以看到 Promise 被用于传递异步操作的结果，并且可以在操作完成后主动设置操作的状态。

另外，PromiseNotifier 类展示了如何使用 Promise 来通知多个监听器：

public class PromiseNotifier<V, F extends Future<V>> implements GenericFutureListener<F> {private final Promise<? super V>[] promises;public PromiseNotifier(Promise<? super V>... promises) {this.promises = promises;}@Overridepublic void operationComplete(F future) throws Exception {if (future.isSuccess()) {V result = future.get();for (Promise<? super V> p : promises) {PromiseNotificationUtil.trySuccess(p, result, null);}} else if (future.isCancelled()) {for (Promise<? super V> p : promises) {PromiseNotificationUtil.tryCancel(p, null);}} else {Throwable cause = future.cause();for (Promise<? super V> p : promises) {PromiseNotificationUtil.tryFailure(p, cause, null);}}}
}

在这个类中，我们可以看到 Promise 被用于通知多个监听器操作的结果，并且可以根据操作的状态调用不同的方法来设置 Promise 的状态。

综上所述，Future 和 Promise 在 Netty 中都是非常重要的组件，它们分别用于处理异步操作的不同方面。通过合理使用 Future 和 Promise，可以有效地处理异步操作的结果，提高代码的可读性和可维护性。

处理多个顺序依赖的异步操作

假设我们需要完成一个包含三个步骤的操作流程：

  1. 连接到服务器
  2. 发送认证请求并等待认证成功
  3. 发送业务数据并接收响应

这三个步骤必须按顺序执行，后一个步骤依赖于前一个步骤的成功完成。以下是实现这种依赖关系的代码示例：

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;public class ChannelPromiseDependencyExample {private static final String SERVER_HOST = "localhost";private static final int SERVER_PORT = 8080;public static void main(String[] args) throws Exception {EventLoopGroup group = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new StringDecoder(),new StringEncoder(),new ClientHandler());}});// 创建主 Promise，用于跟踪整个操作流程的完成状态ChannelPromise mainPromise = bootstrap.config().group().next().newPromise();// 开始执行依赖操作链connectAndProcess(bootstrap, mainPromise);// 等待整个操作流程完成mainPromise.await();if (mainPromise.isSuccess()) {System.out.println("所有操作成功完成");} else {System.out.println("操作失败: " + mainPromise.cause());}} finally {group.shutdownGracefully();}}private static void connectAndProcess(Bootstrap bootstrap, ChannelPromise mainPromise) {// 步骤 1: 连接到服务器ChannelFuture connectFuture = bootstrap.connect(SERVER_HOST, SERVER_PORT);// 为连接操作添加监听器connectFuture.addListener((ChannelFuture future) -> {if (future.isSuccess()) {Channel channel = future.channel();System.out.println("成功连接到服务器");// 步骤 2: 发送认证请求ChannelPromise authPromise = channel.newPromise();sendAuthRequest(channel, authPromise);// 为认证操作添加监听器authPromise.addListener((ChannelFuture authFuture) -> {if (authFuture.isSuccess()) {System.out.println("认证成功");// 步骤 3: 发送业务数据ChannelPromise businessPromise = channel.newPromise();sendBusinessData(channel, businessPromise);// 为业务操作添加监听器businessPromise.addListener((ChannelFuture businessFuture) -> {if (businessFuture.isSuccess()) {System.out.println("业务数据处理成功");mainPromise.setSuccess(); // 标记整个操作成功} else {mainPromise.setFailure(businessFuture.cause()); // 标记整个操作失败}channel.close(); // 关闭连接});} else {mainPromise.setFailure(authFuture.cause()); // 标记整个操作失败channel.close(); // 关闭连接}});} else {mainPromise.setFailure(future.cause()); // 标记整个操作失败}});}private static void sendAuthRequest(Channel channel, ChannelPromise authPromise) {// 发送认证请求channel.writeAndFlush("AUTH username password").addListener(future -> {if (future.isSuccess()) {System.out.println("认证请求已发送");// 认证结果将在 ChannelHandler 中处理} else {authPromise.setFailure(future.cause()); // 认证请求发送失败}});}private static void sendBusinessData(Channel channel, ChannelPromise businessPromise) {// 发送业务数据channel.writeAndFlush("DATA some_business_data").addListener(future -> {if (future.isSuccess()) {System.out.println("业务数据已发送");// 业务响应将在 ChannelHandler 中处理} else {businessPromise.setFailure(future.cause()); // 业务数据发送失败}});}static class ClientHandler extends SimpleChannelInboundHandler<String> {private ChannelPromise authPromise;private ChannelPromise businessPromise;@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 通道激活时，可以获取外部的 Promise 实例// 实际应用中可能需要通过构造函数或其他方式传递}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {System.out.println("收到服务器响应: " + msg);// 根据响应内容判断操作结果if (msg.startsWith("AUTH_SUCCESS")) {if (authPromise != null) {authPromise.setSuccess(); // 认证成功}} else if (msg.startsWith("AUTH_FAILURE")) {if (authPromise != null) {authPromise.setFailure(new Exception("认证失败: " + msg)); // 认证失败}} else if (msg.startsWith("DATA_SUCCESS")) {if (businessPromise != null) {businessPromise.setSuccess(); // 业务数据处理成功}} else if (msg.startsWith("DATA_FAILURE")) {if (businessPromise != null) {businessPromise.setFailure(new Exception("业务数据处理失败: " + msg)); // 业务数据处理失败}}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();// 设置所有未完成的 Promise 为失败状态if (authPromise != null && !authPromise.isDone()) {authPromise.setFailure(cause);}if (businessPromise != null && !businessPromise.isDone()) {businessPromise.setFailure(cause);}ctx.close();}}
}

关键点解析

  1. 创建和使用 ChannelPromise：
    • 通过 EventLoop.newPromise() 或 Channel.newPromise() 创建 ChannelPromise 实例。
    • mainPromise 用于跟踪整个操作流程的完成状态。
  2. 处理依赖关系：
    • 使用 addListener() 方法为每个异步操作添加监听器。
    • 在前一个操作的监听器中检查操作结果，只有成功时才继续执行下一个操作。
    • 如果某个操作失败，立即设置主 Promise 为失败状态并终止后续操作。
  3. 在 ChannelHandler 中处理响应：
    • 在 ClientHandler 中接收服务器响应，并根据响应内容设置相应的 Promise 状态。
    • 这样可以将异步响应与对应的操作关联起来。
  4. 异常处理：
    • 在 exceptionCaught() 方法中捕获异常，并设置所有未完成的 Promise 为失败状态。

更复杂的依赖关系处理

对于更复杂的依赖关系，可以使用 PromiseCombiner 来组合多个 Promise，并在所有 Promise 都成功完成后执行后续操作。以下是一个使用 PromiseCombiner 的示例：

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;public class PromiseCombinerExample {public static void main(String[] args) throws Exception {EventLoopGroup group = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new StringDecoder(),new StringEncoder(),new SimpleChannelInboundHandler<String>() {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {System.out.println("收到消息: " + msg);}});}});// 连接到多个服务器ChannelFuture future1 = bootstrap.connect("server1.example.com", 8080);ChannelFuture future2 = bootstrap.connect("server2.example.com", 8080);ChannelFuture future3 = bootstrap.connect("server3.example.com", 8080);// 创建 PromiseCombiner 来组合多个 FuturePromiseCombiner combiner = new PromiseCombiner(group.next());combiner.add(future1);combiner.add(future2);combiner.add(future3);// 创建一个 Promise 来接收组合结果ChannelPromise allConnectedPromise = group.next().newPromise();combiner.finish(allConnectedPromise);// 为组合结果添加监听器allConnectedPromise.addListener(future -> {if (future.isSuccess()) {System.out.println("所有连接都已成功建立");// 执行后续操作} else {System.out.println("至少有一个连接失败: " + future.cause());}});// 等待所有操作完成allConnectedPromise.await();} finally {group.shutdownGracefully();}}
}

通过 ChannelPromise 和相关工具，我们可以在 Netty 中灵活处理多个异步操作的依赖关系：

  1. 顺序依赖：通过在前一个操作的监听器中启动下一个操作，实现顺序执行。
  2. 并行依赖：使用 PromiseCombiner 等工具组合多个并行操作，等待所有操作完成后执行后续逻辑。
  3. 异常处理：在每个步骤中正确处理异常，并传播给主 Promise。
  4. 状态管理：使用 Promise 跟踪每个操作的状态，确保操作按预期完成。

这种方式使得异步代码更加清晰和易于维护，避免了回调地狱，提高了代码的可读性和可维护性。

http://www.xdnf.cn/news/16210.html

相关文章：

  • 07 51单片机之定时器
  • 魔百和M401H_国科GK6323V100C_安卓9_不分地区免拆卡刷固件包
  • [RPA] Excel中的字典处理
  • 【C#学习Day12笔记】抽象类、密封类与子类构造(继承)
  • C语言————原码 补码 反码 （超绝详细解释）
  • 服务器安装虚拟机全步骤
  • KNN算法：从原理到实战全解析
  • selenium 元素定位
  • OpenCV（04）梯度处理，边缘检测，绘制轮廓，凸包特征检测，轮廓特征查找
  • 医疗器械：DFEMA和PFEMA
  • 零基础也能创作专属歌曲：文心一言+蘑兔AI协同教程
  • 前端学习日记（十三）
  • 在 Ansys CFX Pre 中配置 RGP 表的分步指南
  • ERNIE-4.5-0.3B 实战指南：文心一言 4.5 开源模型的轻量化部署与效能跃升
  • cocos creator 3.8.6 websocke的一直报错WebSocket is not a constructor
  • 武汉烽火民生汇，盛大启航
  • Nginx 安装与 HTTPS 配置指南：使用 OpenSSL 搭建安全 Web 服务器
  • 无印 v1.6 视频解析去水印工具，支持多个平台
  • C++ : list的模拟
  • Qwen-MT：翻得快，译得巧
  • HAProxy 原理及配置
  • Nacos-服务注册，服务发现（一）
  • VS2022专业版安装扩展
  • SpringBoot（黑马）
  • Hive【安装 01】hive-3.1.2版本安装配置（含 mysql-connector-java-5.1.47.jar 网盘资源）
  • 使用 FFmpeg 实现 RTP 音频传输与播放
  • 没有 Mac，如何上架 iOS App？多项目复用与流程标准化实战分享
  • CSS3新特性深度解析：Position Sticky粘性定位完整指南
  • 基于单片机的楼宇门禁系统的设计与实现
  • 三种深度学习模型（GRU、CNN-GRU、贝叶斯优化的CNN-GRU/BO-CNN-GRU）对北半球光伏数据进行时间序列预测