22

Question

How do you create a proper background loader in Java 8? The conditions:

  • data should be loaded in background
  • after the loading the data should be displayed
  • while data are loaded no further requests should be accepted
  • if there were requests while the data were loaded another loading should be scheduled after a certain timeout (e. g. 5 seconds)

The purpose is to e. g. have reload requests accepted, but not the database flooded with the requests.

MCVE

Here's a MCVE. It consists of a background task which simulates the loading by simply invoking Thread.sleep for 2 seconds. The task is scheduled every second which naturally leads to an overlap of the background loading tasks, which should be avoided.

public class LoadInBackgroundExample {

  /**
   * A simple background task which should perform the data loading operation. In this minimal example it simply invokes Thread.sleep
   */
  public static class BackgroundTask implements Runnable {

    private int id;

    public BackgroundTask(int id) {
      this.id = id;
    }

    /**
     * Sleep for a given amount of time to simulate loading.
     */
    @Override
    public void run() {

      try {

        System.out.println("Start #" + id + ": " + Thread.currentThread());

        long sleepTime = 2000; 
        Thread.sleep( sleepTime);

      } catch (InterruptedException e) {
        e.printStackTrace();
      } finally {
        System.out.println("Finish #" + id + ": " + Thread.currentThread());
      }

    }
  }

  /**
   * CompletableFuture which simulates loading and showing data.
   * @param taskId Identifier of the current task
   */
  public static void loadInBackground( int taskId) {

    // create the loading task
    BackgroundTask backgroundTask = new BackgroundTask( taskId);

    // "load" the data asynchronously
    CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(new Supplier<String>() {

      @Override
      public String get() {

        CompletableFuture<Void> future = CompletableFuture.runAsync(backgroundTask);

        try {

          future.get();

        } catch (InterruptedException | ExecutionException e) {
          e.printStackTrace();
        }

        return "task " + backgroundTask.id;
      }
    });

    // display the data after they are loaded
    CompletableFuture<Void> future = completableFuture.thenAccept(x -> {

      System.out.println( "Background task finished:" + x);

    });

  }


  public static void main(String[] args) {

    // runnable which invokes the background loader every second
    Runnable trigger = new Runnable() {

      int taskId = 0;

      public void run() { 

        loadInBackground( taskId++);

      }
    };

    // create scheduler
    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    ScheduledFuture<?> beeperHandle = scheduler.scheduleAtFixedRate(trigger, 0, 1, TimeUnit.SECONDS);

    // cancel the scheudler and the application after 10 seconds
    scheduler.schedule(() -> beeperHandle.cancel(true), 10, TimeUnit.SECONDS);

    try {
      beeperHandle.get();
    } catch (Throwable th) {
    }

    System.out.println( "Cancelled");
    System.exit(0);
  }

}

The output is this:

Start #0: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Start #1: Thread[ForkJoinPool.commonPool-worker-4,5,main]
Start #2: Thread[ForkJoinPool.commonPool-worker-6,5,main]
Finish #0: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Background task finished:task 0
Finish #1: Thread[ForkJoinPool.commonPool-worker-4,5,main]
Background task finished:task 1
Start #3: Thread[ForkJoinPool.commonPool-worker-4,5,main]
Finish #2: Thread[ForkJoinPool.commonPool-worker-6,5,main]
Background task finished:task 2
Start #4: Thread[ForkJoinPool.commonPool-worker-6,5,main]
Start #5: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Finish #3: Thread[ForkJoinPool.commonPool-worker-4,5,main]
Background task finished:task 3
Start #6: Thread[ForkJoinPool.commonPool-worker-4,5,main]
Finish #4: Thread[ForkJoinPool.commonPool-worker-6,5,main]
Background task finished:task 4
Finish #5: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Background task finished:task 5
Start #7: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Finish #6: Thread[ForkJoinPool.commonPool-worker-4,5,main]
Start #8: Thread[ForkJoinPool.commonPool-worker-6,5,main]
Background task finished:task 6
Start #9: Thread[ForkJoinPool.commonPool-worker-4,5,main]
Finish #7: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Background task finished:task 7
Start #10: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Finish #8: Thread[ForkJoinPool.commonPool-worker-6,5,main]
Background task finished:task 8
Cancelled

The goal is to have e. g. #1 and #2 skipped because #0 is still running.

Problem

Where do you properly set the blocking mechanism? Should synchronization be used? Or some AtomicBoolean? And if so, should it be inside the get() method or elsewhere?

8
  • 5
    Have you considered BlockingQueue ? Commented Nov 11, 2019 at 6:12
  • Not yet, but sounds like a valid approach. I'll check it out. Thank you! Commented Nov 11, 2019 at 6:18
  • Have you considered an ExecutorService with a thread pool size of 1? Commented Nov 11, 2019 at 6:18
  • 2
    @Roland What is the bounty about? The question has no updates after the latest comment by you. Did you explore BlockingQueue? Commented Nov 16, 2019 at 18:11
  • 'Blocking background' appears to embody a contradiction in terms. Commented Nov 16, 2019 at 23:04

7 Answers 7

6

You already have a threadpool to execute the task. It's not necessarily and make thing complicated to run the task in another async executor (ForkJoinPool when you use CompletableFuture)

Make it simple:

public static void loadInBackground(int taskId) {
    // create the loading task
    BackgroundTask backgroundTask = new BackgroundTask(taskId);
    // No need to run in async, as it already in executor
    backgroundTask.run();
}

