Follow these steps to implement the example:
- Create a class named ReportGenerator and specify that it implements the Callable interface parameterized by the String class:
public class ReportGenerator implements Callable<String> {
- Declare two private String attributes named sender and title. These attributes will represent the data of the report:
private final String sender;
private final String title;
- Implement the constructor of the class that initializes the two attributes:
public ReportGenerator(String sender, String title){
this.sender=sender;
this.title=title;
}
- Implement the call() method. First, put the thread to sleep for a random period of time:
@Override
public String call() throws Exception {
try {
Long duration=(long)(Math.random()*10);
System.out.printf("%s_%s: ReportGenerator: Generating a
report during %d seconds ",this.sender,
this.title,duration);
TimeUnit.SECONDS.sleep(duration);
} catch (InterruptedException e) {
e.printStackTrace();
}
- Then, generate the report as a string with the sender and title attributes and return that string:
String ret=sender+": "+title;
return ret;
}
- Create a class named ReportRequest and specify that it implements the Runnable interface. Thiss class will simulate some report requests:
public class ReportRequest implements Runnable {
- Declare a private String attribute called name to store the name of ReportRequest:
private final String name;
- Declare a private CompletionService attribute named service. The CompletionService interface is a parameterized interface. Use the String class:
private final CompletionService<String> service;
- Implement the constructor of the class that initializes the two attributes:
public ReportRequest(String name, CompletionService<String>
service){
this.name=name;
this.service=service;
}
- Implement the run() method. Create three ReportGenerator objects and send them to the CompletionService object using the submit() method:
@Override
public void run() {
ReportGenerator reportGenerator=new ReportGenerator(name,
"Report");
service.submit(reportGenerator);
}
- Create a class named ReportProcessor. This class will get the results of the ReportGenerator tasks. Specify that it implements the Runnable interface:
public class ReportProcessor implements Runnable {
- Declare a private CompletionService attribute named service. As the CompletionService interface is a parameterized interface, use the String class as a parameter of this CompletionService interface:
private final CompletionService<String> service;
- Declare a private Boolean attribute named end. Add the volatile keyword to ensure that all the threads have access to the actual value of the attribute:
private volatile boolean end;
- Implement the constructor of the class to initialize the two attributes:
public ReportProcessor (CompletionService<String> service){
this.service=service;
end=false;
}
- Implement the run() method. While the end attribute is false, call the poll() method of the CompletionService interface to get the Future object of the next task executed by the completion service that has finished:
@Override
public void run() {
while (!end){
try {
Future<String> result=service.poll(20, TimeUnit.SECONDS);
- Then, get the results of the task using the get() method of the Future object and write the results to the console:
if (result!=null) {
String report=result.get();
System.out.printf("ReportReceiver: Report Received: %s
",
report);
}
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
System.out.printf("ReportSender: End
");
}
- Implement the stopProcessing() method that modifies the value of the end attribute:
public void stopProcessing() {
this.end = true;
}
- Implement the main class of the example by creating a class named Main and adding the main() method to it:
public class Main {
public static void main(String[] args) {
- Create ThreadPoolExecutor using the newCachedThreadPool() method of the Executors class:
ExecutorService executor=Executors.newCachedThreadPool();
- Create CompletionService using the executor created earlier as a parameter of the constructor:
CompletionService<String> service=new
ExecutorCompletionService<>(executor);
- Create two ReportRequest objects and the threads to execute them:
ReportRequest faceRequest=new ReportRequest("Face", service);
ReportRequest onlineRequest=new ReportRequest("Online", service);
Thread faceThread=new Thread(faceRequest);
Thread onlineThread=new Thread(onlineRequest);
- Create a ReportProcessor object and the thread to execute it:
ReportProcessor processor=new ReportProcessor(service);
Thread senderThread=new Thread(processor);
- Start the three threads:
System.out.printf("Main: Starting the Threads
");
faceThread.start();
onlineThread.start();
senderThread.start();
- Wait for the finalization of the ReportRequest threads:
try {
System.out.printf("Main: Waiting for the report generators.
");
faceThread.join();
onlineThread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
- Finish the executor using the shutdown() method and wait for the finalization of the tasks with the awaitTermination() method:
System.out.printf("Main: Shutting down the executor.
");
executor.shutdown();
try {
executor.awaitTermination(1, TimeUnit.DAYS);
} catch (InterruptedException e) {
e.printStackTrace();
}
- Finish the execution of the ReportSender object setting the value of its end attribute to true:
processor.stopProcessing();
System.out.println("Main: Ends");