java - 异步地将数据放入缓存中

Question

我正在开发一个负载较重的应用程序，并且希望能够将数据异步放入缓存中。我的意思是调用缓存上的操作并接收 IgniteFuture<V>代表结果。

我的意思是我需要能够并行化从持久性存储中提取数据并将其放入缓存中。如果发生什么情况，我可以尝试再次提取数据(数据量不要太多，粒度相当细)。

Answer 1

如果您对 IgniteFuture 没有硬性要求(我认为不应该是这种情况)并且您想要的只是某种机制将数据放入缓存并异步处理它，然后处理该操作返回的结果，然后您可以使用Java's Executor Service .

如果您不太了解 Java 的执行器服务，那么您可能需要阅读文档或 this answer其中突出显示了要点，在下面的示例中我还添加了注释。

以下是有关线程数的其他一些要点:

• 此示例创建 ExecutorService对于“ThreadPoolExecutor ”实现，还有其他选项，例如“Executors.newSingleThreadExecutor() ”和“Executors.newFixedThreadPool(10) ”，它们允许您定义 JVM 中需要多少个线程。
• 您也可以选择直接创建 ThreadPoolExecutor 的对象像这样return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());您可以更好地控制线程数量和队列类型。如果您想创建ThreadPoolExecutor，您将需要阅读更多相关内容。反对自己而不是通过ExecutorService

与您的实现相关的其他几点:

• 当您处理 Future 时对象，那么它是阻塞进程，因为 Future.get()是一个阻塞调用，因此您的主线程将被阻塞，直到返回并处理所有 Future 对象。
• 如果您不想阻塞，那么有几个选项，例如您可以创建一个新线程来完成所有这些处理，从而释放主线程。另一种选择可能是不处理 Future 对象，所以只要你这样做 executorService.submit(callableTask1) ，你的线程将是空闲的，然后在你的 CallableTask对象，您可以将结果推送到队列中(您可以根据需要选择Java's queue implementation)，然后从其他线程处理该队列。另一种选择可能是专门使用另一个线程来处理您的 Future 对象。

示例代码:

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class ExecutorServiceFutureCallableExample {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        List<Future<String>> futuresList = new ArrayList<>();

        ExecutorService executorService = Executors.newCachedThreadPool();
        ExecutorServiceFutureCallableExample.CallableTask callableTask1 = new ExecutorServiceFutureCallableExample.CallableTask(2000);
        ExecutorServiceFutureCallableExample.CallableTask callableTask2 = new ExecutorServiceFutureCallableExample.CallableTask(1000);
        ExecutorServiceFutureCallableExample.CallableTask callableTask3 = new ExecutorServiceFutureCallableExample.CallableTask(3000);

        System.out.println("### Starting submitting tasks");

        // submit the callable and register the returned future object so that it can be processed later.
        futuresList.add(executorService.submit(callableTask1));
        futuresList.add(executorService.submit(callableTask2));
        futuresList.add(executorService.submit(callableTask3));

        System.out.println("### Finished submitting tasks");

        for (int i = 0; i < futuresList.size(); i++) {
            // here "get()" waits for the future tasks to be returned.
            System.out.println(futuresList.get(i).get());
        }

        System.out.println("### Finished.");
    }

    static class CallableTask implements Callable<String>{

        private long timeToSleep;

        CallableTask(long _timeToSleep){
            this.timeToSleep = _timeToSleep;
        }

        @Override
        public String call() throws Exception {
            String str = new Date() + ": Processing - " + this.hashCode() + " | " + Thread.currentThread() + ", slept for seconds - " + timeToSleep; 
            System.out.println(str);
            Thread.sleep(timeToSleep);
            return str + " ||||| completed at: " + new Date();
        }

        public long getTimeToSleep() {
            return timeToSleep;
        }

        public void setTimeToSleep(long timeToSleep) {
            this.timeToSleep = timeToSleep;
        }

    }
}