The ScheduledExecutorService will ensure only one task is run at a time when you invoked it with scheduleAtFixedRate

Creates and executes a periodic action that becomes enabled first after the given initial delay, and subsequently with the given period; that is executions will commence after initialDelay then initialDelay+period, then initialDelay + 2 * period, and so on. If any execution of the task encounters an exception, subsequent executions are suppressed. Otherwise, the task will only terminate via cancellation or termination of the executor. If any execution of this task takes longer than its period, then subsequent executions may start late, but will not concurrently execute.

Sign up to request clarification or add additional context in comments.

Comments

5
+200

Taking the following as the requirements:

  • data should be loaded in background
  • after the loading the data should be displayed
  • while data are loaded no further requests should be accepted
  • if there were requests while the data were loaded another loading should be scheduled after a certain timeout (e. g. 5 seconds)

The solution ca be build based on the Executors.newSingleThreadExecutor(), CompletableFuture and LinkedBlockingQueue:

public class SingleThreadedLoader {

  private static class BackgroundTask extends CompletableFuture<String> {

    private final String query;

    private BackgroundTask(String query) {
      this.query = query;
    }

    public String getQuery() {
      return query;
    }
  }

  private final BlockingQueue<BackgroundTask> tasks = new LinkedBlockingQueue<>();
  // while data are loaded no further requests should be accepted
  private final Executor executor = Executors.newSingleThreadExecutor();

  private final int delaySeconds;

  private AtomicReference<Instant> lastExecution = new AtomicReference<>(Instant.EPOCH);

  public SingleThreadedLoader(int delaySeconds) {
    this.delaySeconds = delaySeconds;
    setupLoading();
  }

  public BackgroundTask loadInBackground(String query) {
    log("Enqueued query " + query);
    BackgroundTask task = new BackgroundTask(query);
    tasks.add(task);
    return task;
  }

  private void setupLoading() {
    // data should be loaded in background
    executor.execute(() -> {
      while (true) {
        try {
          // if there were requests while the data were loaded
          // another loading should be scheduled after a certain timeout (e. g. 5 seconds)
          Instant prev = lastExecution.get();
          long delay = Duration.between(prev, Instant.now()).toSeconds();
          if (delay < delaySeconds) {
            log("Waiting for 5 seconds before next data loading");
            TimeUnit.SECONDS.sleep(delaySeconds - delay);
          }
          BackgroundTask task = tasks.take();
          try {
            String query = task.getQuery();
            String data = loadData(query);
            task.complete(data);
          } catch (Exception e) {
            task.completeExceptionally(e);
          }
          lastExecution.set(Instant.now());
        } catch (InterruptedException e) {
          log(e.getMessage());
          return;
        }
      }
    });
  }

  private String loadData(String query) {
    try {
      log("Loading data for " + query);
      TimeUnit.SECONDS.sleep(2);
      log("Loaded data for " + query);
      return "Result " + query;
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  }

  private static void log(String str) {
    String time = LocalTime.now().truncatedTo(ChronoUnit.SECONDS).format(DateTimeFormatter.ISO_TIME);
    String thread = Thread.currentThread().getName();
    System.out.println(time + ' ' + thread + ": " + str);
  }

  public static void main(String[] args) throws Exception {
    SingleThreadedLoader loader = new SingleThreadedLoader(5);
    // after the loading the data should be displayed
    loader.loadInBackground("1").thenAccept(SingleThreadedLoader::log);
    loader.loadInBackground("2").thenAccept(SingleThreadedLoader::log);
    loader.loadInBackground("3").thenAccept(SingleThreadedLoader::log);

    log("Do another work in the main thread");

    TimeUnit.SECONDS.sleep(30);
  }
}

After the execution the stdout will have the following output:

10:29:26 main: Enqueued query 1
10:29:26 pool-1-thread-1: Loading data for 1
10:29:26 main: Enqueued query 2
10:29:26 main: Enqueued query 3
10:29:26 main: Do another work in the main thread
10:29:28 pool-1-thread-1: Loaded data for 1
10:29:28 pool-1-thread-1: Result 1
10:29:28 pool-1-thread-1: Waiting for 5 seconds before next data loading
10:29:33 pool-1-thread-1: Loading data for 2
10:29:36 pool-1-thread-1: Loaded data for 2
10:29:36 pool-1-thread-1: Result 2
10:29:36 pool-1-thread-1: Waiting for 5 seconds before next data loading
10:29:41 pool-1-thread-1: Loading data for 3
10:29:43 pool-1-thread-1: Loaded data for 3
10:29:43 pool-1-thread-1: Result 3

Comments

3

I have added an AtomicInteger which will act as a counter for running tasks with simple lock() and unlock() methods with this minor change into your original code I got output:

Start #0: Thread[ForkJoinPool.commonPool-worker-2,5,main]
background task cancelled  1
background task cancelled  2
Finish #0: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Background task finished:task 0
Start #3: Thread[ForkJoinPool.commonPool-worker-2,5,main]
background task cancelled  4
Finish #3: Thread[ForkJoinPool.commonPool-worker-2,5,main]
background task cancelled  5
Background task finished:task 3
Start #6: Thread[ForkJoinPool.commonPool-worker-3,5,main]
background task cancelled  7
Finish #6: Thread[ForkJoinPool.commonPool-worker-3,5,main]
background task cancelled  8
Background task finished:task 6
Start #9: Thread[ForkJoinPool.commonPool-worker-2,5,main]
background task cancelled  10
Cancelled

Here is my solution for your task:

public class LoadInBackgroundExample {
    //Added new exception
    public static class AlreadyIsRunningException extends RuntimeException {
        long taskId;

