Redisson 四种 API 模型全解：同步、异步、Reactive 与 RxJava3 实战指南

Redisson 四种 API 模型全解同步、异步、Reactive 与 RxJava3 实战指南【免费下载链接】redissonRedisson: Valkey Redis Java Client and Real-Time Data Platform. Sync/Async/RxJava/Reactive API. Over 50 Valkey and Redis based Java objects and services: Set, Multimap, SortedSet, Map, List, Queue, Deque, Semaphore, Lock, AtomicLong, Map Reduce, Bloom filter, Spring, Tomcat, Scheduler, JCache API, Hibernate, RPC, local cache..项目地址: https://gitcode.com/GitHub_Trending/re/redissonRedisson 为同一套分布式对象与 Redis/Valkey 数据服务提供了四种并存的编程模型同步Synchronous、异步Asynchronous、Reactive基于 Project Reactor与 RxJava3。本文以 docs/api-models.md 为核心骨架结合仓库源码讲解四种模型的入口接口、返回值类型、回调与订阅写法及线程安全注意事项帮助你在命令式、CompletableFuture 风格与响应式流三种技术栈之间自由切换而无须重写业务逻辑。一、四种 API 模型总览Redisson 的分布式对象AtomicLong、Map、Lock、Queue、Set 等 50 余种数据结构与服务几乎都以四种形态暴露给开发者API 模型入口接口返回值类型适用风格同步SynchronousRedissonClient直接返回结果命令式、阻塞当前线程异步AsynchronousRedissonClientRFutureV继承CompletionStage回调 / 链式异步ReactiveRedissonReactiveClientMonoT/FluxTProject Reactor 响应式流RxJava3RedissonRxClientSingleT/MaybeT/FlowableTRxJava3 响应式流四种模型共享同一套分布式语义与底层连接管理同步与异步入口都是 RedissonClientReactive 与 RxJava 则分别通过RedissonClient.reactive()与RedissonClient.rxJava()两个工厂方法获得。二、同步与异步 API2.1 Redisson 实例是线程安全的Redisson 实例被设计为完全线程安全同一个RedissonClient可以被多个线程并发共享用于获取任意数量的分布式对象。因此你只需在应用启动时创建一次客户端所有业务线程共用即可无需按线程创建实例也无需额外的同步措施。2.2 同步接口继承异步接口同步与异步 API 都通过 RedissonClient 接口访问。Redisson 的大部分对象同时实现了同步接口而该同步接口本身继承自对应的异步接口因此每个同步方法都天然带有一个xxxAsync()的异步孪生方法。以RAtomicLong为例其接口声明见 RAtomicLong.javapublic interface RAtomicLong extends RExpirable, RAtomicLongAsync { long get(); long incrementAndGet(); boolean compareAndSet(long expect, long update); // ... }而 RAtomicLongAsync.java 定义了与同步方法一一对应的异步版本返回值统一包装为RFuturepublic interface RAtomicLongAsync extends RExpirableAsync { RFutureLong getAsync(); RFutureLong incrementAndGetAsync(); RFutureBoolean compareAndSetAsync(long expect, long update); // ... }从源码结构可以看到同步与异步方法的差异仅在于同步方法直接返回结果并阻塞当前线程异步方法返回RFuture且不阻塞。文档中的对照示例// RAtomicLong extends RAtomicLongAsync RAtomicLong obj client.getAtomicLong(myLong); obj.compareAndSet(1, 401); RAtomicLongAsync objAsync client.getAtomicLong(myLong); RFutureBoolean future objAsync.compareAndSetAsync(1, 401);2.3 RFuture 是 CompletionStage 的扩展异步方法返回的 RFuture 接口同时继承了java.util.concurrent.FutureV与java.util.concurrent.CompletionStageVpublic interface RFutureV extends java.util.concurrent.FutureV, CompletionStageV {因此它既可以用Future的方式阻塞get()也可以使用CompletionStage提供的一整套声明式回调方法。文档给出的两种典型回调写法future.whenComplete((res, exception) - { // handle both result and exception }); // or future.thenAccept(res - { // handle result }).exceptionally(exception - { // handle exception });其中whenComplete同时接收结果与异常适合统一的收尾逻辑thenAccept只处理成功结果配合exceptionally单独分流异常处理风格更贴近CompletableFuture。2.4 重要不要在 RFuture 监听器中调用阻塞方法文档中特别给出了如下警告Avoid using blocking methods in RFuture listeners. Listeners are executed by netty threads, and delays in them may cause errors in Redis or Valkey request/response processing.原因在于监听器回调运行在 Netty 事件循环线程上。如果在回调中直接执行阻塞方法如future.get()、Thread.sleep()、同步 JDBC 查询等会阻塞 Netty 线程进而拖慢乃至中断 Redis/Valkey 的请求与响应处理——这是 Redisson 高性能路径上的一个典型陷阱。正确的做法是使用xxxAsync变体把回调调度到独立的Executor上执行future.whenCompleteAsync((res, exception) - { // handle both result and exception }, executor); // or future.thenAcceptAsync(res - { // handle result }, executor).exceptionallyAsync(exception - { // handle exception }, executor);带Async后缀的方法whenCompleteAsync、thenAcceptAsync、exceptionallyAsync接受一个Executor参数将阻塞逻辑从 Netty 线程池移出既保持回调链式风格又不影响请求处理吞吐。三、Reactive API基于 Project Reactor3.1 入口与获取方式Reactive API 通过 RedissonReactiveClient 接口访问其实现基于 Project Reactor。在 Redisson.java 中它由同步客户端工厂方法生成Override public RedissonReactiveClient reactive() { return new RedissonReactive(connectionManager, evictionScheduler, writeBehindService); }即先创建同步RedissonClient再调用redissonClient.reactive()得到 RedissonReactive 实例该实例复用同一套连接管理器connectionManager、驱逐调度器与写后置服务。3.2 使用示例文档给出的 Reactive 用法示例RedissonReactiveClient client redissonClient.reactive(); RAtomicLongReactive atomicLong client.getAtomicLong(myLong); MonoBoolean cs atomicLong.compareAndSet(10, 91); MonoLong get atomicLong.get(); get.doOnSuccess(res - { // ... }).subscribe();Reactive 模型的关键特征所有方法返回MonoT单值或FluxT多值等 Reactor 类型方法调用本身不触发执行必须通过.subscribe()等操作符真正订阅后才会向 Redis/Valkey 发送命令支持doOnSuccess、map、flatMap、zip、retry等完整的 Reactor 操作符便于在响应式链路中编排多个分布式操作。对应对象的接口为 RAtomicLongReactive继承RExpirableReactive其原子操作均返回Mono类型。四、RxJava API基于 RxJava34.1 入口与获取方式RxJava API 通过RedissonRxClient接口访问实现基于 RxJava3。同样在 Redisson.java 中提供工厂方法Override public RedissonRxClient rxJava() { return new RedissonRx(connectionManager, evictionScheduler, writeBehindService); }生成的 RedissonRx 与 Reactive 实现共享相同的底层组件只是对外暴露 RxJava3 类型。4.2 使用示例文档给出的 RxJava 用法示例RedissonRxClient client redissonClient.rxJava(); RAtomicLongRx atomicLong client.getAtomicLong(myLong); SingleBoolean cs atomicLong.compareAndSet(10, 91); SingleLong get atomicLong.get(); get.doOnSuccess(res - { // ... }).subscribe();RxJava 模型的关键特征单值操作返回SingleT可能无值的操作返回MaybeT多值流返回FlowableT/ObservableT与 Reactor 类似需要.subscribe()后命令才会真正执行支持doOnSuccess、map、flatMap、zipWith、retryWhen等 RxJava3 操作符。对应对象的接口为 RAtomicLongRx继承RExpirableRx。五、四种模型的选型建议场景推荐模型理由传统命令式业务代码、简单读写同步 API代码直观、易调试阻塞成本在普通业务中可接受高并发 IO 密集、希望避免线程阻塞异步 APIRFuture/CompletionStage回调链式组合可复用CompletableFuture技能全链路响应式架构如 Spring WebFluxReactive API返回Mono/Flux与 Reactor 生态无缝衔接已采用 RxJava3 的技术栈RxJava API返回Single/Flowable等保持技术栈统一无论选择哪一种底层都复用同一套连接管理与分布式对象实现四种模型之间的分布式语义原子性、过期时间、锁行为等完全一致可根据团队技术栈随时切换而无需担心行为差异。需要注意的共性是回调与订阅线程上的阻塞必须谨慎处理——异步模型的监听器运行在 Netty 线程上应避免在其中执行阻塞操作Reactive 与 RxJava 模型则要留意订阅生命周期管理与背压。六、相关文档导航开始使用 Redisson客户端创建与配置配置详解连接、编解码、线程池等数据序列化与编解码器分布式锁与同步器流水线Pipelining批处理以上文档覆盖客户端的启动配置、底层连接模型与各类分布式对象的详细用法与本文的四种 API 模型配合使用即可构建完整的 Redisson 应用。【免费下载链接】redissonRedisson: Valkey Redis Java Client and Real-Time Data Platform. Sync/Async/RxJava/Reactive API. Over 50 Valkey and Redis based Java objects and services: Set, Multimap, SortedSet, Map, List, Queue, Deque, Semaphore, Lock, AtomicLong, Map Reduce, Bloom filter, Spring, Tomcat, Scheduler, JCache API, Hibernate, RPC, local cache..项目地址: https://gitcode.com/GitHub_Trending/re/redisson创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考