带你了解Java中的Mono接口

带你了解Java中的Mono接口

目录

一、Mono 接口的基本概念

二、Mono 与 Flux 的区别

三、Mono的使用

创建 Mono 的常见方法

Mono 的操作符与链式调用

错误处理与回退机制

订阅与消费 Mono

四、Mono性能优化与调试技巧

调试技巧

性能优化建议

五、Mono使用总结


一、Mono 接口的基本概念

Mono 是 Project Reactor 响应式编程库中的核心接口之一,代表一个异步的、可能包含零个或一个元素的流。在响应式编程范式中,Mono 专门用于处理那些预期最多只有一个结果的异步操作。它遵循 Reactive Streams 规范,支持非阻塞的回压机制,使得开发者能够以声明式的方式构建高效的异步应用程序。

Mono 的设计哲学基于发布-订阅模式,数据生产者通过 Mono 发布数据,而消费者订阅这些数据。这种模式使得系统组件之间实现了解耦,提高了代码的可维护性和可扩展性。


二、Mono 与 Flux 的区别

Mono 和 Flux 都是 Project Reactor 的核心抽象,但它们针对不同的场景:

  • Mono:处理零个或一个元素的异步序列。适合用于返回单个结果的异步操作,如查询数据库中的一条记录、调用返回单个对象的 REST API、或执行可能成功或失败但只产生单一结果的操作。
  • Flux:处理零个或多个元素的异步序列。适合流式数据、集合操作或可能返回多个结果的场景,如消息队列的消费、文件读取或数据库查询返回多条记录。

关键区别在于元素的预期数量:Mono 是至多一个,而 Flux 是零到多个


三、Mono的使用

创建 Mono 的常见方法

创建 Mono 有多种方式,适应不同的使用场景:

// 1. 从确定值创建 Mono<String> mono1 = Mono.just("Hello, World!"); // 2. 创建空 Mono Mono<String> mono2 = Mono.empty(); // 3. 从可能为null的值创建(如果value为null,则返回空Mono) Mono<String> mono3 = Mono.justOrNull(someNullableValue); // 4. 从可调用对象创建(延迟计算) Mono<String> mono4 = Mono.fromCallable(() -> { // 可能抛出异常的计算 return expensiveComputation(); }); // 5. 从 Future 创建 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Result"); Mono<String> mono5 = Mono.fromFuture(future); // 6. 从 Supplier 创建 Mono<String> mono6 = Mono.fromSupplier(() -> generateValue()); // 7. 创建错误 Mono Mono<String> mono7 = Mono.error(new RuntimeException("Something went wrong")); // 8. 延迟创建(直到有订阅者时才执行) Mono<String> mono8 = Mono.defer(() -> { return Mono.just(System.currentTimeMillis() + " - generated"); });

每种创建方法都有其适用场景:just()用于已知值,fromCallable()用于可能抛出异常的计算,defer()用于每次订阅都需要重新计算的情况。


Mono 的操作符与链式调用

常用操作符包括:

  • map:同步转换元素
  • flatMap:异步转换到另一个 Mono(或 Flux)
  • filter:过滤元素
  • zipWith:将两个 Mono 的结果组合
  • then:忽略当前结果,返回另一个 Mono

Mono 不仅提供了丰富的操作符,并且支持声明式的链式调用:

Mono<User> userMono = getUserById(userId) .filter(user -> user.isActive()) // 过滤非活跃用户 .map(user -> { // 转换数据 UserDTO dto = new UserDTO(); dto.setName(user.getName()); dto.setEmail(user.getEmail()); return dto; }) .flatMap(dto -> { // 异步转换到另一个Mono return sendNotification(dto.getEmail()) .map(success -> dto); }) .doOnNext(dto -> { // 副作用操作 log.info("Processed user: {}", dto.getName()); }) .doOnError(error -> { // 错误时的副作用 log.error("Failed to process user", error); }) .timeout(Duration.ofSeconds(5)) // 设置超时 .retry(3); // 失败时重试3次

错误处理与回退机制

Mono 提供了多种错误处理策略:

Mono<String> safeMono = riskyOperation() .onErrorResume(error -> { // 发生错误时提供备用Mono log.warn("Operation failed, using fallback", error); return Mono.just("Fallback Value"); }) .onErrorReturn("Default Value") // 发生错误时返回默认值 .onErrorMap(error -> { // 转换错误类型 return new BusinessException("Operation failed", error); }) .doFinally(signal -> { // 最终清理(无论成功或失败) cleanupResources(); });

错误处理的关键是选择适当的策略:

  • onErrorReturn:简单恢复,返回静态值
  • onErrorResume:动态恢复,可以基于错误类型返回不同的备用值
  • onErrorMap:转换错误类型,便于上层统一处理

订阅与消费 Mono

Mono 是惰性的,只有在订阅时才会开始执行,订阅的方式也有多种:

// 1. 最简单的订阅(触发执行但忽略结果) mono.subscribe(); // 2. 带成功回调的订阅 mono.subscribe( value -> System.out.println("Received: " + value) ); // 3. 带成功和错误回调的订阅 mono.subscribe( value -> System.out.println("Received: " + value), error -> System.err.println("Error: " + error.getMessage()) ); // 4. 完整的订阅(成功、错误、完成) mono.subscribe( value -> System.out.println("Received: " + value), error -> System.err.println("Error: " + error.getMessage()), () -> System.out.println("Completed successfully") ); // 5. 带订阅回调的订阅 mono.subscribe( new Subscriber<String>() { @Override public void onSubscribe(Subscription s) { s.request(1); // 请求数据 } @Override public void onNext(String value) { System.out.println("Received: " + value); } @Override public void onError(Throwable t) { System.err.println("Error: " + t.getMessage()); } @Override public void onComplete() { System.out.println("Completed"); } } );

四、Mono性能优化与调试技巧

调试技巧

Mono<String> debugMono = someOperation() .log("my.mono") // 添加日志 .checkpoint("debug point") // 添加检查点 .doOnSubscribe(s -> log.debug("Subscribed")) .doOnNext(v -> log.debug("Next: {}", v)) .doOnError(e -> log.error("Error: {}", e.getMessage())) .doOnCancel(() -> log.debug("Cancelled")) .doOnSuccess(v -> log.debug("Completed with: {}", v));

性能优化建议

  • 避免不必要的订阅:多个订阅会导致重复计算
  • 使用缓存:对不变的结果使用cache()操作符
  • 合理使用调度器:通过subscribeOn和publishOn控制执行线程
  • 背压处理:虽然 Mono 只有一个元素,但仍需注意背压传播
  • 及早错误处理:在链的早期处理错误,避免不必要的计算

五、Mono使用总结

核心优势

  1. 声明式编程:以声明式方式构建异步逻辑,提高代码可读性
  2. 非阻塞操作:充分利用系统资源,提高并发性能
  3. 强大的错误处理:提供丰富的错误恢复和转换机制
  4. 操作符丰富:提供函数式操作符,支持复杂的数据处理流程
  5. 背压支持:内置背压机制,防止生产者压垮消费者

Mono 作为响应式编程的基础构件,合理使用Mono能够构建出高性能、高可维护性的异步系统。


制作不易,如果对你有帮助请点赞,评论,收藏,感谢大家的支持

Read more

MCP客户端与服务端初使用——让deepseek调用查询天气的mcp来查询天气

MCP客户端与服务端初使用——让deepseek调用查询天气的mcp来查询天气

本系列主要通过调用天气的mcp server查询天气这个例子来学习什么是mcp,以及怎么设计mcp。话不多说,我们开始吧。主要参考的是B站的老哥做的一个教程,我把链接放到这里,大家如果有什么不懂的也可以去看一下。 https://www.bilibili.com/video/BV1NLXCYTEbj?spm_id_from=333.788.videopod.episodes&vd_source=32148098d54c83926572ec0bab6a3b1d https://blog.ZEEKLOG.net/fufan_LLM/article/details/146377471 最终的效果:让deepseek-v3使用天气查询的工具来查询指定地方的天气情况 技术介绍 MCP,即Model Context Protocol(模型上下文协议),是由Claude的母公司Anthropic在2024年底推出的一项创新技术协议。在它刚问世时,并未引起太多关注,反响较为平淡。然而,随着今年智能体Agent领域的迅猛发展,MCP逐渐进入大众视野并受到广泛关注。今年2月,

By Ne0inhk
可以在命令行通过大模型使用上下文协议(MCP)与外部工具交互的软件:小巧的MCPHost

可以在命令行通过大模型使用上下文协议(MCP)与外部工具交互的软件:小巧的MCPHost

小巧的MCPHost MCPHost 可以在命令行下使用,使大型语言模型(LLM)能够通过模型上下文协议(MCP)与外部工具进行交互。目前支持Claude 3.5 Sonnet和Ollama等。本次实践使用自己架设的Deepseek v3模型,跑通了Time MCP服务。  官网:GitHub - mark3labs/mcphost: A CLI host application that enables Large Language Models (LLMs) to interact with external tools through the Model Context Protocol (MCP). 下载安装 使用非常方便,直接下载解压即可使用。官网提供Windows、Linux和MacOS三个系统的压缩包: https://github.com/

By Ne0inhk
实战篇:Python开发monogod数据库mcp server看完你就会了

实战篇:Python开发monogod数据库mcp server看完你就会了

原创不易,请关注公众号:【爬虫与大模型开发】,大模型的应用开发之路,整理了大模型在现在的企业级应用的实操及大家需要注意的一些AI开发的知识点!持续输出爬虫与大模型的相关文章。 前言 目前mcp协议是给deepseek大模型插上工具链的翅膀,让大模型不仅拥有超高的推理和文本生成能力,还能具备执行大脑意识的工具能力! 如何开发一个mcp? mcp是一种协议,指的是模型上下文协议 (Model Context Protocol)。 官方结成的mcp https://github.com/modelcontextprotocol/python-sdk mcp库 pip install mcp from mcp.server.fastmcp import FastMCP 我们先来做一个简单的案例 from mcp.server.fastmcp import FastMCP import requests mcp = FastMCP("spider") @mcp.tool() def crawl(

By Ne0inhk
【大模型实战篇】基于Claude MCP协议的智能体落地示例

【大模型实战篇】基于Claude MCP协议的智能体落地示例

1. 背景         之前我们在《MCP(Model Context Protocol) 大模型智能体第一个开源标准协议》一文中,介绍了MCP的概念,虽然了解了其概念、架构、解决的问题,但还缺少具体的示例,来帮助进一步理解整套MCP框架如何落地。         今天我们基于claude的官方例子--获取天气预报【1】,来理解MCP落地的整条链路。 2. MCP示例         该案例是构建一个简单的MCP天气预报服务器,并将其连接到主机,即Claude for Desktop。从基本设置开始,然后逐步发展到更复杂的使用场景。         大模型虽然能力非常强,但其弊端就是内容是过时的,这里的过时不是说内容很旧,只是表达内容具有非实时性。比如没有获取天气预报和严重天气警报的能力。因此我们将使用MCP来解决这一问题。         构建一个服务器,该服务器提供两个工具:获取警报(get-alerts)和获取预报(get-forecast)。然后,将该服务器连接到MCP主机(在本例中为Claude for Desktop)。         首先我们配置下环

By Ne0inhk