        public AlreadyIsRunningException(String message, long taskId) {
            super(message);
            this.taskId = taskId;
        }

        public long getTaskId() {
            return taskId;
        }

        public void setTaskId(long taskId) {
            this.taskId = taskId;
        }
    }


    /**
     * A simple background task which should perform the data loading operation. In this minimal example it simply invokes Thread.sleep
     */
    public static class BackgroundTask implements Runnable {


        //this atomicInteger acts as a global lock counter for BackgroundTask objects
        private static AtomicInteger counter = new AtomicInteger(0);


        private int id;

        public BackgroundTask(int id) {
            this.id = id;
        }

        private void unlock() {
            counter.decrementAndGet();
        }

        private void lock() {
            //we need to check this way to avoid some unlucky timing between threads
            int lockValue = counter.incrementAndGet();
            //if we got counter different than 1 that means that some other task is already running (it has already acquired the lock)
            if (lockValue != 1) {
                //rollback our check
                counter.decrementAndGet();
                //throw an exception
                throw new AlreadyIsRunningException("Some other task already is running", id);
            }
        }

        /**
         * Sleep for a given amount of time to simulate loading.
         */
        @Override
        public void run() {
            //Check if we can acquire lock

            lock();
            //we have a lock to
            try {
                System.out.println("Start #" + id + ": " + Thread.currentThread());
                long sleepTime = 2000;
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.println("Finish #" + id + ": " + Thread.currentThread());
                unlock();
            }
        }
    }

    /**
     * CompletableFuture which simulates loading and showing data.
     *
     * @param taskId Identifier of the current task
     */


    public static void loadInBackground(int taskId) {
        // create the loading task
        BackgroundTask backgroundTask = new BackgroundTask(taskId);
        // "load" the data asynchronously
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                CompletableFuture<Void> future = CompletableFuture.runAsync(backgroundTask);

                try {
                    future.get();
                } catch (ExecutionException e) {
                    if (e.getCause() instanceof AlreadyIsRunningException) {
                        System.out.println("background task cancelled  " + ((AlreadyIsRunningException) e.getCause()).getTaskId());
                        throw (AlreadyIsRunningException) e.getCause();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "task " + backgroundTask.id;
            }
        });
        // display the data after they are loaded
        CompletableFuture<Void> future = completableFuture.thenAccept(x -> {
            System.out.println("Background task finished:" + x);
        });
    }

    ArrayList<BackgroundTask> backgroundTasks = new ArrayList<>();


    public static void main(String[] args) {
        // runnable which invokes the background loader every second
        Runnable trigger = new Runnable() {
            int taskId = 0;

            public void run() {
                loadInBackground(taskId++);
            }
        };

        // create scheduler
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        ScheduledFuture<?> beeperHandle = scheduler.scheduleAtFixedRate(trigger, 0, 1, TimeUnit.SECONDS);

        // cancel the scheudler and the application after 10 seconds
        scheduler.schedule(() -> beeperHandle.cancel(true), 10, TimeUnit.SECONDS);

        try {
            beeperHandle.get();
        } catch (Throwable th) {
        }

        System.out.println("Cancelled");
        System.exit(0);
    }

UPDATE

I have changed lock() and unlock() methods to more simple form:

private static  AtomicBoolean atomicBoolean = new AtomicBoolean(false);
        private void unlock() {
            atomicBoolean.set(false);
        }
        private void lock() {
            //if 'changed' is false that means some other task is already running
            boolean changed = atomicBoolean.compareAndSet(false,true);
            if (!changed) {
                throw new AlreadyIsRunningException("Some other task  is already running", id);
            }
        }

Comments

2

If understand you have several tasks into background at the same time. as these tasks are doing exact same job you don't want to execute them in parallel you need one task to finish the job and share its results to others. So if you get 10 CompletableFuture simultaneously you want one of them call 'reload' into db and share execution results to others in a way that all the CompletableFuture will complete normally with result. I assume this from

The goal is to have e. g. #1 and #2 skipped because #0 is still running.

and

after the loading the data should be displayed

if my guesses are correct you can try my solution.

I have some kind of parent-child relationship between tasks. parent task is the one which is really doing its job and shares gotten result to its children. child task is a task which was added while parent task was still executing, child task waits until parent task finish its execution. As parents task's results is still 'fresh' they are copied into each children and all of them completing their futures.

public class BackgroundService {


    public static class BackgroundJob implements Callable<String> {

        private static BackgroundJob ROOT_JOB = null;

        private synchronized static void addBackgroundJob(BackgroundJob backgroundJob) {
            if (ROOT_JOB != null) {
                ROOT_JOB.addChild(backgroundJob);
            } else {
                System.out.println();
                System.out.println(Thread.currentThread().getName() + " RUNNING ROOT TASK-" + backgroundJob.jobId);
                ROOT_JOB = backgroundJob;
            }
        }

        private synchronized static void unlock() {
            ROOT_JOB = null;
        }

        private final int jobId;
        private List<BackgroundJob> children = new ArrayList<>();
        private BackgroundJob parent;
        private String providedResultFromParent = null;


        public BackgroundJob(int jobId) {
            this.jobId = jobId;
        }

