Quarkus 中的 Redis 和并发错误：没有 MULTI 的 ERR EXEC答案

【问题标题】：Redis and concurrency errors in Quarkus: ERR EXEC without MULTIQuarkus 中的 Redis 和并发错误：没有 MULTI 的 ERR EXEC
【发布时间】：2021-12-11 19:34:39
【问题描述】：

我有一个将数据保存在 redis 中的应用程序，这些数据通过队列中的事件到达，从而生成高并发环境。为了保持数据一致性，我使用了 Redis MULTI / EXEC 语句，如果发生故障，数据不会得到确认。在打开 MULTI 之前，我需要向 Redis 查询数据。

我举一个例子。

我用来模拟事件的测试端点：

@GET
@Path("/foo")
public Response foo() {
    String queue = "foo";

    for (int i = 0; i < 10000; i++) {
        activeMqClient.addToQueue(queue,  new Foo(i, "Message " + 1));
    }

    return OkResponse.toResponse("Foo send OK");
}

事件燃烧（在线程中运行多个实例）：

public class FooRedisCmd {

    public void process(Foo foo) {
        // Example
        RedisClient redisClient = null;
        Response multiResponse = null;
        try {
            redisClient = RedisClient.createClient();
            //get info
            redisClient.hget("foos", "foo_field1:" + foo.getId());
            redisClient.hget("foos", "foo_field2:" + foo.getId());
            redisClient.hget("foos", "foo_field3:" + foo.getId());
            redisClient.hget("foos", "foo_field4:" + foo.getId());
            redisClient.hget("foos", "foo_field5:" + foo.getId());
            redisClient.hget("foos", "foo_field6:" + foo.getId());
            redisClient.hget("foos", "foo_field7:" + foo.getId());

            //write info
            multiResponse = redisClient.multi();
            log.info("Get multi: {} for foo: {}", multiResponse, foo);
            redisClient.hincrby("foos", "foo_field1:" + foo.getId(), Integer.toString(1));
            redisClient.hincrby("foos", "foo_field2:" + foo.getId(), Integer.toString(1));
            redisClient.hincrby("foos", "foo_field3:" + foo.getId(), Integer.toString(1));
            redisClient.hincrby("foos", "foo_field4:" + foo.getId(), Integer.toString(1));
            redisClient.hincrby("foos", "foo_field5:" + foo.getId(), Integer.toString(1));
            redisClient.hincrby("foos", "foo_field6:" + foo.getId(), Integer.toString(1));
            redisClient.hincrby("foos", "foo_field7:" + foo.getId(), Integer.toString(1));
            redisClient.hincrby("foos", "foo_field8:" + foo.getId(), Integer.toString(1));
            redisClient.hincrby("foos", "foo_field9:" + foo.getId(), Integer.toString(1));
            redisClient.hincrby("foos", "foo_field10:" + foo.getId(), Integer.toString(1));
            redisClient.hincrby("foos", "foo_field11:" + foo.getId(), Integer.toString(1));
            redisClient.hincrby("foos", "foo_field12:" + foo.getId(), Integer.toString(1));
            Response execResponse = redisClient.exec();
            log.info("Multi result: {} for foo: {}", execResponse, foo);
        } catch (Exception e) {
            if (multiResponse != null) {
                try {
                    redisClient.discard();
                } catch (Exception e2) {
                    log.error("Error on discard for foo: {}", foo, e2);
                }
            }
            log.error("Error on exec for foo: {}", foo, e);
        } finally {
            if (redisClient != null) {
                redisClient.close();
            }
        }
    }
}

错误：

