Table 14.1 Blocking Queue Methods
You have now seen the low-level building blocks that form the foundations of concurrent programming in Java. However, for practical programming, you want to stay away from the low-level constructs whenever possible. It is much easier and safer to use higher-level structures that have been implemented by concurrency experts.
Many threading problems can be formulated elegantly and safely by using one or more queues. Producer threads insert items into the queue, and consumer threads retrieve them. The queue lets you safely hand over data from one thread to another. For example, consider our bank transfer program. Instead of accessing the bank object directly, the transfer threads insert transfer instruction objects into a queue. Another thread removes the instructions from the queue and carries out the transfers. Only that thread has access to the internals of the bank object. No synchronization is necessary. (Of course, the implementors of the thread-safe queue classes had to worry about locks and conditions, but that was their problem, not yours.)
A blocking queue causes a thread to block when you try to add an element when the queue is currently full or to remove an element when the queue is empty. Blocking queues are a useful tool for coordinating the work of multiple threads. Worker threads can periodically deposit intermediate results into a blocking queue. Other worker threads remove the intermediate results and modify them further. The queue automatically balances the workload. If the first set of threads runs slower than the second, the second set blocks while waiting for the results. If the first set of threads runs faster, the queue fills up until the second set catches up. Table 14.1 shows the methods for blocking queues.
The blocking queue methods fall into three categories that differ by the action they perform when the queue is full or empty. If you use the queue as a thread management tool, use the put and take methods. The add, remove, and element operations throw an exception when you try to add to a full queue or get the head of an empty queue. Of course, in a multithreaded program, the queue might become full or empty at any time, so you will instead want to use the offer, poll, and peek methods. These methods simply return with a failure indicator instead of throwing an exception if they cannot carry out their tasks.
Note
The poll and peek methods return null to indicate failure. Therefore, it is illegal to insert null values into these queues.
There are also variants of the offer and poll methods with a timeout. For example, the call
boolean success = q.offer(x, 100, TimeUnit.MILLISECONDS);
tries for 100 milliseconds to insert an element to the tail of the queue. If it succeeds, it returns true; otherwise, it returns false when it times out. Similarly, the call
Object head = q.poll(100, TimeUnit.MILLISECONDS)
tries for 100 milliseconds to remove the head of the queue. If it succeeds, it returns the head; otherwise, it returns null when it times out.
The put method blocks if the queue is full, and the take method blocks if the queue is empty. These are the equivalents of offer and poll with no timeout.
The java.util.concurrent package supplies several variations of blocking queues. By default, the LinkedBlockingQueue has no upper bound on its capacity, but a maximum capacity can be optionally specified. The LinkedBlockingDeque is a double-ended version. The ArrayBlockingQueue is constructed with a given capacity and an optional parameter to require fairness. If fairness is specified, then the longest-waiting threads are given preferential treatment. As always, fairness exacts a significant performance penalty, and you should only use it if your problem specifically requires it.
The PriorityBlockingQueue is a priority queue, not a first-in/first-out queue. Elements are removed in order of their priority. The queue has unbounded capacity, but retrieval will block if the queue is empty. (See Chapter 9 for more information on priority queues.)
A DelayQueue contains objects that implement the Delayed interface:
interface Delayed extends Comparable<Delayed>
{
long getDelay(TimeUnit unit);
}
The getDelay method returns the remaining delay of the object. A negative value indicates that the delay has elapsed. Elements can only be removed from a DelayQueue if their delay has elapsed. You also need to implement the compareTo method. The DelayQueue uses that method to sort the entries.
Java SE 7 adds a TransferQueue interface that allows a producer thread to wait until a consumer is ready to take on an item. When a producer calls
q.transfer(item);
the call blocks until another thread removes it. The LinkedTransferQueue class implements this interface.
The program in Listing 14.9 shows how to use a blocking queue to control a set of threads. The program searches through all files in a directory and its subdirectories, printing lines that contain a given keyword.
A producer thread enumerates all files in all subdirectories and places them in a blocking queue. This operation is fast, and the queue would quickly fill up with all files in the file system if it was not bounded.
We also start a large number of search threads. Each search thread takes a file from the queue, opens it, prints all lines containing the keyword, and then takes the next file. We use a trick to terminate the application when no further work is required. In order to signal completion, the enumeration thread places a dummy object into the queue. (This is similar to a dummy suitcase with a label “last bag” in a baggage claim belt.) When a search thread takes the dummy, it puts it back and terminates.
Note that no explicit thread synchronization is required. In this application, we use the queue data structure as a synchronization mechanism.
Listing 14.9 blockingQueue/BlockingQueueTest.java
1 package blockingQueue;
2
3 import java.io.*;
4 import java.util.*;
5 import java.util.concurrent.*;
6
7 /**
8 * @version 1.02 2015-06-21
9 * @author Cay Horstmann
10 */
11 public class BlockingQueueTest
12 {
13 private static final int FILE_QUEUE_SIZE = 10;
14 private static final int SEARCH_THREADS = 100;
15 private static final File DUMMY = new File("");
16 private static BlockingQueue<File> queue = new ArrayBlockingQueue<>(FILE_QUEUE_SIZE);
17
18 public static void main(String[] args)
19 {
20 try (Scanner in = new Scanner(System.in))
21 {
22 System.out.print("Enter base directory (e.g. /opt/jdk1.8.0/src): ");
23 String directory = in.nextLine();
24 System.out.print("Enter keyword (e.g. volatile): ");
25 String keyword = in.nextLine();
26
27 Runnable enumerator = () -> {
28 try
29 {
30 enumerate(new File(directory));
31 queue.put(DUMMY);
32 }
33 catch (InterruptedException e)
34 {
35 }
36 };
37
38 new Thread(enumerator).start();
39 for (int i = 1; i <= SEARCH_THREADS; i++) {
40 Runnable searcher = () -> {
41 try
42 {
43 boolean done = false;
44 while (!done)
45 {
46 File file = queue.take();
47 if (file == DUMMY)
48 {
49 queue.put(file);
50 done = true;
51 }
52 else search(file, keyword);
53 }
54 }
55 catch (IOException e)
56 {
57 e.printStackTrace();
58 }
59 catch (InterruptedException e)
60 {
61 }
62 };
63 new Thread(searcher).start();
64 }
65 }
66 }
67
68 /**
69 * Recursively enumerates all files in a given directory and its subdirectories.
70 * @param directory the directory in which to start
71 */
72 public static void enumerate(File directory) throws InterruptedException
73 {
74 File[] files = directory.listFiles();
75 for (File file : files)
76 {
77 if (file.isDirectory()) enumerate(file);
78 else queue.put(file);
79 }
80 }
81
82 /**
83 * Searches a file for a given keyword and prints all matching lines.
84 * @param file the file to search
85 * @param keyword the keyword to search for
86 */
87 public static void search(File file, String keyword) throws IOException
88 {
89 try (Scanner in = new Scanner(file, "UTF-8"))
90 {
91 int lineNumber = 0;
92 while (in.hasNextLine())
93 {
94 lineNumber++;
95 String line = in.nextLine();
96 if (line.contains(keyword))
97 System.out.printf("%s:%d:%s%n", file.getPath(), lineNumber, line);
98 }
99 }
100 }
101 }
java.util.concurrent.PriorityBlockingQueue<E> 5.0
• PriorityBlockingQueue()
• PriorityBlockingQueue(int initialCapacity)
• PriorityBlockingQueue(int initialCapacity, Comparator<? super E> comparator)
constructs an unbounded blocking priority queue implemented as a heap.
If multiple threads concurrently modify a data structure, such as a hash table, it is easy to damage that data structure. (See Chapter 9 for more information on hash tables.) For example, one thread may begin to insert a new element. Suppose it is preempted in the middle of rerouting the links between the hash table’s buckets. If another thread starts traversing the same list, it may follow invalid links and create havoc, perhaps throwing exceptions or being trapped in an infinite loop.
You can protect a shared data structure by supplying a lock, but it is usually easier to choose a thread-safe implementation instead. The blocking queues that we discussed in the preceding section are, of course, thread-safe collections. In the following sections, we discuss the other thread-safe collections that the Java library provides.
The java.util.concurrent package supplies efficient implementations for maps, sorted sets, and queues: ConcurrentHashMap, ConcurrentSkipListMap, ConcurrentSkipListSet, and ConcurrentLinkedQueue.
These collections use sophisticated algorithms that minimize contention by allowing concurrent access to different parts of the data structure.
Unlike most collections, the size method of these classes does not necessarily operate in constant time. Determining the current size of one of these collections usually requires traversal.
Note
Some applications use humongous concurrent hash maps, so large that the size method is insufficient because it returns an int. What is one to do with a map that has over two billion entries? Java SE 8 introduces a mappingCount method that returns the size as a long.
The collections return weakly consistent iterators. That means that the iterators may or may not reflect all modifications that are made after they were constructed, but they will not return a value twice and they will not throw a ConcurrentModificationException.
Note
In contrast, an iterator of a collection in the java.util package throws a ConcurrentModificationException when the collection has been modified after construction of the iterator.
The concurrent hash map can efficiently support a large number of readers and a fixed number of writers. By default, it is assumed that there are up to 16 simultaneous writer threads. There can be many more writer threads, but if more than 16 write at the same time, the others are temporarily blocked. You can specify a higher number in the constructor, but it is unlikely that you will need to.
Note
A hash map keeps all entries with the same hash code in the same “bucket.” Some applications use poor hash functions, and as a result all entries end up in a small number of buckets, severely degrading performance. Even generally reasonable hash functions, such as that of the String class, can be problematic. For example, an attacker can slow down a program by crafting a large number of strings that hash to the same value. As of Java SE 8, the concurrent hash map organizes the buckets as trees, not lists, when the key type implements Comparable, guaranteeing O(log(n)) performance.
java.util.concurrent.ConcurrentHashMap<K, V> 5.0
java.util.concurrent.ConcurrentSkipListMap<K, V> 6
• ConcurrentHashMap<K, V>()
• ConcurrentHashMap<K, V>(int initialCapacity)
• ConcurrentHashMap<K, V>(int initialCapacity, float loadFactor, int concurrencyLevel)
constructs a hash map that can be safely accessed by multiple threads.
• ConcurrentSkipListMap<K, V>()
• ConcurrentSkipListSet<K, V>(Comparator<? super K> comp)
constructs a sorted map that can be safely accessed by multiple threads. The first constructor requires that the keys implement the Comparable interface.
The original version of ConcurrentHashMap only had a few methods for atomic updates, which made for somewhat awkward programming. Suppose we want to count how often certain features are observed. As a simple example, suppose multiple threads encounter words, and we want to count their frequencies.
Can we use a ConcurrentHashMap<String, Long>? Consider the code for incrementing a count. Obviously, the following is not thread safe:
Long oldValue = map.get(word);
Long newValue = oldValue == null ? 1 : oldValue + 1;
map.put(word, newValue); // Error--might not replace oldValue
Another thread might be updating the exact same count at the same time.
Note
Some programmers are surprised that a supposedly thread-safe data structure permits operations that are not thread safe. But there are two entirely different considerations. If multiple threads modify a plain HashMap, they can destroy the internal structure (an array of linked lists). Some of the links may go missing, or even go in circles, rendering the data structure unusable. That will never happen with a ConcurrentHashMap. In the example above, the code for get and put will never corrupt the data structure. But, since the sequence of operations is not atomic, the result is not predictable.
A classic trick is to use the replace operation, which atomically replaces an old value with a new one, provided that no other thread has come before and replaced the old value with something else. You have to keep doing it until replace succeeds:
do {
oldValue = map.get(word);
newValue = oldValue == null ? 1 : oldValue + 1;
} while (!map.replace(word, oldValue, newValue));
Alternatively, you can use a ConcurrentHashMap<String, AtomicLong> or, with Java SE 8, a ConcurrentHashMap<String, LongAdder>. Then the update code is:
map.putIfAbsent(word, new LongAdder());
map.get(word).increment();
The first statement ensures that there is a LongAdder present that we can increment atomically. Since putIfAbsent returns the mapped value (either the existing one or the newly put one), you can combine the two statements:
map.putIfAbsent(word, new LongAdder()).increment();
Java SE 8 provides methods that make atomic updates more convenient. The compute method is called with a key and a function to compute the new value. That function receives the key and the associated value, or null if there is none, and it computes the new value. For example, here is how we can update a map of integer counters:
map.compute(word, (k, v) -> v == null ? 1 : v + 1);
You cannot have null values in a ConcurrentHashMap. There are many methods that use a null value as an indication that a given key is not present in the map.
There are also variants computeIfPresent and computeIfAbsent that only compute a new value when there is already an old one, or when there isn’t yet one. A map of LongAdder counters can be updated with
map.computeIfAbsent(word, k -> new LongAdder()).increment();
That is almost like the call to putIfAbsent that you saw before, but the LongAdder constructor is only called when a new counter is actually needed.
You often need to do something special when a key is added for the first time. The merge method makes this particularly convenient. It has a parameter for the initial value that is used when the key is not yet present. Otherwise, the function that you supplied is called, combining the existing value and the initial value. (Unlike compute, the function does not process the key.)
map.merge(word, 1L, (existingValue, newValue) -> existingValue + newValue);
or, more simply,
map.merge(word, 1L, Long::sum);
It doesn’t get more concise than that.
Note
If the function that is passed to compute or merge returns null, the existing entry is removed from the map.
Caution
When you use compute or merge, keep in mind that the function that you supply should not do a lot of work. While that function runs, some other updates to the map may be blocked. Of course, that function should also not update other parts of the map.
Java SE 8 provides bulk operations on concurrent hash maps that can safely execute even while other threads operate on the map. The bulk operations traverse the map and operate on the elements they find as they go along. No effort is made to freeze a snapshot of the map in time. Unless you happen to know that the map is not being modified while a bulk operation runs, you should treat its result as an approximation of the map’s state.
There are three kinds of operations:
• search applies a function to each key and/or value, until the function yields a non-null result. Then the search terminates and the function’s result is returned.
• reduce combines all keys and/or values, using a provided accumulation function.
• forEach applies a function to all keys and/or values.
Each operation has four versions:
• operationKeys: operates on keys.
• operationValues: operates on values.
• operation: operates on keys and values.
• operationEntries: operates on Map.Entry objects.
With each of the operations, you need to specify a parallelism threshold. If the map contains more elements than the threshold, the bulk operation is parallelized. If you want the bulk operation to run in a single thread, use a threshold of Long.MAX_VALUE. If you want the maximum number of threads to be made available for the bulk operation, use a threshold of 1.
Let’s look at the search methods first. Here are the versions:
U searchKeys(long threshold, BiFunction<? super K, ? extends U> f)
U searchValues(long threshold, BiFunction<? super V, ? extends U> f)
U search(long threshold, BiFunction<? super K, ? super V,? extends U> f)
U searchEntries(long threshold, BiFunction<Map.Entry<K, V>, ? extends U> f)
For example, suppose we want to find the first word that occurs more than 1,000 times. We need to search keys and values:
String result = map.search(threshold, (k, v) -> v > 1000 ? k : null);
Then result is set to the first match, or to null if the search function returns null for all inputs.
The forEach methods have two variants. The first one simply applies a consumer function for each map entry, for example
map.forEach(threshold,
(k, v) -> System.out.println(k + " -> " + v));
The second variant takes an additional transformer function, which is applied first, and its result is passed to the consumer:
map.forEach(threshold,
(k, v) -> k + " -> " + v, // Transformer
System.out::println); // Consumer
The transformer can be used as a filter. Whenever the transformer returns null, the value is silently skipped. For example, here we only print the entries with large values:
map.forEach(threshold,
(k, v) -> v > 1000 ? k + " -> " + v : null, // Filter and transformer
System.out::println); // The nulls are not passed to the consumer
The reduce operations combine their inputs with an accumulation function. For example, here is how you can compute the sum of all values:
Long sum = map.reduceValues(threshold, Long::sum);
As with forEach, you can also supply a transformer function. Here we compute the length of the longest key:
Integer maxlength = map.reduceKeys(threshold,
String::length, // Transformer
Integer::max); // Accumulator
The transformer can act as a filter, by returning null to exclude unwanted inputs. Here, we count how many entries have value > 1000:
Long count = map.reduceValues(threshold,
v -> v > 1000 ? 1L : null,
Long::sum);
Note
If the map is empty, or all entries have been filtered out, the reduce operation returns null. If there is only one element, its transformation is returned, and the accumulator is not applied.
There are specializations for int, long, and double outputs with suffixes ToInt, ToLong, and ToDouble. You need to transform the input to a primitive value and specify a default value and an accumulator function. The default value is returned when the map is empty.
long sum = map.reduceValuesToLong(threshold,
Long::longValue, // Transformer to primitive type
0, // Default value for empty map
Long::sum); // Primitive type accumulator
Caution
These specializations act differently from the object versions where there is only one element to be considered. Instead of returning the transformed element, it is accumulated with the default. Therefore, the default must be the neutral element of the accumulator.
Suppose you want a large, thread-safe set instead of a map. There is no ConcurrentHashSet class, and you know better than trying to create your own. Of course, you can use a ConcurrentHashMap with bogus values, but then you get a map, not a set, and you can’t apply operations of the Set interface.
The static newKeySet method yields a Set<K> that is actually a wrapper around a ConcurrentHashMap<K, Boolean>. (All map values are Boolean.TRUE, but you don’t actually care since you just use it as a set.)
Set<String> words = ConcurrentHashMap.<String>newKeySet();
Of course, if you have an existing map, the keySet method yields the set of keys. That set is mutable. If you remove the set’s elements, the keys (and their values) are removed from the map. But it doesn’t make sense to add elements to the key set, because there would be no corresponding values to add. Java SE 8 adds a second keySet method to ConcurrentHashMap, with a default value, to be used when adding elements to the set:
Set<String> words = map.keySet(1L);
words.add("Java");
If "Java" wasn’t already present in words, it now has a value of one.
The CopyOnWriteArrayList and CopyOnWriteArraySet are thread-safe collections in which all mutators make a copy of the underlying array. This arrangement is useful if the threads that iterate over the collection greatly outnumber the threads that mutate it. When you construct an iterator, it contains a reference to the current array. If the array is later mutated, the iterator still has the old array, but the collection’s array is replaced. As a consequence, the older iterator has a consistent (but potentially outdated) view that it can access without any synchronization expense.
As of Java SE 8, the Arrays class has a number of parallelized operations. The static Arrays.parallelSort method can sort an array of primitive values or objects. For example,
String contents = new String(Files.readAllBytes(
Paths.get("alice.txt")), StandardCharsets.UTF_8); // Read file into string
String[] words = contents.split("[\\P{L}]+"); // Split along nonletters
Arrays.parallelSort(words);
When you sort objects, you can supply a Comparator.
Arrays.parallelSort(words, Comparator.comparing(String::length));
With all methods, you can supply the bounds of a range, such as
values.parallelSort(values.length / 2, values.length); // Sort the upper half
Note
At first glance, it seems a bit odd that these methods have parallel in their name, since the user shouldn’t care how the sorting happens. However, the API designers wanted to make it clear that the sorting is parallelized. That way, users are on notice to avoid comparators with side effects.
The parallelSetAll method fills an array with values that are computed from a function. The function receives the element index and computes the value at that location.
Arrays.parallelSetAll(values, i -> i % 10);
// Fills values with 0 1 2 3 4 5 6 7 8 9 0 1 2 ...
Clearly, this operation benefits from being parallelized. There are versions for all primitive type arrays and for object arrays.
Finally, there is a parallelPrefix method that replaces each array element with the accumulation of the prefix for a given associative operation. Huh? Here is an example. Consider the array [1, 2, 3, 4, ...] and the × operation. After executing Arrays.parallelPrefix(values, (x, y) -> x * y), the array contains
[1, 1 × 2, 1 × 2 × 3, 1 × 2 × 3 × 4, ...]
Perhaps surprisingly, this computation can be parallelized. First, join neighboring elements, as indicated here:
[1, 1 × 2, 3, 3 × 4, 5, 5 × 6, 7, 7 × 8]
The gray values are left alone. Clearly, one can make this computation in parallel in separate regions of the array. In the next step, update the indicated elements by multiplying them with elements that are one or two positions below:
[1, 1 × 2, 1 × 2 × 3, 1 × 2 × 3 × 4, 5, 5 × 6, 5 × 6 × 7, 5 × 6 × 7 × 8]
This can again be done in parallel. After log(n) steps, the process is complete. This is a win over the straightforward linear computation if sufficient processors are available. On special-purpose hardware, this algorithm is commonly used, and users of such hardware are quite ingenious in adapting it to a variety of problems.
Ever since the initial release of Java, the Vector and Hashtable classes provided thread-safe implementations of a dynamic array and a hash table. These classes are now considered obsolete, having been replaced by the ArrayList and HashMap classes. Those classes are not thread safe. Instead, a different mechanism is supplied in the collections library. Any collection class can be made thread safe by means of a synchronization wrapper:
List<E> synchArrayList = Collections.synchronizedList(new ArrayList<E>());
Map<K, V> synchHashMap = Collections.synchronizedMap(new HashMap<K, V>());
The methods of the resulting collections are protected by a lock, providing thread safe access.
You should make sure that no thread accesses the data structure through the original unsynchronized methods. The easiest way to ensure this is not to save any reference to the original object. Simply construct a collection and immediately pass it to the wrapper, as we did in our examples.
You still need to use “client-side” locking if you want to iterate over the collection while another thread has the opportunity to mutate it:
synchronized (synchHashMap)
{
Iterator<K> iter = synchHashMap.keySet().iterator();
while (iter.hasNext()) . . .;
}
You must use the same code if you use a “for each” loop because the loop uses an iterator. Note that the iterator actually fails with a ConcurrentModificationException if another thread mutates the collection while the iteration is in progress. The synchronization is still required so that the concurrent modification can be reliably detected.
You are usually better off using the collections defined in the java.util.concurrent package instead of the synchronization wrappers. In particular, the ConcurrentHashMap map has been carefully implemented so that multiple threads can access it without blocking each other, provided they access different buckets. One exception is an array list that is frequently mutated. In that case, a synchronized ArrayList can outperform a CopyOnWriteArrayList.
A Runnable encapsulates a task that runs asynchronously; you can think of it as an asynchronous method with no parameters and no return value. A Callable is similar to a Runnable, but it returns a value. The Callable interface is a parameterized type, with a single method call.
public interface Callable<V>
{
V call() throws Exception;
}
The type parameter is the type of the returned value. For example, a Callable<Integer> represents an asynchronous computation that eventually returns an Integer object.
A Future holds the result of an asynchronous computation. You can start a computation, give someone the Future object, and forget about it. The owner of the Future object can obtain the result when it is ready.
The Future interface has the following methods:
public interface Future<V>
{
V get() throws . . .;
V get(long timeout, TimeUnit unit) throws . . .;
void cancel(boolean mayInterrupt);
boolean isCancelled();
boolean isDone();
}
A call to the first get method blocks until the computation is finished. The second method throws a TimeoutException if the call timed out before the computation finished. If the thread running the computation is interrupted, both methods throw an InterruptedException. If the computation has already finished, get returns immediately.
The isDone method returns false if the computation is still in progress, true if it is finished.
You can cancel the computation with the cancel method. If the computation has not yet started, it is canceled and will never start. If the computation is currently in progress, it is interrupted if the mayInterrupt parameter is true.
The FutureTask wrapper is a convenient mechanism for turning a Callable into both a Future and a Runnable—it implements both interfaces. For example:
Callable<Integer> myComputation = . . .;
FutureTask<Integer> task = new FutureTask<Integer>(myComputation);
Thread t = new Thread(task); // it's a Runnable
t.start();
...
Integer result = task.get(); // it's a Future
The program in Listing 14.10 puts these concepts to work. This program is similar to the preceding example that found files containing a given keyword. However, now we will merely count the number of matching files. Thus, we have a long-running task that yields an integer value—an example of a Callable<Integer>.
class MatchCounter implements Callable<Integer>
{
public MatchCounter(File directory, String keyword) { . . . }
public Integer call() { . . . } // returns the number of matching files
}
Then we construct a FutureTask object from the MatchCounter and use it to start a thread.
FutureTask<Integer> task = new FutureTask<Integer>(counter);
Thread t = new Thread(task);
t.start();
Finally, we print the result.
System.out.println(task.get() + " matching files.");
Of course, the call to get blocks until the result is actually available.
Inside the call method, we use the same mechanism recursively. For each subdirectory, we produce a new MatchCounter and launch a thread for it. We also stash the FutureTask objects away in an ArrayList<Future<Integer>>. At the end, we add up all results:
for (Future<Integer> result : results)
count += result.get();
Each call to get blocks until the result is available. Of course, the threads run in parallel, so there is a good chance that the results will all be available at about the same time.
Listing 14.10 future/FutureTest.java
1 package future;
2
3 import java.io.*;
4 import java.util.*;
5 import java.util.concurrent.*;
6
7 /**
8 * @version 1.01 2012-01-26
9 * @author Cay Horstmann
10 */
11 public class FutureTest
12 {
13 public static void main(String[] args)
14 {
15 try (Scanner in = new Scanner(System.in))
16 {
17 System.out.print("Enter base directory (e.g. /usr/local/jdk5.0/src): ");
18 String directory = in.nextLine();
19 System.out.print("Enter keyword (e.g. volatile): ");
20 String keyword = in.nextLine();
21
22 MatchCounter counter = new MatchCounter(new File(directory), keyword);
23 FutureTask<Integer> task = new FutureTask<>(counter);
24 Thread t = new Thread(task);
25 t.start();
26 try
27 {
28 System.out.println(task.get() + " matching files.");
29 }
30 catch (ExecutionException e)
31 {
32 e.printStackTrace();
33 }
34 catch (InterruptedException e)
35 {
36 }
37 }
38 }
39 }
40
41 /**
42 * This task counts the files in a directory and its subdirectories that contain a given keyword.
43 */
44 class MatchCounter implements Callable<Integer>
45 {
46 private File directory;
47 private String keyword;
48
49 /**
50 * Constructs a MatchCounter.
51 * @param directory the directory in which to start the search
52 * @param keyword the keyword to look for
53 */
54 public MatchCounter(File directory, String keyword)
55 {
56 this.directory = directory;
57 this.keyword = keyword;
58 }
59
60 public Integer call()
61 {
62 int count = 0;
63 try
64 {
65 File[] files = directory.listFiles();
66 List<Future<Integer>> results = new ArrayList<>();
67
68 for (File file : files)
69 if (file.isDirectory())
70 {
71 MatchCounter counter = new MatchCounter(file, keyword);
72 FutureTask<Integer> task = new FutureTask<>(counter);
73 results.add(task);
74 Thread t = new Thread(task);
75 t.start();
76 }
77 else
78 {
79 if (search(file)) count++;
80 }
81
82 for (Future<Integer> result : results)
83 try
84 {
85 count += result.get();
86 }
87 catch (ExecutionException e)
88 {
89 e.printStackTrace();
90 }
91 }
92 catch (InterruptedException e)
93 {
94 }
95 return count;
96 }
97
98 /**
99 * Searches a file for a given keyword.
100 * @param file the file to search
101 * @return true if the keyword is contained in the file
102 */
103 public boolean search(File file)
104 {
105 try
106 {
107 try (Scanner in = new Scanner(file, "UTF-8"))
108 {
109 boolean found = false;
110 while (!found && in.hasNextLine())
111 {
112 String line = in.nextLine();
113 if (line.contains(keyword)) found = true;
114 }
115 return found;
116 }
117 }
118 catch (IOException e)
119 {
120 return false;
121 }
122 }
123 }
Constructing a new thread is somewhat expensive because it involves interaction with the operating system. If your program creates a large number of short-lived threads, it should use a thread pool instead. A thread pool contains a number of idle threads that are ready to run. You give a Runnable to the pool, and one of the threads calls the run method. When the run method exits, the thread doesn’t die but stays around to serve the next request.
Another reason to use a thread pool is to throttle the number of concurrent threads. Creating a huge number of threads can greatly degrade performance and even crash the virtual machine. If you have an algorithm that creates lots of threads, you should use a “fixed” thread pool that bounds the total number of concurrent threads.
The Executors class has a number of static factory methods for constructing thread pools; see Table 14.2 for a summary.
Let us look at the first three methods in Table 14.2 (we will discuss the remaining methods in Section 14.9.2, “Scheduled Execution,” on p. 926). The newCachedThreadPool method constructs a thread pool that executes each task immediately, using an existing idle thread when available and creating a new thread otherwise. The newFixedThreadPool method constructs a thread pool with a fixed size. If more tasks are submitted than there are idle threads, the unserved tasks are placed on a queue. They are run when other tasks have completed. The newSingleThreadExecutor is a degenerate pool of size 1 where a single thread executes the submitted tasks, one after another. These three methods return an object of the ThreadPoolExecutor class that implements the ExecutorService interface.
You can submit a Runnable or Callable to an ExecutorService with one of the following methods:
Future<?> submit(Runnable task)
Future<T> submit(Runnable task, T result)
Future<T> submit(Callable<T> task)
The pool will run the submitted task at its earliest convenience. When you call submit, you get back a Future object that you can use to query the state of the task.
The first submit method returns an odd-looking Future<?>. You can use such an object to call isDone, cancel, or isCancelled, but the get method simply returns null upon completion.
The second version of submit also submits a Runnable, and the get method of the Future returns the given result object upon completion.
The third version submits a Callable, and the returned Future gets the result of the computation when it is ready.
When you are done with a thread pool, call shutdown. This method initiates the shutdown sequence for the pool. An executor that is shut down accepts no new tasks. When all tasks are finished, the threads in the pool die. Alternatively, you can call shutdownNow. The pool then cancels all tasks that have not yet begun and attempts to interrupt the running threads.
Here, in summary, is what you do to use a thread pool:
1. Call the static newCachedThreadPool or newFixedThreadPool method of the Executors class.
2. Call submit to submit Runnable or Callable objects.
3. If you want to be able to cancel a task, or if you submit Callable objects, hang on to the returned Future objects.
4. Call shutdown when you no longer want to submit any tasks.
For example, the preceding example program produced a large number of short-lived threads, one per directory. The program in Listing 14.11 uses a thread pool to launch the tasks instead.
For informational purposes, this program prints out the largest pool size during execution. This information is not available through the ExecutorService interface. For that reason, we had to cast the pool object to the ThreadPoolExecutor class.
Listing 14.11 threadPool/ThreadPoolTest.java
1 package threadPool;
2
3 import java.io.*;
4 import java.util.*;
5 import java.util.concurrent.*;
6
7 /**
8 * @version 1.02 2015-06-21
9 * @author Cay Horstmann
10 */
11 public class ThreadPoolTest
12 {
13 public static void main(String[] args) throws Exception
14 {
15 try (Scanner in = new Scanner(System.in))
16 {
17 System.out.print("Enter base directory (e.g. /usr/local/jdk5.0/src): ");
18 String directory = in.nextLine();
19 System.out.print("Enter keyword (e.g. volatile): ");
20 String keyword = in.nextLine();
21
22 ExecutorService pool = Executors.newCachedThreadPool();
23
24 MatchCounter counter = new MatchCounter(new File(directory), keyword, pool);
25 Future<Integer> result = pool.submit(counter);
26
27 try
28 {
29 System.out.println(result.get() + " matching files.");
30 }
31 catch (ExecutionException e)
32 {
33 e.printStackTrace();
34 }
35 catch (InterruptedException e)
36 {
37 }
38 pool.shutdown();
39
40 int largestPoolSize = ((ThreadPoolExecutor) pool).getLargestPoolSize();
41 System.out.println("largest pool size=" + largestPoolSize);
42 }
43 }
44 }
45
46 /**
47 * This task counts the files in a directory and its subdirectories that contain a given keyword.
48 */
49 class MatchCounter implements Callable<Integer>
50 {
51 private File directory;
52 private String keyword;
53 private ExecutorService pool;
54 private int count;
55
56 /**
57 * Constructs a MatchCounter.
58 * @param directory the directory in which to start the search
59 * @param keyword the keyword to look for
60 * @param pool the thread pool for submitting subtasks
61 */
62 public MatchCounter(File directory, String keyword, ExecutorService pool)
63 {
64 this.directory = directory;
65 this.keyword = keyword;
66 this.pool = pool;
67 }
68
69 public Integer call()
70 {
71 count = 0;
72 try
73 {
74 File[] files = directory.listFiles();
75 List<Future<Integer>> results = new ArrayList<>();
76
77 for (File file : files)
78 if (file.isDirectory())
79 {
80 MatchCounter counter = new MatchCounter(file, keyword, pool);
81 Future<Integer> result = pool.submit(counter);
82 results.add(result);
83 }
84 else
85 {
86 if (search(file)) count++;
87 }
88
89 for (Future<Integer> result : results)
90 try
91 {
92 count += result.get();
93 }
94 catch (ExecutionException e)
95 {
96 e.printStackTrace();
97 }
98 }
99 catch (InterruptedException e)
100 {
101 }
102 return count;
103 }
104
105 /**
106 * Searches a file for a given keyword.
107 * @param file the file to search
108 * @return true if the keyword is contained in the file
109 */
110 public boolean search(File file)
111 {
112 try
113 {
114 try (Scanner in = new Scanner(file, "UTF-8"))
115 {
116 boolean found = false;
117 while (!found && in.hasNextLine())
118 {
119 String line = in.nextLine();
120 if (line.contains(keyword)) found = true;
121 }
122 return found;
123 }
124 }
125 catch (IOException e)
126 {
127 return false;
128 }
129 }
130 }
The ScheduledExecutorService interface has methods for scheduled or repeated execution of tasks. It is a generalization of java.util.Timer that allows for thread pooling. The newScheduledThreadPool and newSingleThreadScheduledExecutor methods of the Executors class return objects that implement the ScheduledExecutorService interface.
You can schedule a Runnable or Callable to run once, after an initial delay. You can also schedule a Runnable to run periodically. See the API notes for details.
You have seen how to use an executor service as a thread pool to increase the efficiency of task execution. Sometimes, an executor is used for a more tactical reason, simply to control a group of related tasks. For example, you can cancel all tasks in an executor with the shutdownNow method.
The invokeAny method submits all objects in a collection of Callable objects and returns the result of a completed task. You don’t know which task that is—presumably, it is the one that finished most quickly. Use this method for a search problem in which you are willing to accept any solution. For example, suppose that you need to factor a large integer—a computation that is required for breaking the RSA cipher. You could submit a number of tasks, each attempting a factorization with numbers in a different range. As soon as one of these tasks has an answer, your computation can stop.
The invokeAll method submits all objects in a collection of Callable objects, blocks until all of them complete, and returns a list of Future objects that represent the solutions to all tasks. You can process the results of the computation when they are available, like this:
List<Callable<T>> tasks = . . .;
List<Future<T>> results = executor.invokeAll(tasks);
for (Future<T> result : results)
processFurther(result.get());
A disadvantage of this approach is that you may wait needlessly if the first task happens to take a long time. It would make more sense to obtain the results in the order in which they are available. This can be arranged with the ExecutorCompletionService.
Start with an executor, obtained in the usual way. Then construct an ExecutorCompletionService. Submit tasks to the completion service. The service manages a blocking queue of Future objects, containing the results of the submitted tasks as they become available. Thus, a more efficient organization for the preceding computation is the following:
ExecutorCompletionService<T> service = new ExecutorCompletionService<>(executor);
for (Callable<T> task : tasks) service.submit(task);
for (int i = 0; i < tasks.size(); i++)
processFurther(service.take().get());
Some applications use a large number of threads that are mostly idle. An example would be a web server that uses one thread per connection. Other applications use one thread per processor core, in order to carry out computationally intensive tasks, such as image or video processing. The fork-join framework, which appeared in Java SE 7, is designed to support the latter. Suppose you have a processing task that naturally decomposes into subtasks, like this:
if (problemSize < threshold)
solve problem directly
else
{
break problem into subproblems
recursively solve each subproblem
combine the results
}
One example is image processing. To enhance an image, you can transform the top half and the bottom half. If you have enough idle processors, those operations can run in parallel. (You will need to do a bit of extra work along the strip that separates the two halves, but that’s a technical detail.)
Here, we will discuss a simpler example. Suppose we want to count how many elements of an array fulfill a particular property. We cut the array in half, compute the counts of each half, and add them up.
To put the recursive computation in a form that is usable by the framework, supply a class that extends RecursiveTask<T> (if the computation produces a result of type T) or RecursiveAction (if it doesn’t produce a result). Override the compute method to generate and invoke subtasks, and to combine their results.
class Counter extends RecursiveTask<Integer>
{
...
protected Integer compute()
{
if (to - from < THRESHOLD)
{
solve problem directly
}
else
{
int mid = (from + to) / 2;
Counter first = new Counter(values, from, mid, filter);
Counter second = new Counter(values, mid, to, filter);
invokeAll(first, second);
return first.join() + second.join();
}
}
}
Here, the invokeAll method receives a number of tasks and blocks until all of them have completed. The join method yields the result. Here, we apply join to each subtask and return the sum.
Note
There is also a get method for getting the current result, but it is less attractive since it can throw checked exceptions that we are not allowed to throw in the compute method.
Listing 14.12 shows the complete example.
Behind the scenes, the fork-join framework uses an effective heuristic for balancing the workload among available threads, called work stealing. Each worker thread has a deque (double-ended queue) for tasks. A worker thread pushes subtasks onto the head of its own deque. (Only one thread accesses the head, so no locking is required.) When a worker thread is idle, it “steals” a task from the tail of another deque. Since large subtasks are at the tail, such stealing is rare.
Listing 14.12 forkJoin/ForkJoinTest.java
1 package forkJoin;
2
3 import java.util.concurrent.*;
4 import java.util.function.*;
5
6 /**
7 * This program demonstrates the fork-join framework.
8 * @version 1.01 2015-06-21
9 * @author Cay Horstmann
10 */
11 public class ForkJoinTest
12 {
13 public static void main(String[] args)
14 {
15 final int SIZE = 10000000;
16 double[] numbers = new double[SIZE];
17 for (int i = 0; i < SIZE; i++) numbers[i] = Math.random();
18 Counter counter = new Counter(numbers, 0, numbers.length, x -> x > 0.5);
19 ForkJoinPool pool = new ForkJoinPool();
20 pool.invoke(counter);
21 System.out.println(counter.join());
22 }
23 }
24
25 class Counter extends RecursiveTask<Integer>
26 {
27 public static final int THRESHOLD = 1000;
28 private double[] values;
29 private int from;
30 private int to;
31 private DoublePredicate filter;
32
33 public Counter(double[] values, int from, int to, DoublePredicate filter)
34 {
35 this.values = values;
36 this.from = from;
37 this.to = to;
38 this.filter = filter;
39 }
40
41 protected Integer compute()
42 {
43 if (to - from < THRESHOLD)
44 {
45 int count = 0;
46 for (int i = from; i < to; i++)
47 {
48 if (filter.test(values[i])) count++;
49 }
50 return count;
51 }
52 else
53 {
54 int mid = (from + to) / 2;
55 Counter first = new Counter(values, from, mid, filter);
56 Counter second = new Counter(values, mid, to, filter);
57 invokeAll(first, second);
58 return first.join() + second.join();
59 }
60 }
61 }
The traditional approach for dealing with nonblocking calls is to use event handlers, where the programmer registers a handler for the action that should occur after a task completes. Of course, if the next action is also asynchronous, the next action after that is in a different event handler. Even though the programmer thinks in terms of “first do step 1, then step 2, then step 3,” the program logic becomes dispersed in different handlers. It gets worse when one has to add error handling. Suppose step 2 is “the user logs in.” You may need to repeat that step since the user can mistype the credentials. Trying to implement such a control flow in a set of event handlers, or to understand it once it has been implemented, is challenging.
The CompletableFuture class of Java SE 8 provides an alternative approach. Unlike event handlers, completable futures can be composed.
For example, suppose we want to extract all links from a web page in order to build a web crawler. Let’s say we have a method
public void CompletableFuture<String> readPage(URL url)
that yields the text of a web page when it becomes available. If the method
public static List<URL> getLinks(String page)
yields the URLs in an HTML page, you can schedule it to be called when the page is available:
CompletableFuture<String> contents = readPage(url);
CompletableFuture<List<URL>> links = contents.thenApply(Parser::getLinks);
The thenApply method doesn’t block either. It returns another future. When the first future has completed, its result is fed to the getLinks method, and the return value of that method becomes the final result.
With completable futures, you just specify what you want to have done and in which order. It won’t all happen right away, of course, but what is important is that all the code is in one place.
Conceptually, CompletableFuture is a simple API, but there are many variants of methods for composing completable futures. Let us first look at those that deal with a single future (see Table 14.3). (For each method shown, there are also two Async variants that I don’t show. One of them uses a shared ForkJoinPool, and the other has an Executor parameter.) In the table, I use a shorthand notation for the ponderous functional interfaces, writing T -> U instead of Function<? super T, U>. These aren’t actual Java types, of course.
You have already seen the thenApply method. The calls
CompletableFuture<U> future.thenApply(f);
CompletableFuture<U> future.thenApplyAsync(f);
return a future that applies f to the result of future when it is available. The second call runs f in yet another thread.
The thenCompose method, instead of taking a function T -> U, takes a function T -> CompletableFuture<U>. That sounds rather abstract, but it can be quite natural. Consider the action of reading a web page from a given URL. Instead of supplying a method
public String blockingReadPage(URL url)
it is more elegant to have that method return a future:
public CompletableFuture<String> readPage(URL url)
Now, suppose we have another method that gets the URL from user input, perhaps from a dialog that won’t reveal the answer until the user has clicked the OK button. That, too, is an event in the future:
public CompletableFuture<URL> getURLInput(String prompt)
Here we have two functions T -> CompletableFuture<U> and U -> CompletableFuture<V>. Clearly, they compose to a function T -> CompletableFuture<V> if the second function is called when the first one has completed. That is exactly what thenCompose does.
The third method in Table 14.3 focuses on a different aspect that I have ignored so far: failure. When an exception is thrown in a CompletableFuture, it is captured and wrapped in an unchecked ExecutionException when the get method is called. But perhaps get is never called. In order to handle an exception, use the handle method. The supplied function is called with the result (or null if none) and the exception (or null if none), and it gets to make sense of the situation.
The remaining methods have void result and are normally used at the end of a processing pipeline.
Now let us turn to methods that combine multiple futures (see Table 14.4).
The first three methods run a CompletableFuture<T> and a CompletableFuture<U> action in parallel and combine the results.
The next three methods run two CompletableFuture<T> actions in parallel. As soon as one of them finishes, its result is passed on, and the other result is ignored.
Finally, the static allOf and anyOf methods take a variable number of completable futures and yield a CompletableFuture<Void> that completes when all of them, or any one of them, completes. No results are propagated.
Technically speaking, the methods in this section accept parameters of type CompletionStage, not CompletableFuture. That is an interface with almost forty abstract methods, implemented only by CompletableFuture. The interface is provided so that third-party frameworks can implement it.
The java.util.concurrent package contains several classes that help manage a set of collaborating threads—see Table 14.5. These mechanisms have “canned functionality” for common rendezvous patterns between threads. If you have a set of collaborating threads that follow one of these behavior patterns, you should simply reuse the appropriate library class instead of trying to come up with a handcrafted collection of locks and conditions.
Conceptually, a semaphore manages a number of permits. The number is supplied in the constructor. To proceed past the semaphore, a thread requests a permit by calling acquire. (There are no actual permit objects. The semaphore simply keeps a count.) Since only a fixed number of permits is available, a semaphore limits the number of threads that are allowed to pass. Other threads may issue permits by calling release. Moreover, a permit doesn’t have to be released by the thread that acquires it. Any thread can release any number of permits, potentially increasing the number of permits beyond the initial count.
Semaphores were invented by Edsger Dijkstra in 1968, for use as a synchronization primitive. Dijkstra showed that semaphores can be efficiently implemented and that they are powerful enough to solve many common thread synchronization problems. In just about any operating systems textbook, you will find implementations of bounded queues using semaphores.
Of course, application programmers shouldn’t reinvent bounded queues. Usually, semaphores do not map directly to common application situations.
A CountDownLatch lets a set of threads wait until a count has reached zero. The countdown latch is one-time only. Once the count has reached 0, you cannot increment it again.
A useful special case is a latch with a count of 1. This implements a one-time gate. Threads are held at the gate until another thread sets the count to 0.
Imagine, for example, a set of threads that need some initial data to do their work. The worker threads are started and wait at the gate. Another thread prepares the data. When it is ready, it calls countDown, and all worker threads proceed.
You can then use a second latch to check when all worker threads are done. Initialize the latch with the number of threads. Each worker thread counts down that latch just before it terminates. Another thread that harvests the work results waits on the latch, and proceeds as soon as all workers have terminated.
The CyclicBarrier class implements a rendezvous called a barrier. Consider a number of threads that are working on parts of a computation. When all parts are ready, the results need to be combined. When a thread is done with its part, we let it run against the barrier. Once all threads have reached the barrier, the barrier gives way and the threads can proceed.
Here are the details. First, construct a barrier, giving the number of participating threads:
CyclicBarrier barrier = new CyclicBarrier(nthreads);
Each thread does some work and calls await on the barrier upon completion:
public void run()
{
doWork();
barrier.await();
...
}
The await method takes an optional timeout parameter:
barrier.await(100, TimeUnit.MILLISECONDS);
If any of the threads waiting for the barrier leaves the barrier, then the barrier breaks. (A thread can leave because it called await with a timeout or because it was interrupted.) In that case, the await method for all other threads throws a BrokenBarrierException. Threads that are already waiting have their await call terminated immediately.
You can supply an optional barrier action that is executed when all threads have reached the barrier:
Runnable barrierAction = . . .;
CyclicBarrier barrier = new CyclicBarrier(nthreads, barrierAction);
The action can harvest the results of the individual threads.
The barrier is called cyclic because it can be reused after all waiting threads have been released. In this regard, it differs from a CountDownLatch which can only be used once.
The Phaser class adds more flexibility, allowing you to vary the number of participating threads between phases.
An Exchanger is used when two threads are working on two instances of the same data buffer. Typically, one thread fills the buffer, and the other consumes its contents. When both are done, they exchange their buffers.
A synchronous queue is a mechanism that pairs up producer and consumer threads. When a thread calls put on a SynchronousQueue, it blocks until another thread calls take, and vice versa. Unlike the case with an Exchanger, data are only transferred in one direction, from the producer to the consumer.
Even though the SynchronousQueue class implements the BlockingQueue interface, it is not conceptually a queue. It does not contain any elements—its size method always returns 0.
As we mentioned in the introduction to this chapter, one of the reasons to use threads in your programs is to make your programs more responsive. When your program needs to do something time consuming, you should fire up another worker thread instead of blocking the user interface.
However, you have to be careful what you do in a worker thread because, perhaps surprisingly, Swing is not thread safe. If you try to manipulate user interface elements from multiple threads, your user interface can become corrupted.
To see the problem, run the upcoming test program in Listing 14.13. When you click the Bad button, a new thread is started whose run method tortures a combo box, randomly adding and removing values.
public void run()
{
try
{
while (true)
{
int i = Math.abs(generator.nextInt());
if (i % 2 == 0)
combo.insertItemAt(new Integer(i), 0);
else if (combo.getItemCount() > 0)
combo.removeItemAt(i % combo.getItemCount());
sleep(1);
}
catch (InterruptedException e) {}
}
}
Try it out. Click the Bad button. Click the combo box a few times. Move the scrollbar. Move the window. Click the Bad button again. Keep clicking the combo box. Eventually, you should see an exception report (Figure 14.8).
What is going on? When an element is inserted into the combo box, the combo box fires an event to update the display. Then, the display code springs into action, reading the current size of the combo box and preparing to display the values. But the worker thread keeps going—occasionally resulting in a reduction of the count of the values in the combo box. The display code then thinks that there are more values in the model than there actually are, asks for a nonexistent value, and triggers an ArrayIndexOutOfBounds exception.
This situation could have been avoided if programmers could lock the combo box object while displaying it. However, the designers of Swing decided not to expend any effort to make Swing thread safe, for two reasons. First, synchronization takes time, and nobody wanted to slow down Swing any further. More importantly, the Swing team checked out the experience other teams had with thread-safe user interface toolkits. What they found was not encouraging. Programmers using thread-safe toolkits turned out to be confused by the demands for synchronization and often created deadlock-prone programs.
When you use threads together with Swing, you have to follow two simple rules.
1. If an action takes a long time, do it in a separate worker thread and never in the event dispatch thread.
2. Do not touch Swing components in any thread other than the event dispatch thread.
The reason for the first rule is easy to understand. If you take a long time in the event dispatch thread, the application seems “dead” because it cannot respond to any events. In particular, the event dispatch thread should never make input/output calls, which might block indefinitely, and it should never call sleep. (If you need to wait for a specific amount of time, use timer events.)
The second rule is often called the single-thread rule for Swing programming. We discuss it further on page 951.
These two rules seem to be in conflict with each other. Suppose you fire up a separate thread to run a time-consuming task. You would usually want to update the user interface to indicate progress while your thread is working. When your task is finished, you’d want to update the GUI again. But you can’t touch Swing components from your thread. For example, if you want to update a progress bar or a label text, you can’t simply set its value from your thread.
To solve this problem, you can use, in any thread, two utility methods to add arbitrary actions to the event queue. For example, suppose you want to periodically update a label in a thread to indicate progress. You can’t call label.setText from your thread.
Instead, use the invokeLater and invokeAndWait methods of the EventQueue class to have that call executed in the event dispatching thread.
Here is what you do. Place the Swing code into the run method of a class that implements the Runnable interface. Then, create an object of that class and pass it to the static invokeLater or invokeAndWait method. For example, here is how to update a label text:
EventQueue.invokeLater(() -> {
label.setText(percentage + "% complete");
});
The invokeLater method returns immediately when the event is posted to the event queue. The run method of the Runnable is executed asynchronously. The invokeAndWait method waits until the run method has actually been executed.
For updating a progress label, the invokeLater method is more appropriate. Users would rather have the worker thread make more progress than have the most precise progress indicator.
Both methods execute the run method in the event dispatch thread. No new thread is created.
Listing 14.13 demonstrates how to use the invokeLater method to safely modify the contents of a combo box. If you click the Good button, a thread inserts and removes numbers. However, the actual modification takes place in the event dispatching thread.
Listing 14.13 swing/SwingThreadTest.java
1 package swing;
2
3 import java.awt.*;
4 import java.util.*;
5
6 import javax.swing.*;
7
8 /**
9 * This program demonstrates that a thread that runs in parallel with the event
10 * dispatch thread can cause errors in Swing components.
11 * @version 1.24 2015-06-21
12 * @author Cay Horstmann
13 */
14 public class SwingThreadTest
15 {
16 public static void main(String[] args)
17 {
18 EventQueue.invokeLater(() -> {
19 JFrame frame = new SwingThreadFrame();
20 frame.setTitle("SwingThreadTest");
21 frame.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
22 frame.setVisible(true);
23 });
24 }
25 }
26
27 /**
28 * This frame has two buttons to fill a combo box from a separate thread. The
29 * "Good" button uses the event queue, the "Bad" button modifies the combo box
30 * directly.
31 */
32 class SwingThreadFrame extends JFrame
33 {
34 public SwingThreadFrame()
35 {
36 final JComboBox<Integer> combo = new JComboBox<>();
37 combo.insertItemAt(Integer.MAX_VALUE, 0);
38 combo.setPrototypeDisplayValue(combo.getItemAt(0));
39 combo.setSelectedIndex(0);
40
41 JPanel panel = new JPanel();
42
43 JButton goodButton = new JButton("Good");
44 goodButton.addActionListener(event ->
45 new Thread(new GoodWorkerRunnable(combo)).start());
46 panel.add(goodButton);
47 JButton badButton = new JButton("Bad");
48 badButton.addActionListener(event ->
49 new Thread(new BadWorkerRunnable(combo)).start());
50 panel.add(badButton);
51
52 panel.add(combo);
53 add(panel);
54 pack();
55 }
56 }
57
58 /**
59 * This runnable modifies a combo box by randomly adding and removing numbers.
60 * This can result in errors because the combo box methods are not synchronized
61 * and both the worker thread and the event dispatch thread access the combo
62 * box.
63 */
64 class BadWorkerRunnable implements Runnable
65 {
66 private JComboBox<Integer> combo;
67 private Random generator;
68
69 public BadWorkerRunnable(JComboBox<Integer> aCombo)
70 {
71 combo = aCombo;
72 generator = new Random();
73 }
74
75 public void run()
76 {
77 try
78 {
79 while (true)
80 {
81 int i = Math.abs(generator.nextInt());
82 if (i % 2 == 0)
83 combo.insertItemAt(i, 0);
84 else if (combo.getItemCount() > 0)
85 combo.removeItemAt(i % combo.getItemCount());
86 Thread.sleep(1);
87 }
88 }
89 catch (InterruptedException e)
90 {
91 }
92 }
93 }
94
95 /**
96 * This runnable modifies a combo box by randomly adding and removing numbers.
97 * In order to ensure that the combo box is not corrupted, the editing
98 * operations are forwarded to the event dispatch thread.
99 */
100 class GoodWorkerRunnable implements Runnable
101 {
102 private JComboBox<Integer> combo;
103 private Random generator;
104
105 public GoodWorkerRunnable(JComboBox<Integer> aCombo)
106 {
107 combo = aCombo;
108 generator = new Random();
109 }
110
111 public void run()
112 {
113 try
114 {
115 while (true)
116 {
117 EventQueue.invokeLater(() ->
118 {
119 int i = Math.abs(generator.nextInt());
120 if (i % 2 == 0)
121 combo.insertItemAt(i, 0);
122 else if (combo.getItemCount() > 0)
123 combo.removeItemAt(i % combo.getItemCount());
124 });
125 Thread.sleep(1);
126 }
127 }
128 catch (InterruptedException e)
129 {
130 }
131 }
132 }
When a user issues a command for which processing takes a long time, you will want to fire up a new thread to do the work. As you saw in the preceding section, that thread should use the EventQueue.invokeLater method to update the user interface. The SwingWorker class reduces the tedium of implementing background tasks.
The program in Listing 14.14 has commands for loading a text file and for canceling the file loading process. You should try the program with a long file, such as the full text of The Count of Monte Cristo, supplied in the gutenberg directory of the book’s companion code. The file is loaded in a separate thread. While the file is being read, the Open menu item is disabled and the Cancel item is enabled (see Figure 14.9). After each line is read, a line counter in the status bar is updated. After the reading process is complete, the Open menu item is reenabled, the Cancel item is disabled, and the status line text is set to Done.
This example shows the typical UI activities of a background task:
• After each work unit, update the UI to show progress.
• After the work is finished, make a final change to the UI.
The SwingWorker class makes it easy to implement such a task. Override the doInBackground method to do the time-consuming work and occasionally call publish to communicate work progress. This method is executed in a worker thread. The publish method causes a process method to execute in the event dispatch thread to deal with the progress data. When the work is complete, the done method is called in the event dispatch thread so that you can finish updating the UI.
Whenever you want to do some work in the worker thread, construct a new worker. (Each worker object is meant to be used only once.) Then call the execute method. You will typically call execute on the event dispatch thread, but that is not a requirement.
It is assumed that a worker produces a result of some kind; therefore, SwingWorker<T, V> implements Future<T>. This result can be obtained by the get method of the Future interface. Since the get method blocks until the result is available, you don’t want to call it immediately after calling execute. It is a good idea to call it only when you know that the work has been completed. Typically, you call get from the done method. (There is no requirement to call get. Sometimes, processing the progress data is all you need.)
Both the intermediate progress data and the final result can have arbitrary types. The SwingWorker class has these types as type parameters. A SwingWorker<T, V> produces a result of type T and progress data of type V.
To cancel the work in progress, use the cancel method of the Future interface. When the work is canceled, the get method throws a CancellationException.
As already mentioned, the worker thread’s call to publish will cause calls to process on the event dispatch thread. For efficiency, the results of several calls to publish may be batched up in a single call to process. The process method receives a List<V> containing all intermediate results.
Let us put this mechanism to work for reading in a text file. As it turns out, a JTextArea is quite slow. Appending lines from a long text file (such as all lines in The Count of Monte Cristo) takes considerable time.
To show the user that progress is being made, we want to display the number of lines read in a status line. Thus, the progress data consist of the current line number and the current line of text. We package these into a trivial inner class:
private class ProgressData
{
public int number;
public String line;
}
The final result is the text that has been read into a StringBuilder. Thus, we need a SwingWorker<StringBuilder, ProgressData>.
In the doInBackground method, we read a file, a line at a time. After each line, we call publish to publish the line number and the text of the current line.
@Override public StringBuilder doInBackground() throws IOException, InterruptedException
{
int lineNumber = 0;
Scanner in = new Scanner(new FileInputStream(file), "UTF-8");
while (in.hasNextLine())
{
String line = in.nextLine();
lineNumber++;
text.append(line).append("\n");
ProgressData data = new ProgressData();
data.number = lineNumber;
data.line = line;
publish(data);
Thread.sleep(1); // to test cancellation; no need to do this in your programs
}
return text;
}
We also sleep for a millisecond after every line so that you can test cancellation without getting stressed out, but you wouldn’t want to slow down your own programs by sleeping. If you comment out this line, you will find that The Count of Monte Cristo loads quite quickly, with only a few batched user interface updates.
Note
You can make this program behave quite smoothly by updating the text area from the worker thread, but this is not possible for most Swing components. We show you the general approach in which all component updates occur in the event dispatch thread.
In the process method, we ignore all line numbers but the last one, and we concatenate all lines for a single update of the text area.
@Override public void process(List<ProgressData> data)
{
if (isCancelled()) return;
StringBuilder b = new StringBuilder();
statusLine.setText("" + data.get(data.size() - 1).number);
for (ProgressData d : data) b.append(d.line).append("\n");
textArea.append(b.toString());
}
In the done method, the text area is updated with the complete text, and the Cancel menu item is disabled.
Note how the worker is started in the event listener for the Open menu item.
This simple technique allows you to execute time-consuming tasks while keeping the user interface responsive.
Listing 14.14 swingWorker/SwingWorkerTest.java
1 package swingWorker;
2
3 import java.awt.*;
4 import java.io.*;
5 import java.util.*;
6 import java.util.List;
7 import java.util.concurrent.*;
8
9 import javax.swing.*;
10
11 /**
12 * This program demonstrates a worker thread that runs a potentially time-consuming task.
13 * @version 1.11 2015-06-21
14 * @author Cay Horstmann
15 */
16 public class SwingWorkerTest
17 {
18 public static void main(String[] args) throws Exception
19 {
20 EventQueue.invokeLater(() -> {
21 JFrame frame = new SwingWorkerFrame();
22 frame.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
23 frame.setVisible(true);
24 });
25 }
26 }
27
28 /**
29 * This frame has a text area to show the contents of a text file, a menu to open a file and
30 * cancel the opening process, and a status line to show the file loading progress.
31 */
32 class SwingWorkerFrame extends JFrame
33 {
34 private JFileChooser chooser;
35 private JTextArea textArea;
36 private JLabel statusLine;
37 private JMenuItem openItem;
38 private JMenuItem cancelItem;
39 private SwingWorker<StringBuilder, ProgressData> textReader;
40 public static final int TEXT_ROWS = 20;
41 public static final int TEXT_COLUMNS = 60;
42
43 public SwingWorkerFrame()
44 {
45 chooser = new JFileChooser();
46 chooser.setCurrentDirectory(new File("."));
47
48 textArea = new JTextArea(TEXT_ROWS, TEXT_COLUMNS);
49 add(new JScrollPane(textArea));
50
51 statusLine = new JLabel(" ");
52 add(statusLine, BorderLayout.SOUTH);
53
54 JMenuBar menuBar = new JMenuBar();
55 setJMenuBar(menuBar);
56
57 JMenu menu = new JMenu("File");
58 menuBar.add(menu);
59
60 openItem = new JMenuItem("Open");
61 menu.add(openItem);
62 openItem.addActionListener(event -> {
63 // show file chooser dialog
64 int result = chooser.showOpenDialog(null);
65
66 // if file selected, set it as icon of the label
67 if (result == JFileChooser.APPROVE_OPTION)
68 {
69 textArea.setText("");
70 openItem.setEnabled(false);
71 textReader = new TextReader(chooser.getSelectedFile());
72 textReader.execute();
73 cancelItem.setEnabled(true);
74 }
75 });
76
77 cancelItem = new JMenuItem("Cancel");
78 menu.add(cancelItem);
79 cancelItem.setEnabled(false);
80 cancelItem.addActionListener(event -> textReader.cancel(true));
81 pack();
82 }
83
84 private class ProgressData
85 {
86 public int number;
87 public String line;
88 }
89
90 private class TextReader extends SwingWorker<StringBuilder, ProgressData>
91 {
92 private File file;
93 private StringBuilder text = new StringBuilder();
94
95 public TextReader(File file)
96 {
97 this.file = file;
98 }
99
100 // The following method executes in the worker thread; it doesn't touch Swing components.
101
102 @Override
103 public StringBuilder doInBackground() throws IOException, InterruptedException
104 {
105 int lineNumber = 0;
106 try (Scanner in = new Scanner(new FileInputStream(file), "UTF-8"))
107 {
108 while (in.hasNextLine())
109 {
110 String line = in.nextLine();
111 lineNumber++;
112 text.append(line).append("\n");
113 ProgressData data = new ProgressData();
114 data.number = lineNumber;
115 data.line = line;
116 publish(data);
117 Thread.sleep(1); // to test cancellation; no need to do this in your programs
118 }
119 }
120 return text;
121 }
122
123 // The following methods execute in the event dispatch thread.
124
125 @Override
126 public void process(List<ProgressData> data)
127 {
128 if (isCancelled()) return;
129 StringBuilder b = new StringBuilder();
130 statusLine.setText("" + data.get(data.size() - 1).number);
131 for (ProgressData d : data) b.append(d.line).append("\n");
132 textArea.append(b.toString());
133 }
134
135 @Override
136 public void done()
137 {
138 try
139 {
140 StringBuilder result = get();
141 textArea.setText(result.toString());
142 statusLine.setText("Done");
143 }
144 catch (InterruptedException ex)
145 {
146 }
147 catch (CancellationException ex)
148 {
149 textArea.setText("");
150 statusLine.setText("Cancelled");
151 }
152 catch (ExecutionException ex)
153 {
154 statusLine.setText("" + ex.getCause());
155 }
156
157 cancelItem.setEnabled(false);
158 openItem.setEnabled(true);
159 }
160 };
161 }
Every Java application starts with a main method that runs in the main thread. In a Swing program, the main thread is short lived. It schedules the construction of the user interface in the event dispatch thread and then exits. After the user interface construction, the event dispatch thread processes event notifications, such as calls to actionPerformed or paintComponent. Other threads, such as the thread that posts events into the event queue, are running behind the scenes, but those threads are invisible to the application programmer.
Earlier in the chapter, we introduced the single-thread rule: “Do not touch Swing components in any thread other than the event dispatch thread.” In this section, we investigate that rule further.
There are a few exceptions to the single-thread rule.
• You can safely add and remove event listeners in any thread. Of course, the listener methods will be invoked in the event dispatch thread.
• A small number of Swing methods are thread safe. They are specially marked in the API documentation with the sentence “This method is thread safe, although most Swing methods are not.” The most useful among these thread-safe methods are:
JTextComponent.setText
JTextArea.insert
JTextArea.append
JTextArea.replaceRange
JComponent.repaint
JComponent.revalidate
Note
We used the repaint method many times in this book, but the revalidate method is less common. Its purpose is to force a layout of a component after the contents have changed. The traditional AWT has a validate method to force the layout of a component. For Swing components, you should simply call revalidate instead. (However, to force the layout of a JFrame, you still need to call validate—a JFrame is a Component but not a JComponent.)
Historically, the single-thread rule was more permissive. Any thread was allowed to construct components, set their properties, and add them into containers, as long as none of these components had been realized. A component is realized if it can receive paint or validation events. This is the case after the setVisible(true) or pack (!) methods have been invoked on the component, or after the component has been added to a container that has been realized.
That version of the single-thread rule was convenient. It allowed you to create the GUI in the main method and then call setVisible(true) on the top-level frame of the application. There was no bothersome scheduling of a Runnable on the event dispatch thread.
Unfortunately, some component implementors did not pay attention to the subtleties of the original single-thread rule. They launched activities on the event dispatch thread without ever bothering to check whether the component was realized. For example, if you call setSelectionStart or setSelectionEnd on a JTextComponent, a caret movement is scheduled in the event dispatch thread, even if the component is not visible.
It might well have been possible to detect and fix these problems, but the Swing designers took the easy way out. They decreed that it is never safe to access components from any thread other than the event dispatch thread. Therefore, you need to construct the user interface in the event dispatch thread, using the calls to EventQueue.invokeLater that you have seen in all our sample programs.
Of course, there are plenty of programs that are not so careful and live by the old version of the single-thread rule, initializing the user interface on the main thread. Those programs incur the slight risk that some of the user interface initialization may cause actions on the event dispatch thread that conflict with actions on the main thread. As we said in Chapter 10, you don’t want to be one of the unlucky few who run into trouble and waste time debugging an intermittent threading bug. Therefore, you should simply follow the strict single-thread rule.
You have now reached the end of Volume I of Core Java. This volume covered the fundamentals of the Java programming language and the parts of the standard library that you need for most programming projects. We hope that you enjoyed your tour through the Java fundamentals and that you found useful information along the way. For advanced topics, such as networking, advanced AWT/Swing, security, and internationalization, please turn to Volume II.