        private  void addChild(BackgroundJob backgroundJob) {
            backgroundJob.parent = this;
            this.children.add(backgroundJob);
        }

        @Override
        public String call() throws Exception {
            addBackgroundJob(this);
            if (parent == null) {
                String result = logic();

                synchronized (ROOT_JOB) {
                    for (final BackgroundJob backgroundJob : children) {
                        backgroundJob.providedResultFromParent = result;
                        synchronized (backgroundJob) {
                            backgroundJob.notify();
                        }
                    }
                    unlock();
                }

                return "\t\tROOT task" + jobId + "'s " + result;
            } else {
                synchronized (this) {
                    System.out.println(Thread.currentThread().getName() + "\t\tskipping task-" + jobId + " and waiting running task-" + parent.jobId + " to finish");
                    this.wait();
                }
                return "\t\t\t\ttask-" + jobId + "'s " + providedResultFromParent;
            }
        }

        private String logic() throws InterruptedException {
            Thread.sleep(2000);
            return (int) (Math.random() * 1000) + " ";
        }
    }

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        AtomicInteger atomicInteger = new AtomicInteger();
        ExecutorService pool = Executors.newCachedThreadPool();
        Supplier<String> job = () -> {
            int taskId = atomicInteger.incrementAndGet();
            BackgroundJob backgroundJob = new BackgroundJob(taskId);
            try {
                return backgroundJob.call();
            } catch (Exception e) {
                e.printStackTrace();
            }
            return "finished " + taskId;
        };

        for (int i = 100; i > 0; i--) {
            CompletableFuture.supplyAsync(job, pool).thenAccept(s -> System.out.println(Thread.currentThread().getName()+" "+ s + " result is readable"));
            Thread.sleep((long) (Math.random() * 500));
        }

