I am using code [1] to get rankings accross several keys [2] with values [3] in an app [4]. In short, those are rankings for several quizzes for specific clients. Each ranking can contain millions of entries. Thus, I am wondering whether the “zunionstore” can be avoid somehow. E.g. I need to calculate rankings accross all quizzes, so I am using zunionstore to sum all rankings into one combined ranking. Then it is easy to apply rank function to get the user actual rank.
Unfortunately this takes about 0.5 seconds for every million users. So it is the bottleneck here.
Any suggestions for other way to apply such ranking in redis?
[1]
import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.redis.client.RedisAPI;
import io.vertx.redis.client.Response;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import org.jboss.logging.Logger;
@ApplicationScoped
public class RankingService {
private static final Logger LOGGER = Logger.getLogger(RankingService.class);
@Inject RedisAPI redisAPI;
public Future<Map<String, Map<String, Object>>> getCombinedRanking(
String clientId, List<String> quizIds, int limit, int offset) {
Promise<Map<String, Map<String, Object>>> promise = Promise.promise();
String tempCombinedKey =
clientId + ":temp:combined:ranking:" + UUID.randomUUID();
List<String> keys = new ArrayList<>();
for (String quizId : quizIds) {
keys.add(clientId + ":quiz:" + quizId + ":ranking");
}
List<String> zunionstoreArgs = new ArrayList<>();
zunionstoreArgs.add(tempCombinedKey);
zunionstoreArgs.add(String.valueOf(keys.size()));
zunionstoreArgs.addAll(keys);
zunionstoreArgs.add("AGGREGATE");
zunionstoreArgs.add("SUM");
AtomicLong startTime = new AtomicLong(System.nanoTime());
redisAPI
.zunionstore(zunionstoreArgs)
.onComplete(
ar -> {
if (ar.succeeded()) {
long endTime = System.nanoTime();
long duration = TimeUnit.NANOSECONDS.toMillis(endTime - startTime.get());
LOGGER.infof("zunionstore command took %d ms", duration);
startTime.set(System.nanoTime());
redisAPI
.zrevrange(
Arrays.asList(
tempCombinedKey,
String.valueOf(offset),
String.valueOf(offset + limit - 1),
"WITHSCORES"))
.onComplete(
ar2 -> {
if (ar2.succeeded()) {
long endTime2 = System.nanoTime();
long duration2 =
TimeUnit.NANOSECONDS.toMillis(endTime2 - startTime.get());
LOGGER.infof("zrevrange command took %d ms", duration2);
Response response = ar2.result();
LOGGER.infof("zrevrange response: %s", response.toString());
Map<String, Map<String, Object>> combinedScores = new LinkedHashMap<>();
List<Future> rankFutures = new ArrayList<>();
for (int i = 0; i < response.size(); i++) {
Response pair = response.get(i);
if (pair.size() < 2) {
LOGGER.warnf("Invalid pair in response: %s", pair);
continue;
}
String userId = pair.get(0).toString();
String scoreString = pair.get(1).toString();
LOGGER.infof("User ID: %s, Score String: %s", userId, scoreString);
try {
Double score = Double.parseDouble(scoreString);
Map<String, Object> scoreDetails = new HashMap<>();
scoreDetails.put("score", score);
combinedScores.put(userId, scoreDetails);
Future<Response> rankFuture =
redisAPI.zcount(tempCombinedKey, "(" + score, "+inf");
rankFutures.add(
rankFuture
.onSuccess(
rankResponse -> {
Long rank =
rankResponse.toLong()
+ 1;
scoreDetails.put("rank", rank);
})
.onFailure(
err -> {
LOGGER.errorf(
"Error fetching rank for user %s: %s",
userId, err);
}));
} catch (NumberFormatException e) {
LOGGER.errorf(
"Error parsing score for user %s: %s", userId, scoreString, e);
}
}
CompositeFuture.all(rankFutures)
.onComplete(
ar3 -> {
if (ar3.succeeded()) {
long processingEndTime = System.nanoTime();
long processingDuration =
TimeUnit.NANOSECONDS.toMillis(
processingEndTime - startTime.get());
LOGGER.infof(
"Processing combined ranking took %d ms",
processingDuration);
promise.complete(combinedScores);
} else {
promise.fail(ar3.cause());
}
long cleanupStartTime = System.nanoTime();
redisAPI
.del(Arrays.asList(tempCombinedKey))
.onComplete(
delAr -> {
if (delAr.succeeded()) {
long cleanupEndTime = System.nanoTime();
long cleanupDuration =
TimeUnit.NANOSECONDS.toMillis(
cleanupEndTime - cleanupStartTime);
LOGGER.infof(
"Cleanup (del) command took %d ms",
cleanupDuration);
} else {
LOGGER.errorf(
"Failed to delete temporary key %s: %s",
tempCombinedKey, delAr.cause());
}
});
});
} else {
promise.fail(ar2.cause());
}
});
} else {
promise.fail(ar.cause());
}
});
return promise.future();
}
}
[2]
client1:quiz:quiz_1:ranking
client1:quiz:quiz_2:ranking
…
[3]
[ { "score" : 0, "content" : "user_0" }, { "score" : 10, "content" : "user_1" }, { "score" : 20, "content" : "user_2" }, { "score" : 20, "content" : "user_3" } ... ]
[4]
Quarkus app with redis (vertx-redis-client-4.5.4).