INFO  [com.foo.FooRedisCmd] (pool-17-thread-3) Get multi: OK for foo: Foo(id=9965, message=Message 1)
INFO  [com.foo.FooRedisCmd] (pool-17-thread-3) Multi result: [3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3] for foo: Foo(id=9965, message=Message 1)
INFO  [com.foo.FooRedisCmd] (pool-17-thread-1) Get multi: OK for foo: Foo(id=9966, message=Message 1)
ERROR [com.foo.FooRedisCmd] (pool-17-thread-1) Error on discard for foo: Foo(id=9966, message=Message 1): java.util.concurrent.CompletionException: ERR DISCARD without MULTI
        at io.smallrye.mutiny.operators.uni.UniBlockingAwait.await(UniBlockingAwait.java:73)
        at io.smallrye.mutiny.groups.UniAwait.atMost(UniAwait.java:61)
        at io.quarkus.redis.client.runtime.RedisClientImpl.await(RedisClientImpl.java:1026)
        at io.quarkus.redis.client.runtime.RedisClientImpl.discard(RedisClientImpl.java:142)
        at com.foo.FooRedisCmd.process(FooRedisCmd.java:48)
        at com.foo.FooRedisCmd_Subclass.process$$superforward1(FooRedisCmd_Subclass.zig:94)
        at com.foo.FooRedisCmd_Subclass$$function$$1.apply(FooRedisCmd_Subclass$$function$$1.zig:33)
        at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:54)
        at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.proceed(InvocationInterceptor.java:62)
        at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.monitor(InvocationInterceptor.java:51)
        at io.quarkus.arc.runtime.devconsole.InvocationInterceptor_Bean.intercept(InvocationInterceptor_Bean.zig:521)
        at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:41)
        at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:41)
        at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)
        at com.foo.FooRedisCmd_Subclass.process(FooRedisCmd_Subclass.zig:158)
        at com.foo.FooRedisCmd_ClientProxy.process(FooRedisCmd_ClientProxy.zig:128)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: ERR DISCARD without MULTI

INFO  [com.foo.FooRedisCmd] (pool-17-thread-2) Get multi: OK for foo: Foo(id=9967, message=Message 1)
ERROR [com.foo.FooRedisCmd] (pool-17-thread-1) Error on exec for foo: Foo(id=9966, message=Message 1): java.util.concurrent.CompletionException: ERR EXEC without MULTI
        at io.smallrye.mutiny.operators.uni.UniBlockingAwait.await(UniBlockingAwait.java:73)
        at io.smallrye.mutiny.groups.UniAwait.atMost(UniAwait.java:61)
        at io.quarkus.redis.client.runtime.RedisClientImpl.await(RedisClientImpl.java:1026)
        at io.quarkus.redis.client.runtime.RedisClientImpl.exec(RedisClientImpl.java:167)
        at com.foo.FooRedisCmd.process(FooRedisCmd.java:43)
        at com.foo.FooRedisCmd_Subclass.process$$superforward1(FooRedisCmd_Subclass.zig:94)
        at com.foo.FooRedisCmd_Subclass$$function$$1.apply(FooRedisCmd_Subclass$$function$$1.zig:33)
        at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:54)
        at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.proceed(InvocationInterceptor.java:62)
        at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.monitor(InvocationInterceptor.java:51)
        at io.quarkus.arc.runtime.devconsole.InvocationInterceptor_Bean.intercept(InvocationInterceptor_Bean.zig:521)
        at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:41)
        at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:41)
        at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)
        at com.foo.FooRedisCmd_Subclass.process(FooRedisCmd_Subclass.zig:158)
        at com.foo.FooRedisCmd_ClientProxy.process(FooRedisCmd_ClientProxy.zig:128)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: ERR EXEC without MULTI

INFO  [com.foo.FooRedisCmd] (pool-17-thread-2) Multi result: [3, 3, 3, 3, 3, 3, 3, 3, 3, 3] for foo: Foo(id=9967, message=Message 1)

当第一个失败时同时执行 exec 和 discard 时会产生错误。为什么会这样？我正在使用 quarkus 2 及其自己的客户端作为 redis quarkus-redis-client。

【问题讨论】：

标签： java redis concurrency quarkus

【解决方案1】：

由于您在线程中运行多个实例，multi 和 exec 命令交替执行。

像这样。

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set dst jjj
QUEUED
127.0.0.1:6379> multi
(error) ERR MULTI calls can not be nested
127.0.0.1:6379> exec
1) OK
127.0.0.1:6379> exec
(error) ERR EXEC without MULTI
127.0.0.1:6379> discard
(error) ERR DISCARD without MULTI

解决方案是pipeline，将你的multi、hincrby、exec命令打包成一个命令。 Redis 将按正确的顺序执行它们。您可以参考https://redis.io/topics/pipelining了解更多信息。

【讨论】：

【解决方案2】：

你可以使用管道，如果你使用 Redis 的 jedis 客户端，你可以看看这个

https://www.tabnine.com/code/java/methods/redis.clients.jedis.Jedis/pipelined

【讨论】：