        pool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
        pool.shutdown();
    }

And here is the output:

pool-1-thread-1 RUNNING ROOT TASK-1
pool-1-thread-2     skipping task-2 and waiting running task-1 to finish
pool-1-thread-3     skipping task-3 and waiting running task-1 to finish
pool-1-thread-4     skipping task-4 and waiting running task-1 to finish
pool-1-thread-5     skipping task-5 and waiting running task-1 to finish
pool-1-thread-6     skipping task-6 and waiting running task-1 to finish
pool-1-thread-7     skipping task-7 and waiting running task-1 to finish
pool-1-thread-8     skipping task-8 and waiting running task-1 to finish
pool-1-thread-3                 task-3's 165  result is readable
pool-1-thread-8                 task-8's 165  result is readable
pool-1-thread-6                 task-6's 165  result is readable
pool-1-thread-5                 task-5's 165  result is readable
pool-1-thread-7                 task-7's 165  result is readable
pool-1-thread-2                 task-2's 165  result is readable
pool-1-thread-1         ROOT task1's 165  result is readable
pool-1-thread-4                 task-4's 165  result is readable

pool-1-thread-4 RUNNING ROOT TASK-9
pool-1-thread-1     skipping task-10 and waiting running task-9 to finish
pool-1-thread-2     skipping task-11 and waiting running task-9 to finish
pool-1-thread-7     skipping task-12 and waiting running task-9 to finish
pool-1-thread-5     skipping task-13 and waiting running task-9 to finish
pool-1-thread-8     skipping task-14 and waiting running task-9 to finish
pool-1-thread-6     skipping task-15 and waiting running task-9 to finish
pool-1-thread-3     skipping task-16 and waiting running task-9 to finish
pool-1-thread-9     skipping task-17 and waiting running task-9 to finish
pool-1-thread-10        skipping task-18 and waiting running task-9 to finish
pool-1-thread-1                 task-10's 370  result is readable
pool-1-thread-10                task-18's 370  result is readable
pool-1-thread-4         ROOT task9's 370  result is readable
pool-1-thread-9                 task-17's 370  result is readable
pool-1-thread-7                 task-12's 370  result is readable
pool-1-thread-6                 task-15's 370  result is readable
pool-1-thread-8                 task-14's 370  result is readable
pool-1-thread-2                 task-11's 370  result is readable
pool-1-thread-3                 task-16's 370  result is readable
pool-1-thread-5                 task-13's 370  result is readable

pool-1-thread-5 RUNNING ROOT TASK-19
pool-1-thread-3     skipping task-20 and waiting running task-19 to finish
pool-1-thread-2     skipping task-21 and waiting running task-19 to finish
pool-1-thread-8     skipping task-22 and waiting running task-19 to finish
pool-1-thread-6     skipping task-23 and waiting running task-19 to finish
pool-1-thread-7     skipping task-24 and waiting running task-19 to finish
pool-1-thread-9     skipping task-25 and waiting running task-19 to finish
pool-1-thread-4     skipping task-26 and waiting running task-19 to finish
pool-1-thread-10        skipping task-27 and waiting running task-19 to finish
pool-1-thread-1     skipping task-28 and waiting running task-19 to finish
pool-1-thread-5         ROOT task19's 574  result is readable
pool-1-thread-8                 task-22's 574  result is readable
pool-1-thread-4                 task-26's 574  result is readable
pool-1-thread-7                 task-24's 574  result is readable
pool-1-thread-6                 task-23's 574  result is readable
pool-1-thread-3                 task-20's 574  result is readable
pool-1-thread-9                 task-25's 574  result is readable
pool-1-thread-2                 task-21's 574  result is readable
pool-1-thread-1                 task-28's 574  result is readable
pool-1-thread-10                task-27's 574  result is readable

pool-1-thread-10 RUNNING ROOT TASK-29
pool-1-thread-1     skipping task-30 and waiting running task-29 to finish
pool-1-thread-2     skipping task-31 and waiting running task-29 to finish
pool-1-thread-9     skipping task-32 and waiting running task-29 to finish
pool-1-thread-3     skipping task-33 and waiting running task-29 to finish
pool-1-thread-6     skipping task-34 and waiting running task-29 to finish
pool-1-thread-7     skipping task-35 and waiting running task-29 to finish
pool-1-thread-4     skipping task-36 and waiting running task-29 to finish
pool-1-thread-8     skipping task-37 and waiting running task-29 to finish
pool-1-thread-5     skipping task-38 and waiting running task-29 to finish
pool-1-thread-11        skipping task-39 and waiting running task-29 to finish
pool-1-thread-1                 task-30's 230  result is readable
pool-1-thread-11                task-39's 230  result is readable
pool-1-thread-8                 task-37's 230  result is readable
pool-1-thread-5                 task-38's 230  result is readable
pool-1-thread-4                 task-36's 230  result is readable
pool-1-thread-7                 task-35's 230  result is readable

pool-1-thread-12 RUNNING ROOT TASK-40
pool-1-thread-6                 task-34's 230  result is readable
pool-1-thread-10        ROOT task29's 230  result is readable
pool-1-thread-3                 task-33's 230  result is readable
pool-1-thread-9                 task-32's 230  result is readable
pool-1-thread-2                 task-31's 230  result is readable
pool-1-thread-2     skipping task-41 and waiting running task-40 to finish
pool-1-thread-9     skipping task-42 and waiting running task-40 to finish
pool-1-thread-3     skipping task-43 and waiting running task-40 to finish
pool-1-thread-10        skipping task-44 and waiting running task-40 to finish
pool-1-thread-6     skipping task-45 and waiting running task-40 to finish
pool-1-thread-7     skipping task-46 and waiting running task-40 to finish
pool-1-thread-2                 task-41's 282  result is readable
pool-1-thread-10                task-44's 282  result is readable
pool-1-thread-6                 task-45's 282  result is readable
pool-1-thread-7                 task-46's 282  result is readable
pool-1-thread-3                 task-43's 282  result is readable
pool-1-thread-9                 task-42's 282  result is readable
pool-1-thread-12        ROOT task40's 282  result is readable

pool-1-thread-12 RUNNING ROOT TASK-47
pool-1-thread-9     skipping task-48 and waiting running task-47 to finish
pool-1-thread-3     skipping task-49 and waiting running task-47 to finish
pool-1-thread-7     skipping task-50 and waiting running task-47 to finish
pool-1-thread-6     skipping task-51 and waiting running task-47 to finish
pool-1-thread-10        skipping task-52 and waiting running task-47 to finish
pool-1-thread-2     skipping task-53 and waiting running task-47 to finish
pool-1-thread-12        ROOT task47's 871  result is readable
pool-1-thread-10                task-52's 871  result is readable
pool-1-thread-2                 task-53's 871  result is readable
pool-1-thread-3                 task-49's 871  result is readable
pool-1-thread-6                 task-51's 871  result is readable
pool-1-thread-7                 task-50's 871  result is readable
pool-1-thread-9                 task-48's 871  result is readable

pool-1-thread-9 RUNNING ROOT TASK-54
pool-1-thread-7     skipping task-55 and waiting running task-54 to finish
pool-1-thread-6     skipping task-56 and waiting running task-54 to finish
pool-1-thread-3     skipping task-57 and waiting running task-54 to finish
pool-1-thread-2     skipping task-58 and waiting running task-54 to finish
pool-1-thread-10        skipping task-59 and waiting running task-54 to finish
pool-1-thread-12        skipping task-60 and waiting running task-54 to finish
pool-1-thread-4     skipping task-61 and waiting running task-54 to finish
pool-1-thread-5     skipping task-62 and waiting running task-54 to finish
pool-1-thread-9         ROOT task54's 345  result is readable
pool-1-thread-2                 task-58's 345  result is readable
pool-1-thread-5                 task-62's 345  result is readable
pool-1-thread-7                 task-55's 345  result is readable
pool-1-thread-10                task-59's 345  result is readable
pool-1-thread-6                 task-56's 345  result is readable
pool-1-thread-3                 task-57's 345  result is readable
pool-1-thread-4                 task-61's 345  result is readable
pool-1-thread-12                task-60's 345  result is readable

pool-1-thread-12 RUNNING ROOT TASK-63
pool-1-thread-4     skipping task-64 and waiting running task-63 to finish
pool-1-thread-3     skipping task-65 and waiting running task-63 to finish
pool-1-thread-6     skipping task-66 and waiting running task-63 to finish
pool-1-thread-10        skipping task-67 and waiting running task-63 to finish
pool-1-thread-7     skipping task-68 and waiting running task-63 to finish
pool-1-thread-5     skipping task-69 and waiting running task-63 to finish
pool-1-thread-2     skipping task-70 and waiting running task-63 to finish
pool-1-thread-12        ROOT task63's 670  result is readable
pool-1-thread-2                 task-70's 670  result is readable
pool-1-thread-5                 task-69's 670  result is readable
pool-1-thread-7                 task-68's 670  result is readable
pool-1-thread-10                task-67's 670  result is readable
pool-1-thread-6                 task-66's 670  result is readable
pool-1-thread-3                 task-65's 670  result is readable
pool-1-thread-4                 task-64's 670  result is readable

pool-1-thread-4 RUNNING ROOT TASK-71
pool-1-thread-3     skipping task-72 and waiting running task-71 to finish
pool-1-thread-6     skipping task-73 and waiting running task-71 to finish
pool-1-thread-10        skipping task-74 and waiting running task-71 to finish
pool-1-thread-7     skipping task-75 and waiting running task-71 to finish
pool-1-thread-5     skipping task-76 and waiting running task-71 to finish
pool-1-thread-2     skipping task-77 and waiting running task-71 to finish
pool-1-thread-12        skipping task-78 and waiting running task-71 to finish
pool-1-thread-9     skipping task-79 and waiting running task-71 to finish
pool-1-thread-8     skipping task-80 and waiting running task-71 to finish
pool-1-thread-4         ROOT task71's 445  result is readable
pool-1-thread-6                 task-73's 445  result is readable
pool-1-thread-9                 task-79's 445  result is readable
pool-1-thread-3                 task-72's 445  result is readable
pool-1-thread-8                 task-80's 445  result is readable
pool-1-thread-12                task-78's 445  result is readable
pool-1-thread-5                 task-76's 445  result is readable
pool-1-thread-10                task-74's 445  result is readable
pool-1-thread-2                 task-77's 445  result is readable
pool-1-thread-7                 task-75's 445  result is readable

pool-1-thread-7 RUNNING ROOT TASK-81
pool-1-thread-2     skipping task-82 and waiting running task-81 to finish
pool-1-thread-10        skipping task-83 and waiting running task-81 to finish
pool-1-thread-5     skipping task-84 and waiting running task-81 to finish
pool-1-thread-12        skipping task-85 and waiting running task-81 to finish
pool-1-thread-8     skipping task-86 and waiting running task-81 to finish
pool-1-thread-3     skipping task-87 and waiting running task-81 to finish
pool-1-thread-9     skipping task-88 and waiting running task-81 to finish
pool-1-thread-6     skipping task-89 and waiting running task-81 to finish
pool-1-thread-7         ROOT task81's 141  result is readable
pool-1-thread-6                 task-89's 141  result is readable
pool-1-thread-9                 task-88's 141  result is readable
pool-1-thread-3                 task-87's 141  result is readable
pool-1-thread-10                task-83's 141  result is readable
pool-1-thread-5                 task-84's 141  result is readable
pool-1-thread-12                task-85's 141  result is readable
pool-1-thread-8                 task-86's 141  result is readable
pool-1-thread-2                 task-82's 141  result is readable

pool-1-thread-2 RUNNING ROOT TASK-90
pool-1-thread-8     skipping task-91 and waiting running task-90 to finish
pool-1-thread-12        skipping task-92 and waiting running task-90 to finish
pool-1-thread-5     skipping task-93 and waiting running task-90 to finish
pool-1-thread-10        skipping task-94 and waiting running task-90 to finish
pool-1-thread-3     skipping task-95 and waiting running task-90 to finish
pool-1-thread-9     skipping task-96 and waiting running task-90 to finish
pool-1-thread-6     skipping task-97 and waiting running task-90 to finish
pool-1-thread-7     skipping task-98 and waiting running task-90 to finish
pool-1-thread-4     skipping task-99 and waiting running task-90 to finish
pool-1-thread-11        skipping task-100 and waiting running task-90 to finish
pool-1-thread-2         ROOT task90's 321  result is readable
pool-1-thread-3                 task-95's 321  result is readable
pool-1-thread-7                 task-98's 321  result is readable
pool-1-thread-8                 task-91's 321  result is readable
pool-1-thread-11                task-100's 321  result is readable
pool-1-thread-4                 task-99's 321  result is readable
pool-1-thread-5                 task-93's 321  result is readable
pool-1-thread-9                 task-96's 321  result is readable
pool-1-thread-12                task-92's 321  result is readable
pool-1-thread-10                task-94's 321  result is readable
pool-1-thread-6                 task-97's 321  result is readable

Comments

0

if you just want one single accessing thread a simple synchronized would do the job...

output:

Start #2: Thread[ForkJoinPool.commonPool-worker-6,5,main]
Finish #0: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Background task finished:task 0 finished getting data...
Start #3: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Finish #2: Thread[ForkJoinPool.commonPool-worker-6,5,main]
Start #4: Thread[ForkJoinPool.commonPool-worker-6,5,main]
Background task finished:task 2 finished getting data...
Finish #1: Thread[ForkJoinPool.commonPool-worker-4,5,main]
Background task finished:task 1 finished getting data...
Start #6: Thread[ForkJoinPool.commonPool-worker-3,5,main]
Finish #4: Thread[ForkJoinPool.commonPool-worker-6,5,main]
Start #5: Thread[ForkJoinPool.commonPool-worker-6,5,main]
Background task finished:task 4 finished getting data...
Finish #3: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Start #7: Thread[ForkJoinPool.commonPool-worker-2,5,main]
Background task finished:task 3 finished getting data...
Cancelled

code:

package queue;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

public class LoadInBackgroundExample {

