{"id":417164,"date":"2024-06-30T01:23:23","date_gmt":"2024-06-30T01:23:23","guid":{"rendered":"http:\/\/savepearlharbor.com\/?p=417164"},"modified":"-0001-11-30T00:00:00","modified_gmt":"2024-06-30T01:23:23","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=417164","title":{"rendered":"<span>Delayed queue in Java and Redis<\/span>"},"content":{"rendered":"<div><!--[--><!--]--><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-1\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<p>Several years ago we had to solve how to enqueue events with an arbitrary delay, e.g. check a status of a payment 3 hours later, or send notification to a client in 45 minutes. At that point of time, we didn&#8217;t find suitable libraries to accomplish this task, which didn&#8217;t require us to spend time on configuration and maintenance. After analysing possible solutions we ended up building our own small library <a href=\"https:\/\/github.com\/fred84\/delayedQueue\" rel=\"nofollow\">delayed queue<\/a> in Java language on top of <code>Redis<\/code> storage engine. In this article I&#8217;ll explain capabilities of this library, alternatives and problems we solved during creation process. <\/p>\n<p><a name=\"habracut\"><\/a>  <\/p>\n<h2 id=\"functionality\">Functionality<\/h2>\n<p>  <\/p>\n<p>So what exactly is <code>delayed queue<\/code> capable of? An event, added to delayed queue, is delivered to a handler after arbitrary delay. If event handling is unsuccessful, it would be delivered again later. However, the number of retries is limited. <code>Redis<\/code> does not provide any resilient guarantees, thus users should be prepared to deal with this. Regardless, in clustered configuration <code>Redis<\/code> shows sufficiently high reliability and we haven&#8217;t faced any issues during 1.5 years of usage.<\/p>\n<p>  <\/p>\n<h3 id=\"api\">API<\/h3>\n<p>  <\/p>\n<h4 id=\"add-an-event-to-a-queue\">Add an event to a queue<\/h4>\n<p>  <\/p>\n<pre><code class=\"java\">eventService.enqueueWithDelayNonBlocking(new DummyEvent(\"1\"), Duration.ofHours(1)).subscribe();<\/code><\/pre>\n<p>  <\/p>\n<p>Please beware, that the method returns <code>Mono<\/code>, so you have to call one of the methods below to launch execution:<\/p>\n<p>  <\/p>\n<ul>\n<li><code>subscribe(...)<\/code><\/li>\n<li><code>block()<\/code><\/li>\n<\/ul>\n<p>  <\/p>\n<p>More details on this could be found in the documentation of <code>Project Reactor<\/code>. Event context could be added in the following manner:<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">eventService.enqueueWithDelayNonBlocking(new DummyEvent(\"1\"), Duration.ofHours(1), Map.of(\"key\", \"value\")).subscribe();<\/code><\/pre>\n<p>  <\/p>\n<h4 id=\"register-an-event-handler\">Register an event handler<\/h4>\n<p>  <\/p>\n<pre><code class=\"java\">eventService.addHandler(DummyEvent.class, e -> Mono.just(true), 1);<\/code><\/pre>\n<p>  <\/p>\n<p>the same action, but with event context:<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">eventService.addHandler(         DummyEvent.class,         e -> Mono             .subscriberContext()             .doOnNext(ctx -> {                 Map&lt;String, String> eventContext = ctx.get(\"eventContext\");                 log.info(\"context key {}\", eventContext.get(\"key\"));             })             .thenReturn(true),         1 );<\/code><\/pre>\n<p>  <\/p>\n<h4 id=\"remove-event-handler\">Remove event handler<\/h4>\n<p>  <\/p>\n<pre><code class=\"java\">eventService.removeHandler(DummyEvent.class);<\/code><\/pre>\n<p>  <\/p>\n<h4 id=\"create-service\">Create service<\/h4>\n<p>  <\/p>\n<p>You can rely on defaults:<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">import static com.github.fred84.queue.DelayedEventService.delayedEventService;  var eventService = delayedEventService().client(redisClient).build();<\/code><\/pre>\n<p>  <\/p>\n<p>or configure everything by yourself:<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">import static com.github.fred84.queue.DelayedEventService.delayedEventService;  var eventService = delayedEventService()         .client(redisClient)         .mapper(objectMapper)         .handlerScheduler(Schedulers.fromExecutorService(executor))         .schedulingInterval(Duration.ofSeconds(1))         .schedulingBatchSize(SCHEDULING_BATCH_SIZE)         .enableScheduling(false)         .pollingTimeout(POLLING_TIMEOUT)         .eventContextHandler(new DefaultEventContextHandler())         .dataSetPrefix(\"\")         .retryAttempts(10)         .metrics(new NoopMetrics())         .refreshSubscriptionsInterval(Duration.ofMinutes(5))         .build();<\/code><\/pre>\n<p>  <\/p>\n<p>Shutting down the service (and all open connections to <code>Redis<\/code>) could be done via <code>eventService.close()<\/code> or framework with support for the lifecycle annotation <code>@javax.annotation.PreDestroy<\/code>.<\/p>\n<p>  <\/p>\n<h4 id=\"metrics\">Metrics<\/h4>\n<p>  <\/p>\n<p>Any system is prone to face faults and we have to monitor it. For <code>delayed queue<\/code> we should look out for:<\/p>\n<p>  <\/p>\n<ul>\n<li><code>Redis<\/code> overall memory usage<\/li>\n<li>size of <code>list<\/code> for every event type (&#171;delayed.queue.ready.for.handling.count&#187; + a tag with event type)<\/li>\n<\/ul>\n<p>  <\/p>\n<h2 id=\"history\">History<\/h2>\n<p>  <\/p>\n<p>Here is a brief overview on how <code>delayed queue<\/code> evolved over time. In 2018, we launched our small project in <a href=\"https:\/\/aws.amazon.com\/\" rel=\"nofollow\">Amazon Web Services<\/a>. Only 2 engineers were in charge of this project, so adding more components, which required configuration and maintenance was discouraged. The main aim was &#171;to use components managed by AWS unless they were too pricey&#187;.<\/p>\n<p>  <\/p>\n<h3 id=\"possible-candidates\">Possible candidates<\/h3>\n<p>  <\/p>\n<ul>\n<li><a href=\"https:\/\/www.rabbitmq.com\" rel=\"nofollow\">RabbitMQ<\/a><\/li>\n<li>JMS<\/li>\n<li><a href=\"https:\/\/aws.amazon.com\/sqs\/\" rel=\"nofollow\">AWS SQS<\/a><\/li>\n<\/ul>\n<p>  <\/p>\n<p>The first two were rejected due to maintenance requirements. The last one (SQS) was not considered as the maximum delay could not be bigger than 15 minutes.<\/p>\n<p>  <\/p>\n<h3 id=\"overlooked-candidates\">Overlooked candidates<\/h3>\n<p>  <\/p>\n<p>Unfortunately, we missed some libraries, which could have solved our needs and were discovered much-much later:<\/p>\n<p>  <\/p>\n<ul>\n<li><a href=\"https:\/\/github.com\/Netflix\/dyno-queues\" rel=\"nofollow\">Netflix dyno-queues<\/a><\/li>\n<li><a href=\"https:\/\/aws.amazon.com\/amazon-mq\/\" rel=\"nofollow\">AWS MQ<\/a><\/li>\n<\/ul>\n<p>  <\/p>\n<p>The first one uses the same technology stack (Java and Redis), the latter is built on top of <a href=\"https:\/\/activemq.apache.org\/\" rel=\"nofollow\">ActiveMQ<\/a>.<\/p>\n<p>  <\/p>\n<h3 id=\"first-naive-implementation\">First naive implementation<\/h3>\n<p>  <\/p>\n<p>Initially, we already had a backup mechanism with polling a relational database once a day. After reading several articles on organising simple delayed queues, we decided to build our solution around <code>Redis<\/code>, not RDBMS. The structure inside <code>Redis<\/code> is as following:<\/p>\n<p>  <\/p>\n<ul>\n<li>an event is added to <a href=\"https:\/\/redis.io\/topics\/data-types\" rel=\"nofollow\">sorted sets<\/a>, where <code>weight<\/code> serves as future execution time<\/li>\n<li>once <code>weight<\/code> becomes lower that <code>now<\/code>, the event is moved from <code>sorted_set<\/code> to <code>list<\/code> (which could be used as a <code>queue<\/code> with <code>push<\/code> and <code>pop<\/code> methods)<\/li>\n<\/ul>\n<p>  <\/p>\n<p>First version of the dispatcher, responsible for moving events from <code>sorted set<\/code> to <code>list<\/code> was:<br \/>  (simplified code is shown here and after):<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">var events = redis.zrangebyscore(\"delayed_events\", Range.create(-1, System.currentTimeMillis()), 100); events.forEach(key -> {   var payload = extractPayload(key);   var listName = extractType(key);   redis.lpush(listName, payload);   redis.zrem(\"delayed_events\", key); });<\/code><\/pre>\n<p>  <\/p>\n<p>Event handlers were built on top of <a href=\"https:\/\/spring.io\/projects\/spring-integration\" rel=\"nofollow\">Spring Integration<\/a>,<br \/>  which executed the following command under the hood:<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">redis.brpop(listName)<\/code><\/pre>\n<p>  <\/p>\n<p>The first problems came soon.<\/p>\n<p>  <\/p>\n<h3 id=\"unreliable-dispatcher\">Unreliable dispatcher<\/h3>\n<p>  <\/p>\n<p>If an error appeared in the process of adding an element to the <code>list<\/code> (e.g. a connection timeout after the element was added), the dispatcher retried to do this operation, which resulted in multiple copies of an event being added to the <code>list<\/code>. Luckily, <code>Redis<\/code> supports transactions, so we wrapped the 2 commands above into transaction.<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">events.forEach(key -> {   ...   redis.multi();   redis.zrem(\"delayed_events\", key);   redis.lpush(listName, payload);   redis.exec(); });<\/code><\/pre>\n<p>  <\/p>\n<h3 id=\"unreliable-handler\">Unreliable handler<\/h3>\n<p>  <\/p>\n<p>On the other side of the <code>list<\/code> lurked another problem. If a handler failed, an event would be lost forever. As a solution, we chose to reschedule an event to a later period of time (unless a maximum number of attempts had been reached) and delete it only after successful processing by the handler.<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">events.forEach(key -> {   ...   redis.multi();   redis.zadd(\"delayed_events\", nextAttempt(key))   redis.zrem(\"delayed_events\", key);   redis.lpush(listName, payload);   redis.exec(); });<\/code><\/pre>\n<p>  <\/p>\n<h3 id=\"non-unique-event\">Non-unique event<\/h3>\n<p>  <\/p>\n<p>As I mentioned before, we already had the fallback mechanism, which polled RDBMS and re-added all &#171;pending&#187; entities to <code>delayed queue<\/code>. At that time, <code>key<\/code> in <code>sorted set<\/code> was structured as <code>metadata;payload<\/code>, with a mutable metadata (e.g. an attempt number, a log context, &#8230;) and an immutable payload. So again, it resulted in multiple copies of an event being added to the <code>list<\/code>. To solve this problem, we moved <code>metadata;payload<\/code> to new structure <code>Redis hset<\/code> and kept only<br \/>  <code>event type + event id<\/code> as a key in <code>sorted set<\/code>. Consequently, event enqueueing transformed from:<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">var envelope = metadata + SEPARATOR + payload; redis.zadd(envelope, scheduledAt);<\/code><\/pre>\n<p>  <\/p>\n<p>into<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">var envelope = metadata + SEPARATOR + payload; var key = eventType + SEPARATOR + eventId;  redis.multi(); redis.zadd(key, scheduledAt); redis.hset(\"metadata\", key, envelope) redis.exec();<\/code><\/pre>\n<p>  <\/p>\n<h3 id=\"sequential-dispatcher-launch\">Sequential dispatcher launch<\/h3>\n<p>  <\/p>\n<p>All our handlers were idempotent, so we didn&#8217;t pay much attention to event duplicates. However, there was still room for improvement. The dispatchers were running on all our application instances and from time to time were launched simultaneously. This again resulted in duplicate events in <code>list<\/code>. The solution was the trivial lock with small TTL:<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">redis.set(lockKey, \"value\", ex(lockTimeout.toMillis() * 1000).nx());<\/code><\/pre>\n<p>  <\/p>\n<h2 id=\"fork-into-independent-project\">Fork into independent project<\/h2>\n<p>  <\/p>\n<p>When the necessity for using <code>delayed queue<\/code> in a project without <code>Spring<\/code> emerged, we moved it to a standalone project. To accomplish it, we were forced to remove the following dependencies:<\/p>\n<p>  <\/p>\n<ul>\n<li><a href=\"https:\/\/spring.io\/projects\/spring-data-redis\" rel=\"nofollow\">Spring Data Redis<\/a><\/li>\n<li><a href=\"https:\/\/spring.io\/projects\/spring-integration\" rel=\"nofollow\">Spring integration<\/a><\/li>\n<\/ul>\n<p>  <\/p>\n<p>The first one was easily replaced with the <code>Lettuce<\/code> Redis driver. The second led to much more changes. At that point of time I had already acquired some experience working with reactive streams in general and with <code>Project Reactor<\/code> in particular. So we chose &#171;hot stream&#187; as a source for our handlers.<br \/>  To achieve the uniform distribution of events among handlers on different application instances we had to implement our own <a href=\"https:\/\/www.reactive-streams.org\/reactive-streams-1.0.0-javadoc\/org\/reactivestreams\/Subscriber.html\" rel=\"nofollow\">Subscriber<\/a>:<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">redis   .reactive()   .brpop(timeout, queue)   .map(e -> deserialize(e))   .subscribe(new InnerSubscriber&lt;>(handler, ... params ..))<\/code><\/pre>\n<p>  <\/p>\n<p>and<\/p>\n<p>  <\/p>\n<pre><code class=\"java\">class InnerSubscriber&lt;T extends Event> extends BaseSubscriber&lt;EventEnvelope&lt;T>> {      @Override     protected void hookOnNext(@NotNull EventEnvelope&lt;T> envelope) {         Mono&lt;Boolean> promise = handler.apply(envelope.getPayload());         promise.subscribe(r -> request(1));     } }<\/code><\/pre>\n<p>  <\/p>\n<p>As a result, we created a library, which delivers events to registered handlers (unlike <code>Netflix dyno queue<\/code>, where you have to poll a storage for events).<\/p>\n<p>  <\/p>\n<h3 id=\"whats-next\">What&#8217;s next?<\/h3>\n<p>  <\/p>\n<ul>\n<li>add Kotlin DSL. Currently, our new projects are created in Kotlin language, so it would be handy to use <code>suspend fun<\/code> instead of<br \/>  direct interaction with <code>Project Reactor<\/code> API<\/li>\n<li>add configurable intervals for retries<\/li>\n<li>replace <code>Redis transactions<\/code> with <code>LUA script<\/code><\/li>\n<\/ul>\n<p>  <\/p>\n<h2 id=\"links\">Links<\/h2>\n<p>  <\/p>\n<ul>\n<li><a href=\"https:\/\/github.com\/fred84\/delayedQueue\" rel=\"nofollow\">delayed queue<\/a><\/li>\n<li><a href=\"https:\/\/redis.io\/topics\/data-types\" rel=\"nofollow\">Redis sorted sets<\/a><\/li>\n<li><a href=\"https:\/\/projectreactor.io\" rel=\"nofollow\">Project reactor<\/a><\/li>\n<li><a href=\"https:\/\/github.com\/FasterXML\/jackson\" rel=\"nofollow\">Jackson<\/a><\/li>\n<li><a href=\"https:\/\/micrometer.io\" rel=\"nofollow\">Micrometer<\/a><\/li>\n<li><a href=\"https:\/\/lettuce.io\" rel=\"nofollow\">Lettuce<\/a><\/li>\n<li><a href=\"https:\/\/github.com\/Netflix\/dyno-queues\" rel=\"nofollow\">Netflix dyno-queues<\/a><\/li>\n<\/ul>\n<\/div>\n<\/div>\n<\/div>\n<p><!----><!----><\/div>\n<p><!----><!----><br \/> \u0441\u0441\u044b\u043b\u043a\u0430 \u043d\u0430 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438 <a href=\"https:\/\/habr.com\/ru\/articles\/516708\/\"> https:\/\/habr.com\/ru\/articles\/516708\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"<div><!--[--><!--]--><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-1\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<p>Several years ago we had to solve how to enqueue events with an arbitrary delay, e.g. check a status of a payment 3 hours later, or send notification to a client in 45 minutes. At that point of time, we didn&#8217;t find suitable libraries to accomplish this task, which didn&#8217;t require us to spend time on configuration and maintenance. After analysing possible solutions we ended up building our own small library <a href=\"https:\/\/github.com\/fred84\/delayedQueue\" rel=\"nofollow\">delayed queue<\/a> in Java language on top of <code>Redis<\/code> storage engine. In this article I&#8217;ll explain capabilities of this library, alternatives and problems we solved during creation process. <\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[],"class_list":["post-417164","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/417164","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=417164"}],"version-history":[{"count":0,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/417164\/revisions"}],"wp:attachment":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=417164"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=417164"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=417164"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}