This article only analyzes the connection timeout exception and read timeout exception in the Redis pooled connection of the Letture type.
1. RedisConnectionException
The default is 10 seconds.
It can be configured by:
public class MyLettuceClientConfigurationBuilderCustomizer implements LettuceClientConfigurationBuilderCustomizer {<!-- -->
@Override
public void customize(LettuceClientConfiguration. LettuceClientConfigurationBuilder clientConfigurationBuilder) {<!-- -->
// spring.redis.timeout equivalent
clientConfigurationBuilder.commandTimeout(Duration.ofSeconds(100));
// Control the connection timeout, the default is 10 seconds
ClientOptions.Builder builder = ClientOptions.builder().socketOptions(SocketOptions.builder().connectTimeout(Duration.ofMillis(10)).build());
clientConfigurationBuilder.clientOptions(builder.build());
}
}
2. RedisCommandTimeoutException
Conclusion: Throwing a RedisCommandTimeoutException does not mean that the data writing must fail. It’s just that Redis’s internal timeout logic and write logic are processed asynchronously, so there may still be exceptions thrown after successful writing. But as long as the exception occurs, it must be handled.
io.lettuce.core.RedisCommandTimeoutException: Command timed out after 2 second(s)
2.1.CommandHandler limitations
CommandHandler is the handler in Netty. It’s just that it acts as a push and pop handler at the same time. In layman’s terms, it is the only way for Redis Write & Flush data to channel NioSocketChannel.
There are two core functions of AsyncCommand. One is to wrap user data, and the other is to implement the CompletableFuture interface to facilitate asynchronous related operations.
public class CommandHandler extends ChannelDuplexHandler implements HasQueuedCommands {<!-- -->
private void writeSingleCommand(ChannelHandlerContext ctx, RedisCommand<?, ?, ?> command, ChannelPromise promise)
{<!-- -->
if (!isWriteable(command)) {<!-- -->// Here is the core of judging whether there is a timeout, borrowing CompletableFuture asynchronous operation function
promise.trySuccess();// At this time, it means that the timeout period set by the user has exceeded during the data writing process
return;
}
...
ctx.write(command, promise);//The data is written normally, and the final destination is the server
}
\t
private static boolean isWriteable(RedisCommand<?, ?, ?> command) {<!-- -->
return !command.isDone();
}
}
public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {<!-- -->
public boolean isDone() {<!-- -->
return result != null;
}
}
According to the above, as long as the variable result inside CompletableFuture is not empty, it will be handled as TimeoutException. So when is the result value assigned?
As shown below, CommandHandler will set the value of result in the following way after processing the data.
public class AsyncCommand<K, V, T> extends CompletableFuture<T> implements RedisCommand<K, V, T>, RedisFuture<T>,
CompleteableCommand<T>, DecoratedCommand<K, V, T> {<!-- -->
protected void completeResult() {<!-- -->
if (command. getOutput() == null) {<!-- -->
complete(null);
} else if (command. getOutput(). hasError()) {<!-- -->
doCompleteExceptionally(ExceptionFactory. createExecutionException(command. getOutput(). getError()));
} else {<!-- -->
// If it is successfully sent, the Output is "ok" of string type
complete(command. getOutput(). get());
}
}
}
3. Timeout processing timing
3.1. Blocking wait timeout
class FutureSyncInvocationHandler extends AbstractInvocationHandler {<!-- -->
protected Object handleInvocation(Object proxy, Method method, Object[] args) throws Throwable {<!-- -->
try {<!-- -->
Method targetMethod = this. translator. get(method);
// CommandExpiryWriter#writer
Object result = targetMethod.invoke(asyncApi, args);
if (result instanceof RedisFuture<?>) {<!-- -->
...
long timeout = getTimeoutNs(command);//User-defined timeout
//command:AsyncCommand utilizes the asynchronous feature of CompletableFuture
return LettuceFutures. awaitOrCancel(command, timeout, TimeUnit. NANOSECONDS);
}
return result;
} catch (InvocationTargetException e) {<!-- -->
throw e. getTargetException();
}
}
}
public class LettuceFutures {<!-- -->
public static <T> T awaitOrCancel(RedisFuture<T> cmd, long timeout, TimeUnit unit) {<!-- -->
try {<!-- -->
//Blocking In fact, the CompletableFuture#get blocking method called by memory. If TimeoutExeception is returned, it means that AsyncCommand has not been executed, and timeout processing
if (!cmd.await(timeout, unit)) {<!-- -->
cmd. cancel(true);
throw ExceptionFactory.createTimeoutException(Duration.ofNanos(unit.toNanos(timeout)));
}
return cmd.get();//This method still calls the CompletableFuture#get blocking method, but this method is judged according to the internal attribute result
} catch (RuntimeException e) {<!-- -->
throw e;
} catch (ExecutionException e) {<!-- -->
if (e.getCause() instanceof RedisCommandExecutionException) {<!-- -->
throw ExceptionFactory.createExecutionException(e.getCause().getMessage(), e.getCause());
}
if (e.getCause() instanceof RedisCommandTimeoutException) {<!-- -->
throw new RedisCommandTimeoutException(e. getCause());
}
throw new RedisException(e. getCause());
} catch (InterruptedException e) {<!-- -->
Thread. currentThread(). interrupt();
throw new RedisCommandInterruptedException(e);
} catch (Exception e) {<!-- -->
throw ExceptionFactory. createExecutionException(null, e);
}
}
}
As shown in the following pseudo code: if the result is not null, a related exception is thrown according to the result type. Otherwise, block and wait within the specified time through timedGet, and throw TimeoutException when timed out.
public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {<!-- -->
public T get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {<!-- -->
Object r;
long nanos = unit.toNanos(timeout);
return reportGet((r = result) == null ? timedGet(nanos) : r);
}
}
3.2. Timing task timeout processing
public class CommandExpiryWriter implements RedisChannelWriter {<!-- -->
public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {<!-- -->
potentiallyExpire(command, getExecutorService());//Start scheduled tasks
return writer.write(command);//DefaultEndPoint#write
}
\t
private void potentiallyExpire(RedisCommand<?, ?, ?> command, ScheduledExecutorService executors) {<!-- -->
long timeout = applyConnectionTimeout ? this.timeout : source.getTimeout(command);//User-defined timeout
if (timeout <= 0) {<!-- -->
return;
}
// Start timing tasks after the timeout time. At this time, if the result value exists, it means timeout
ScheduledFuture<?> schedule = executors. schedule(() -> {<!-- -->
if (!command.isDone()) {<!-- -->
command.completeExceptionally(ExceptionFactory.createTimeoutException(Duration.ofNanos(timeUnit
.toNanos(timeout))));
}
}, timeout, timeUnit);
if (command instanceof CompleteableCommand) {<!-- -->
((CompleteableCommand) command). onComplete((o, o2) -> {<!-- -->
if (!schedule.isDone()) {<!-- -->
schedule. cancel(false);
}
});
}
}
}
Important: The only condition for judging the success of a scheduled task is whether there is a value in the result. In case 1, if result = ok at this time, it means that the data has been successfully placed on the server side, but RedisCommandTimeoutException will also be thrown. In case 2, if an exception occurs before sending, the result value is an abnormal value of type Throwable.