    public static class SyncronizedBackend {
        public synchronized String getData() {

            long sleepTime = 2000;
            try {
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }

            return new String("finished getting data...");
        }
    }

    /**
     * A simple background task which should perform the data loading operation. In
     * this minimal example it simply invokes Thread.sleep
     */
    public static class BackgroundTask implements Runnable {

        private int id;
        private SyncronizedBackend syncronizedBackend;
        private String result;

        public BackgroundTask(SyncronizedBackend syncronizedBackend, int id) {
            this.syncronizedBackend = syncronizedBackend;
            this.id = id;
        }

        /**
         * Sleep for a given amount of time to simulate loading.
         */
        @Override
        public void run() {

            System.out.println("Start #" + id + ": " + Thread.currentThread());

            result = this.syncronizedBackend.getData();

            System.out.println("Finish #" + id + ": " + Thread.currentThread());
        }

        public String getResult() {
            return result;
        }
    }

    /**
     * CompletableFuture which simulates loading and showing data.
     * @param syncronizedBackend 
     * 
     * @param taskId Identifier of the current task
     */
    public static void loadInBackground(SyncronizedBackend syncronizedBackend, int taskId) {

        // create the loading task
        BackgroundTask backgroundTask = new BackgroundTask(syncronizedBackend, taskId);

        // "load" the data asynchronously
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(new Supplier<String>() {

            @Override
            public String get() {

                CompletableFuture<Void> future = CompletableFuture.runAsync(backgroundTask);

                try {

                    future.get();

                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }

                return "task " + backgroundTask.id + " " + backgroundTask.getResult();
            }
        });

        // display the data after they are loaded
        CompletableFuture<Void> future = completableFuture.thenAccept(x -> {

            System.out.println("Background task finished:" + x);

        });

    }

