Jedis 和 Lettuce 异步能力

Posted 2023-03-26

【中文标题】Jedis 和 Lettuce 异步能力

【英文标题】：Jedis and Lettuce async abilities

【发布时间】：2015-12-27 18:10:10

【问题描述】：

我在 Akka 中使用 redis，所以我不需要阻塞调用。生菜内置了异步未来调用。但是 Jedis 是 Redis 推荐的客户端。有人可以告诉我我是否以正确的方式使用它们。如果是的话，哪个更好。

绝地 我正在使用静态 Jedis 连接池来获取 con 并使用 Akka 未来回调来处理结果。我在这里担心的是，当我使用另一个线程（可调用）来获得线程最终会阻塞结果的结果时。虽然生菜可能有一些更有效的方法来做到这一点。

 private final class OnSuccessExtension extends OnSuccess<String> 
            private final ActorRef senderActorRef;
            private final Object message;
            @Override
            public void onSuccess(String valueRedis) throws Throwable 
                log.info(getContext().dispatcher().toString());
                senderActorRef.tell((String) message, ActorRef.noSender());
            
             public OnSuccessExtension(ActorRef senderActorRef,Object message) 
                    this.senderActorRef = senderActorRef;
                    this.message=message;
                
        
        ActorRef senderActorRef = getSender(); //never close over a future
            if (message instanceof String) 
        Future<String> f =akka.dispatch.Futures.future(new Callable<String>() 
                    public String call() 
                        String result;
                        try(Jedis jedis=JedisWrapper.redisPool.getResource()) 
                            result = jedis.get("name");
                        
                        return result;
                    
                , ex);
                f.onSuccess(new OnSuccessExtension(senderActorRef,message), ex);
    

生菜

ExecutorService executorService = Executors.newFixedThreadPool(10);
public void onReceive(Object message) throws Exception 
        ActorRef senderActorRef = getSender(); //never close over a future
        if (message instanceof String) 

            final RedisFuture<String> future = lettuce.connection.get("name");
            future.addListener(new Runnable() 
                final ActorRef sender = senderActorRef;
                final String msg =(String) message;
                @Override
                public void run() 
                    try 
                        String value = future.get();
                        log.info(value);
                        sender.tell(message, ActorRef.noSender());
                     catch (Exception e) 
                    
                
            , executorService);

如果生菜是异步调用的更好选择。那么我应该在生产环境中使用哪种类型的执行器。如果可能的话，我可以使用 Akka 调度程序作为 Letture 未来调用的执行上下文。

【参考方案1】：

您的问题没有一个答案，因为它取决于。

Jedis 和 lettuce 都是成熟的客户。为了完善 Java 客户端列表，还有 Redisson，它添加了另一层抽象（Collection/Queue/Lock/... 接口而不是原始 Redis 命令）。

这在很大程度上取决于您如何与客户合作。一般来说，Jedis（连接到 redis 的基于 java 的客户端）在数据访问方面是单线程的，因此您通过并发获得的唯一好处是将协议和 I/O 工作卸载到不同的线程。这对 lettuce 和 Redisson 来说并不完全正确，因为它们在底层使用了 netty（netty 将一个套接字通道绑定到一个特定的事件循环线程）。

使用 Jedis，您一次只能使用一个连接和一个线程。这与 Akka Actor 模型有很好的相关性，因为一个 Actor 实例一次只被一个线程占用。

另一方面，您需要与处理特定参与者的线程一样多的 Jedis 连接。如果您开始在不同的 Actor 之间共享 Jedis 连接，您可以选择连接池，或者您需要为每个 Actor 实例提供一个专用的 Jedis 连接。请记住，您需要自己处理重新连接（一旦 Redis 连接中断）。

使用 Redisson 和 lettuce，如果您愿意，您可以获得透明的重新连接（这是 lettuce 的默认值，不确定 Redisson）。

通过使用 lettuce 和 Redisson，您可以在所有参与者之间共享一个连接，因为它们是线程安全的。在以下两种情况下，您不能共享一个 lettuce 连接：

阻止操作（因为您会阻止连接的所有其他用户） 事务（MULTI/EXEC，因为您会在事务中混合不同的操作，这当然是您不想这样做的事情）

Jedis 没有异步接口，因此需要您自己完成。这是可行的，我对 MongoDB 做了类似的事情，将 I/O 部分卸载/解耦给其他参与者。您可以在代码中使用该方法，但您不需要提供自己的执行器服务，因为您在可运行侦听器中执行非阻塞操作。

使用 lettuce 4.0，您将获得 Java 8 支持（由于 CompletionStage 接口，这在异步 API 方面要好得多），您甚至可以使用 RxJava（反应式编程）来处理并发。

Lettuce 对您的并发模型没有意见。它允许您根据需要使用它，除了 Java 6/7 的普通 Future/ListenableFuture API 和 Guava 不太好用。

HTH，马克

【参考方案2】：

试试Redisson 框架。它通过与 Project Reactor 和 RxJava3 库的集成提供异步 API 和 Reactive Streams API 支持。

异步 ​​API 使用示例：

RedissonClient client = Redisson.create(config);
RMap<String, String> map = client.getMap("myMap");

// implements CompletionStage interface
RFuture<String> future = map.get("myKey");

future.whenComplete((res, exception) -> 
  // ...
);

带有 Project Reactor 库的 Reactive Streams API 使用示例：

RedissonReactiveClient client = Redisson.createReactive(config);
RMapReactive<String, String> map = client.getMap("myMap");

Mono<String> resp = map.get("myKey");

带有 RxJava2 库的 Reactive Streams API 使用示例：

RedissonRxClient client = Redisson.createRx(config);
RMapRx<String, String> map = client.getMap("myMap");

Flowable<String> resp = map.get("myKey");