2
\$\begingroup\$

To prevent an XY problem, I'll first explain the problem I am trying to solve:

Problem I am trying to solve

I want to send the contents of a MongoDB collection with a size of around 3 million records to a queuing microservice, so that other microservices might subscribe to my topic and read the messages at their convenience. To keep memory usage into bounds, I stream the collection from the database to the code that sends the messages to the queuing service, and I want to limit the amount of messages that are being created and sent at any moment to a maximum of 4000. I currently store the messages that need to be sent in a ArrayBlockingQueue, and an ExecutorService picks them off there. Multiple threads may be using the same code to send other messages of their own, so I need to check if there is any room left in this list, before adding a message to it.

Minimal example

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.BiConsumer;

import static java.util.concurrent.TimeUnit.HOURS;
import static java.util.concurrent.TimeUnit.SECONDS;

public class Main {
    private static final Logger LOG = LogManager.getLogger();

    private ExecutorService executorService;
    private ArrayBlockingQueue<Runnable> messageQueue;
    private int threadCount = 2;
    private int maxMessageQueueSize = 4000;

    public static void main(String[] args) {
        new Main().go();
    }

    private void go() {
        initExecutorService();
        sendMessages();
        shutdownExecutorService();
    }

    private void initExecutorService() {
        messageQueue = new ArrayBlockingQueue<>(maxMessageQueueSize);
        executorService = new ThreadPoolExecutor(
                threadCount, threadCount, 1L, HOURS, messageQueue);
    }

    private void sendMessages() {
        int fullQueueDetectedCount = 0;

        // simulate the sending of messages to a queue
        for (int i = 0; i < 10_000; i++) {
            if (maxMessageQueueSize - messageQueue.size() > 0) {
                if (fullQueueDetectedCount > 0) fullQueueDetectedCount = 0;
                submitMessage(i);
            } else {
                fullQueueDetectedCount++;
                if (fullQueueDetectedCount > 6) {
                    LOG.error("Queue is full, even after waiting {} times " +
                            "for an empty spot", fullQueueDetectedCount);
                    return;
                }

                if (!sleepBasedOn(fullQueueDetectedCount)) return;

                LOG.info("Slept while waiting for a place on the queue");
            }
        }

        LOG.info("Submitted all messages");
    }

    private boolean sleepBasedOn(int power) {
        try {
            Thread.sleep((long) (1000 * Math.pow(2, power)));
        } catch (InterruptedException e) {
            LOG.error(e);
            return false;
        }
        return true;
    }

    private void shutdownExecutorService() {
        executorService.shutdown();
        try {
            executorService.awaitTermination(2, SECONDS);
        } catch (InterruptedException e) {
            LOG.error(e);
        }
    }

    private void submitMessage(int num) {

        Runnable sendTask = () -> {
            try {
                LOG.debug("Sending message with number {}", num);
                // simulate the amount of time needed to assemble
                // and send the message to the queueing service
                Thread.sleep(25);
            } catch (InterruptedException e) {
                LOG.error(e);
            }
        };

        BiConsumer<Void, Throwable> logException = (retValue, exception) -> {
            if (exception != null) {
                LOG.error("Failed to submit message");
            }
        };

        try {
            CompletableFuture
                    .runAsync(sendTask, executorService)
                    .whenComplete(logException);
        } catch (Exception e) {
            LOG.error(e);
        }
    }
}

Question

Is there a better, maybe more efficient or "clean" way to do what I want? Did I miss any of the built-in Executors that do this already or is there a library? I need all messages to be sent, so I cannot allow messages to be dropped when the queue is full. Also, although this code example doesn't show it, I'm running a Spring web application.

Naturally, general remarks about this code are appreciated as well (code style, etc.).

\$\endgroup\$
  • \$\begingroup\$ Is the queue just that? An ArrayBlockingQueue in memory? Or does this represent any not-included-here concept? And apart from that: I cannot spot the actual putting things to the queue in your code... \$\endgroup\$ – mtj Apr 24 '18 at 11:38
  • \$\begingroup\$ I send messages to a microservice that queues these messages so that other microservices can pick them up. So in the sendTask, the real code would be something like queuingService.submit(msg) instead of Thread.sleep(...). The ArrayBlockingQueue is just used for storing Runnables that actually submit stuff to this queuing microservice. Does that make sense? \$\endgroup\$ – Stefan van den Akker Apr 24 '18 at 16:54
  • 1
    \$\begingroup\$ Yes, somewhat. If it was just a simple ArrayBlockingQueue, I might have come up with some ideas, but in this case, I am afraid that your question does not match the "real code for review" requirements on codereview. The more senior members of this site would have some ready-made comments with links to "what is on topic here", "how should questions look on codereview" and so on. So, please just take my opinion that this is not suitable for code review and a big "sorry, interesting topic" in prose. \$\endgroup\$ – mtj Apr 24 '18 at 20:09

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.