    public static void main(String[] args) {

        SyncronizedBackend syncronizedBackend = new SyncronizedBackend();

        // runnable which invokes the background loader every second
        Runnable trigger = new Runnable() {

            int taskId = 0;

            public void run() {

                loadInBackground(syncronizedBackend, taskId++);

            }
        };

        // create scheduler
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        ScheduledFuture<?> beeperHandle = scheduler.scheduleAtFixedRate(trigger, 0, 1, TimeUnit.SECONDS);

        // cancel the scheudler and the application after 10 seconds
        scheduler.schedule(() -> beeperHandle.cancel(true), 10, TimeUnit.SECONDS);

        try {
            beeperHandle.get();
        } catch (Throwable th) {
        }

        System.out.println("Cancelled");
        System.exit(0);
    }

}

Comments

0

I tried a solution using a Thread dual switch, see class BackgroundTaskDualSwitch, it simulates loading using the CompletableFuture. The idea is to let a second task wait until the currently running task is finished, see change in BackgroundTask. This ensures that max one task Thread is running and max one task Thread is waiting. Further requests are skipped until the running task is completed and becomes 'free' to handle the next request.

public static class BackgroundTask extends Thread {

  private int id;
  private Thread pendingTask;

  public BackgroundTask(int id) {
    this.id = id;
  }

  public BackgroundTask(int id, Thread pendingTask) {
    this(id);
    this.pendingTask = pendingTask;
  }

  /**
   * Sleep for a given amount of time to simulate loading.
   */
  @Override
  public void run() {

    try {

      if (pendingTask != null && pendingTask.isAlive()) {
        pendingTask.join();
      }

      System.out.println("Start #" + id + ": " + Thread.currentThread());
      ...
    }
  }

public static class BackgroundTaskDualSwitch {

  private static BackgroundTask task1;
  private static BackgroundTask task2;

  public static synchronized boolean runTask(int taskId) {
    if (! isBusy(task1)) {
      if (isBusy(task2)) {
        task1 = new BackgroundTask(taskId, task2);
      } else {
        task1 = new BackgroundTask(taskId);
      }
      runAsync(task1);
      return true;
    } else if (! isBusy(task2)) {
      if (isBusy(task1)) {
        task2 = new BackgroundTask(taskId, task1);
      } else {
        task2 = new BackgroundTask(taskId);
      }
      runAsync(task2);
      return true;
    } else {
      return false; // SKIPPED
    }
  }

  private static void runAsync(BackgroundTask task) {
    CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(new Supplier<String>() {

      @Override
      public String get() {

        try {
          task.start();
          task.join();
        }
        catch (InterruptedException e) {
          e.printStackTrace();
        }
        return "task " + task.id;
      }
    });

    // display the data after they are loaded
    CompletableFuture<Void> future = completableFuture.thenAccept(x -> {

      System.out.println( "Background task finished:" + x); 

    });
  }

  private static boolean isBusy(BackgroundTask task) {
    return task != null && task.isAlive();
  }
}

/**
 * Simulates loading and showing data.
 * @param taskId Identifier of the current task
 */
public static void loadInBackground(int taskId) {

  // create the loading task
  if (! BackgroundTaskDualSwitch.runTask(taskId)) {
    System.out.println( "Background task ignored:task " + taskId); // SKIPPED
  }
}
...

Output is:

Start #0: Thread[Thread-0,5,main]
Background task ignored:task 2
Finish #0: Thread[Thread-0,5,main]
Start #1: Thread[Thread-1,5,main]
Background task finished:task 0
Background task ignored:task 4
Finish #1: Thread[Thread-1,5,main]
Start #3: Thread[Thread-2,5,main]
Background task finished:task 1
Background task ignored:task 6
Finish #3: Thread[Thread-2,5,main]
Start #5: Thread[Thread-3,5,main]
Background task finished:task 3
Background task ignored:task 8
Finish #5: Thread[Thread-3,5,main]
Start #7: Thread[Thread-4,5,main]
Background task finished:task 5
Background task ignored:task 10
Finish #7: Thread[Thread-4,5,main]
Start #9: Thread[Thread-5,5,main]
Background task finished:task 7
Cancelled

Comments

0

The first thread who starts to do the expensive work will notify with a callback the result. Other threads that tries to execute it will be registered in ExpensiveWork.notificables, so once the expensive work finish the thread who did the work will notify them.

Meanwhile the threads are checking for the result every 5 seconds.

public class ExpensiveWorkTest {

    private final static int THREADS = 20;
    private final static long THREAD_TIMEOUT = 5000L;

    @Test
    public void example() throws InterruptedException {
        ExpensiveWork<String> expensiveWork = new ExpensiveWorkImpl();
        ExecutorService service = Executors.newFixedThreadPool(THREADS);
        for(int i=0; i<THREADS;i++) {
            service.execute(() ->{
                Notificable<String> notificable = new NotificableImpl();
                expensiveWork.execute(notificable);
                while(notificable.getExpensiveResult() == null) {
                    try {
                        Thread.sleep(THREAD_TIMEOUT);
                    } catch (InterruptedException e) {}
                }
                System.out.println(Thread.currentThread().getName()+" has the message: "+notificable.getExpensiveResult());
            });
        }
        service.awaitTermination(60, TimeUnit.SECONDS);
    }

    public static abstract class ExpensiveWork<T> {

        private final AtomicBoolean runnning = new AtomicBoolean(false);
        private List<Notificable<T>> notificables = Collections.synchronizedList(new ArrayList<>());

        public void execute(Notificable<T> notificable) {
            String id = Thread.currentThread().getName();
            System.out.println("Loading data for "+id);
            notificables.add(notificable);
            if(!runnning.getAndSet(true)) {
                System.out.println("Running the expensive work "+id);
                T expensiveResult = expensiveWork();
                notificables.stream().forEach(n -> n.callback(expensiveResult));
            } else {
                System.out.println(id+" will receive the response later");
            }
        }

        protected abstract T expensiveWork();

    }

    public static class ExpensiveWorkImpl extends ExpensiveWork<String>{

        @Override
        public String expensiveWork() {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {}
            return "<Expensive result>";
        }

    }

    public static interface Notificable<T> {
        void callback(T expensiveResult);
        T getExpensiveResult();
    }

    public static class NotificableImpl implements Notificable<String> {

        private volatile String expensiveResult;

        @Override
        public void callback(String expensiveResult) {
            this.expensiveResult = expensiveResult;
        }

        @Override
        public String getExpensiveResult() {
            return expensiveResult;
        }

    }

}

And this is the output:

Loading data for pool-1-thread-6
Loading data for pool-1-thread-7
Loading data for pool-1-thread-9
pool-1-thread-9 will receive the response later
Loading data for pool-1-thread-8
pool-1-thread-8 will receive the response later
Loading data for pool-1-thread-1
Loading data for pool-1-thread-2
Loading data for pool-1-thread-12
pool-1-thread-12 will receive the response later
Loading data for pool-1-thread-3
pool-1-thread-3 will receive the response later
Loading data for pool-1-thread-4
Loading data for pool-1-thread-5
pool-1-thread-5 will receive the response later
pool-1-thread-4 will receive the response later
Loading data for pool-1-thread-14
pool-1-thread-14 will receive the response later
Loading data for pool-1-thread-13
Loading data for pool-1-thread-15
pool-1-thread-15 will receive the response later
pool-1-thread-2 will receive the response later
pool-1-thread-1 will receive the response later
Loading data for pool-1-thread-11
pool-1-thread-11 will receive the response later
Loading data for pool-1-thread-10
pool-1-thread-10 will receive the response later
pool-1-thread-7 will receive the response later
Running the expensive work pool-1-thread-6
Loading data for pool-1-thread-18
pool-1-thread-18 will receive the response later
Loading data for pool-1-thread-17
Loading data for pool-1-thread-16
pool-1-thread-16 will receive the response later
pool-1-thread-13 will receive the response later
Loading data for pool-1-thread-19
pool-1-thread-19 will receive the response later
pool-1-thread-17 will receive the response later
Loading data for pool-1-thread-20
pool-1-thread-20 will receive the response later
pool-1-thread-6 has the message: <Expensive result>
pool-1-thread-8 has the message: <Expensive result>
pool-1-thread-12 has the message: <Expensive result>
pool-1-thread-9 has the message: <Expensive result>
pool-1-thread-11 has the message: <Expensive result>
pool-1-thread-1 has the message: <Expensive result>
pool-1-thread-2 has the message: <Expensive result>
pool-1-thread-3 has the message: <Expensive result>
pool-1-thread-15 has the message: <Expensive result>
pool-1-thread-4 has the message: <Expensive result>
pool-1-thread-14 has the message: <Expensive result>
pool-1-thread-10 has the message: <Expensive result>
pool-1-thread-5 has the message: <Expensive result>
pool-1-thread-13 has the message: <Expensive result>
pool-1-thread-16 has the message: <Expensive result>
pool-1-thread-19 has the message: <Expensive result>
pool-1-thread-20 has the message: <Expensive result>
pool-1-thread-7 has the message: <Expensive result>
pool-1-thread-18 has the message: <Expensive result>
pool-1-thread-17 has the message: <Expensive result>

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.