Chapter 13 Concurrency Notes

Question 1

Introducing Threads
* thread
* process
* single-threaded process
* multithreaded process
* shared environment
* task
* A thread can complete multiple independent tasks but only one task at a time.
* shared memory,

Answer 1

• A thread is the smallest unit of execution that can be scheduled by the operating system.
• A process is a group of associated threads that execute in the same shared environment.
• A task is a single unit of work performed by a thread.
• shared environment, we mean that the threads in the same process share the same memory space and can communicate directly with one another.
• shared memory, we are generally referring to static variables as well as instance and local variables passed to a thread.

Question 2

Understanding Thread Concurrency

Answer 2

• The property of executing multiple threads and processes at the same time is referred to as concurrency.
• Operating systems use a thread scheduler to determine which threads should be currently executing
• A context switch is the process of storing a thread’s current state and later restoring the state of the thread to continue execution.
• Finally, a thread can interrupt or supersede another thread if it has a higher thread priority than the other thread. A thread priority is a numeric value associated with a thread that is taken into consideration by the thread scheduler when determining which threads should currently be executing. In Java, thread priorities are specified as integer values.

Question 3

How does the system decide what to execute when there are more threads available than CPUs?

Answer 3

Operating systems use a thread scheduler to determine which threads should be currently executing, as shown in Figure 13.1. For example, a thread scheduler may employ a round-robin schedule in which each available thread receives an equal number of CPU cycles with which to execute, with threads visited in a circular order.

Question 4

context switch

Answer 4

A context switch is the process of storing a thread’s current state and later restoring the state of the thread to continue execution.

Question 5

thread priority

Answer 5

A thread priority is a numeric value associated with a thread that is taken into consideration by the thread scheduler when determining which threads should currently be executing. In Java, thread priorities are specified as integer values.

Question 6

Runnable

@FunctionalInterface public interface Runnable {
    void run();
}

Answer 6

• One of the most common ways to define a task for a thread is by using the Runnable instance.
• Runnable is a functional interface that takes no arguments and returns no data.

Question 7

Creating a Thread

new Thread(() -> System.out.print("Hello")).start();
System.out.print("World");

Answer 7

• Provide a Runnable object or lambda expression to the Thread constructor.
• start() start new thread

Remember that order of thread execution is not often guaranteed. The exam commonly presents questions in which multiple tasks are started at the same time, and you must determine the result.

Question 8

Calling run() Instead of start()

System.out.println("begin");
new Thread(printInventory).run();
new Thread(printRecords).run();
new Thread(printInventory).run();
System.out.println("end");

Answer 8

Calling run() on a Thread or a Runnable does not start a new thread.

Question 9

we can create a Thread and its associated task one of two ways in Java:

Answer 9

More generally, we can create a Thread and its associated task one of two ways in Java:

• Provide a Runnable object or lambda expression to the Thread constructor.
• Create a class that extends Thread and overrides the run() method.

Creating a class that extends Thread is relatively uncommon and should only be done under certain circumstances, such as if you need to overwrite other thread methods.

Question 10

Distinguishing Thread Types
* System thread
* User-defined thread
* Daemon thread

Answer 10

• System thread, created by the(JVM) and runs in the background of the application. For example, garbage collection is managed by a system thread created by the JVM.
• User-defined thread, created by the application developer to accomplish a specific task.
• Daemon threads
  
  • System and user-defined threads can both be created as daemon threads.
  • by default, user-defined threads are not daemons
  • A Java application terminates when the only threads that are running are daemon threads.
  • set thread as daemon, before start()
job.setDaemon(true);

Question 11

Managing a Thread’s Life Cycle

Answer 11

• NEW (thread created but no started )
• RUNNABLE (running or able to be run)
  
  • BLOCKED (waiting to enter synchronized block)
  • WAITING (waiting indefinitely to be notified)
  • TIMED_WAITING (waiting a specified time)
• TERMINATED (task complete)

create thread -> NEW - start() -> RUNNABLE - run() completes-> TERMINATED

RUNNABLE -resource requested -> BLOCKED
RUNNABLE <-resource granted- BLOCKED

RUNNABLE -wait() -> WAITING
RUNNABLE <-notify() - WAITING

RUNNABLE -sleep() -> TIMED_WAITING
RUNNABLE <-time elapsed - TIMED_WAITING

Question 12

Polling with Sleep

Thread.sleep(1_000); // 1 SECOND

Answer 12

Even though multithreaded programming allows you to execute multiple tasks at the same time, one thread often needs to wait for the results of another thread to proceed. One solution is to use polling. Polling is the process of intermittently checking data at some fixed interval.

We can improve this result by using the Thread.sleep() method to implement polling and sleep for 1,000 milliseconds, aka 1 second:

public class CheckResultsWithSleep {
    private static int counter = 0;
    public static void main(String[] a) {
        new Thread(() -> {
            for(int i = 0; i < 1_000_000; i++) counter++;
        }).start();
        while(counter < 1_000_000) {
            System.out.println("Not reached yet");
            try {
                Thread.sleep(1_000); // 1 SECOND
            } catch (InterruptedException e) {
                System.out.println("Interrupted!");
            }
        }
        System.out.println("Reached: "+counter);
} }

Question 13

Polling

Answer 13

Polling is the process of intermittently checking data at some fixed interval.

Question 14

Interrupting a Thread

mainThread.interrupt();

Answer 14

One way to improve this program is to allow the thread to interrupt the main() thread when it’s done:

public class CheckResultsWithSleepAndInterrupt {
    private static int counter = 0;
    public static void main(String[] a) {
        final var mainThread = Thread.currentThread();
        new Thread(() -> {
            for(int i = 0; i < 1_000_000; i++) counter++;
            mainThread.interrupt();
        }).start();
        while(counter < 1_000_000) {
            System.out.println("Not reached yet");
            try {
                Thread.sleep(1_000); // 1 SECOND
            } catch (InterruptedException e) {
                System.out.println("Interrupted!");
            }
        }
        System.out.println("Reached: "+counter);
} }

final var mainThread = Thread.currentThread();

mainThread.interrupt();

Calling interrupt() on a thread in the TIMED_WAITING or WAITING state causes the main() thread to become RUNNABLE again, triggering an InterruptedException. The thread may also move to a BLOCKED state if it needs to reacquire resources when it wakes up.

Question 15

> [!NOTE]
Calling interrupt() on a thread already in a RUNNABLE state doesn’t change the state.
In fact, it only changes the behavior if the thread is periodically checking the Thread.isInterrupted() value state.

Answer 15

interrupt()
Thread.isInterrupted()

Question 16

Thread methods

Answer 16

• public void run()
  If this thread was constructed using a separate Runnable run object, then that Runnable object’s run method is called; otherwise, this method does nothing and returns.
  Subclasses of Thread should override this method.
• public void start()
  Causes this thread to begin execution; the Java Virtual Machine calls the run method of this thread.
  The result is that two threads are running concurrently: the current thread (which returns from the call to the start method) and the other thread (which executes its run method).
  It is never legal to start a thread more than once. In particular, a thread may not be restarted once it has completed execution.
• public final void join() throws InterruptedException
  Waits for this thread to die.
  An invocation of this metho behaves in exactly the same way as the invocation
• public static void sleep(long millis) throws InterruptedException
  Causes the currently executing thread to sleep (temporarily cease execution) for the specified number of milliseconds, subject to the precision and accuracy of system timers and schedulers. The thread does not lose ownership of any monitors.
• public static Thread currentThread()
  Returns a reference to the currently executing thread object.
• public void interrupt()
  Interrupts this thread.
• public static boolean interrupted()
  Tests whether the current thread has been interrupted. The interrupted status of the thread is cleared by this method. In other words, if this method were to be called twice in succession, the second call would return false (unless the current thread were interrupted again, after the first call had cleared its interrupted status and before the second call had examined it).
  A thread interruption ignored because a thread was not alive at the time of the interrupt will be reflected by this method returning false.
• public boolean isInterrupted()
  Tests whether this thread has been interrupted. The interrupted status of the thread is unaffected by this method.
  A thread interruption ignored because a thread was not alive at the time of the interrupt will be reflected by this method returning false.
• public Thread.State getState()
  Returns the state of this thread. This method is designed for use in monitoring of the system state, not for synchronization control.
• public final void setDaemon(boolean on)
  Marks this thread as either a daemon thread or a user thread. The Java Virtual Machine exits when the only threads running are all daemon threads.
  This method must be invoked before the thread is started.

Question 17

Creating Threads with the Concurrency API

• java.util.concurrent
• public interface Executor
• public class Executors extends Object
• public interface ExecutorService extends Executor

Answer 17

• java.util.concurrent package
• The Concurrency API includes the ExecutorService interface, which defines services that create and manage threads.
  
  • You first obtain an instance of an ExecutorService interface,
  • and then you send the service tasks to be processed.
  • The framework includes numerous useful features, such as thread pooling and scheduling.
• It is recommended that you use this framework any time you need to create and execute a separate task, even if you need only a single thread.

Question 18

Introducing the Single-Thread Executor

ExecutorService service = Executors.newSingleThreadExecutor();
try {
    System.out.println("begin");
    service.execute(printInventory);
    service.execute(printRecords);
    service.execute(printInventory);
    System.out.println("end");
} finally {
    service.shutdown();
}

Answer 18

Possible output:
begin
Printing zoo inventory
Printing record: 0
Printing record: 1
end
Printing record: 2
Printing zoo inventory

ExecutorService service = Executors.newSingleThreadExecutor();

• The Concurrency API includes the Executors factory class that can be used to create instances of the ExecutorService interface.
• we use the newSingleThreadExecutor() method to create the service.
• With a single-thread executor, tasks are guaranteed to be executed sequentially.
• Notice that the end text is output while our thread executor tasks are still running.
• This is because the main() method is still an independent thread from the ExecutorService.

Question 19

Shutting Down a Thread Executor

• boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException
• boolean isTerminated()
• boolean isShutdown()
• void shutdown()
• List<Runnable> shutdownNow()

Answer 19

• A thread executor creates a non-daemon thread on the first task that is executed, so failing to call shutdown() will result in your application never terminating.
• The shutdown process for a thread executor involves
  
  • first rejecting any new tasks submitted to the thread executor
  • while continuing to execute any previously submitted tasks.
  • During this time, calling isShutdown() will return true, while isTerminated() will return false.
  • If a new task is submitted to the thread executor while it is shutting down, a RejectedExecutionException will be thrown.
  • Once all active tasks have been completed, isShutdown() and isTerminated() will both return true.
• Figure 13.3 shows the life cycle of an ExecutorService object.

For the exam, you should be aware that shutdown() does not stop any tasks that have already been submitted to the thread executor.

Question 20

What if you want to cancel all running and upcoming tasks?

Answer 20

• The ExecutorService provides a method called shutdownNow(),
• which attempts to stop all running tasks and discards any that have not been started yet.
• It is not guaranteed to succeed
• because it is possible to create a thread that will never terminate, so any attempt to interrupt it may be ignored.

List<Runnable> shutdownNow()
* Attempts to stop all actively executing tasks,
* halts the processing of waiting tasks,
* and returns a list of the tasks that were awaiting execution.
* This method does not wait for actively executing tasks to terminate.
* Use awaitTermination to do that.
* There are no guarantees beyond best-effort attempts to stop processing actively executing tasks.
* For example, typical implementations will cancel via Thread.interrupt(), so any task that fails to respond to interrupts may never terminate.

Question 21

Submitting Tasks
ExecutorService

• <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException
• <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException
• Future<?> submit(Runnable task)
• <T> Future<T> submit(Runnable task, T result)
• <T> Future<T> submit(Callable<T> task)

Executor

• void execute(Runnable command)

Answer 21

You can submit tasks to an ExecutorService instance multiple ways.
1. void execute(Runnable command)
Executes Runnable task at some point in future.
2. Future<?> submit(Runnable task)
Executes Runnable task at some point in future and returns Future representing task.
3. <T> Future<T> submit(Callable<T> task)
Executes Callable task at some point in future and returns Future representing pending results of task.
4. <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
Executes given tasks and waits for all tasks to complete. Returns List of Future instances in same order in which they were in original collection.
5. <T> T invokeAny(Collection<? extends Callable<T>> tasks)
Executes given tasks and waits for at least one to complete.

Question 22

Submitting Tasks: execute() vs. submit()

Answer 22

The submit() method advantage over execute() method is a return object that can be used to track the result.

For the exam, you need to be familiar with both execute() and submit(), but in your own code we recommend submit() over execute() whenever possible.

Question 23

How do we know when a task submitted to an ExecutorService is complete?

Answer 23

submit() method returns a Future<V> instance that can be used to determine this result.

Future<?> future = service.submit(() -> System.out.println("Hello"));

• The Future type is actually an interface.
• Future instance is returned by various API methods.

Question 24

Future methods

Answer 24

1. boolean isDone()
   Returns true if task was completed, threw exception, or was cancelled.
2. boolean isCancelled()
   Returns true if task was cancelled before it completed normally.
3. boolean cancel(boolean mayInterruptIfRunning)
   Attempts to cancel execution of task and returns true if it was successfully cancelled or false if it could not be cancelled or is complete.
4. V get()
   Retrieves result of task, waiting endlessly if it is not yet available.
5. V get(long timeout, TimeUnit unit)
   Retrieves result of task, waiting specified amount of time. If result is not ready by time timeout is reached, checked TimeoutException will be thrown.

Question 25

The following is an updated version of our earlier polling example CheckResults class, which uses a Future instance to wait for the results: ``` import java.util.concurrent.*; public class CheckResults { private static int counter = 0; public static void main(String[] unused) throws Exception { ExecutorService service = Executors.newSingleThreadExecutor(); try { Future result = service.submit(() -> { for(int i = 0; i < 1_000_000; i++) counter++; }); result.get(10, TimeUnit.SECONDS); // Returns null for Runnable System.out.println("Reached!"); } catch (TimeoutException e) { System.out.println("Not reached in time"); } finally { service.shutdown(); } } } ```

Answer 25

***`import java.util.concurrent.*;`*** ***`ExecutorService service = Executors.newSingleThreadExecutor();`*** * **Obtain ExecutorService instance using a factory method in the Executors class** * **Creates an Executor that uses a single worker thread operating off an unbounded queue.** ***`Future result = service.submit(() -> { for(int i = 0; i < 1_000_000; i++) counter++; });`*** * Since the **return type of Runnable run() is void**, **the *`get()`* method always returns *`null`* when working with Runnable expressions.** ***`result.get(10, TimeUnit.SECONDS); // Returns null for Runnable`*** * It also waits at most **10 seconds**, throwing a ***`TimeoutException`*** on the call to ***`result.get()`*** if the task is not done. ***` Future submit(Runnable task, T result)`*** * Submits a Runnable task for execution and returns a Future representing that task. The Future's get method will return the given result upon successful completion.

Question 26

***`java.util.concurrent.TimeUnit`*** **Concurrency API use this enum**

Answer 26

**TimeUnit enum** * ***`TimeUnit.NANOSECONDS`*** Time in one-billionths of a second (1/1,000,000,000) * ***`TimeUnit.MICROSECONDS`*** Time in one-millionths of a second (1/1,000,000) * ***`TimeUnit.MILLISECONDS`*** Time in one-thousandths of a second (1/1,000) * ***`TimeUnit.SECONDS`*** Time in seconds * ***`TimeUnit.MINUTES`*** Time in minutes * ***`TimeUnit.HOURS`*** Time in hours * ***`TimeUnit.DAYS`*** Time in days

Question 27

**Introducing Callable** ``` @FunctionalInterface public interface Callable { V call() throws Exception; } ```

Answer 27

* ***`java.util.concurrent.Callable functional interface`*** * ***`call()`*** method **`returns a value`** and **can throw a checked exception**. ``` @FunctionalInterface public interface Callable { V call() throws Exception; } ``` * The **Callable** interface is often **preferable over Runnable,** since it allows more details to be retrieved easily from the task after it is completed.

Question 28

Example using **Callable**: ``` var service = Executors.newSingleThreadExecutor(); try { Future result = service.submit(() -> 30 + 11); System.out.println(result.get()); // 41 } finally { service.shutdown(); } ```

Answer 28

Callable

Question 29

Waiting for All Tasks to Finish ``` ExecutorService service = Executors.newSingleThreadExecutor(); try { // Add tasks to the thread executor … } finally { service.shutdown(); } service.awaitTermination(1, TimeUnit.MINUTES); // Check whether all tasks are finished if(service.isTerminated()) System.out.println("Finished!"); else System.out.println("At least one task is still running"); ```

Answer 29

If we don't need the results of the tasks and are finished using our thread executor, there is a simpler approach. 1. First, we shut down the thread executor using the ***`shutdown()`*** method. 2. Next, we use the ***`awaitTermination()`*** method available for all thread executors. The method waits the specified time to complete all tasks, returning sooner if all tasks finish or an ***`InterruptedException`*** is detected. 3. call ***`isTerminated()`*** after the ***`awaitTermination()`*** method finishes to confirm that all tasks are finished.

Question 30

Scheduling Tasks

Answer 30

Creates a thread pool that can schedule commands to run after a given delay, or to execute periodically. ***`ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();`***

Question 31

**ScheduledExecutorService methods**

Answer 31

* ***`schedule(Callable callable, long delay, TimeUnit unit)`*** Creates and executes Callable task after given delay * ***`schedule(Runnable command, long delay, TimeUnit unit)`*** Creates and executes Runnable task after given delay * ***`scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)`*** Creates and executes Runnable task after given initial delay, creating new task every period value that passes * ***`scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)`*** Creates and executes Runnable task after given initial delay and subsequently with given delay between termination of one execution and commencement of next

Question 32

***`scheduleWithFixedDelay()`***

Answer 32

***`scheduleWithFixedDelay()`*** method **creates a new task only after the previous task has finished.** For example, if a task runs at **12:00** and takes **five minutes** to finish, with a period between executions of **two minutes**, the next task will start at **12:07**. ***`service.scheduleWithFixedDelay(task1, 0, 2, TimeUnit.MINUTES);`***

Question 33

**Increasing Concurrency with Pools**

Answer 33

A ***`thread pool`*** is a **group of pre-instantiated reusable threads** that are available to perform a set of arbitrary tasks.

Question 34

**Executors factory methods**

Answer 34

* ***`ExecutorService newSingleThreadExecutor()`*** Creates single-threaded executor that uses single worker thread operating off unbounded queue. Results are processed sequentially in order in which they are submitted. * ***`ScheduledExecutorService newSingleThreadScheduledExecutor()`*** Creates single-threaded executor that can schedule commands to run after given delay or to execute periodically. * ***`ExecutorService newCachedThreadPool()`*** Creates thread pool that creates new threads as needed but reuses previously constructed threads when they are available. * ***`ExecutorService newFixedThreadPool(int)`*** Creates thread pool that reuses fixed number of threads operating off shared unbounded queue. * ***`ScheduledExecutorService newScheduledThreadPool(int)`*** Creates thread pool that can schedule commands to run after given delay or execute periodically.

Question 35

**Writing Thread-Safe Code**

Answer 35

* ***`Thread-safety`*** is the property of an object that guarantees safe execution by multiple threads at the same time. * Since threads run in a **shared environment and memory space**, how do we prevent two threads from interfering with each other? We must **organize access to data** so that we don't end up with invalid or unexpected results. * how to use a variety of techniques to protect data, including * ***`atomic classes`***, * ***`synchronized blocks`***, * the ***`Lock framework`***, * and ***`cyclic barriers`***.

Question 36

**Understanding Thread-Safety**

Answer 36

``` 1: import java.util.concurrent.*; 2: public class SheepManager { 3: private int sheepCount = 0; 4: private void incrementAndReport() { 5: System.out.print((++sheepCount)+" "); 6: } 7: public static void main(String[] args) { 8: ExecutorService service = Executors.newFixedThreadPool(20); 9: try { 10: SheepManager manager = new SheepManager(); 11: for(int i = 0; i < 10; i++) 12: service.submit(() -> manager.incrementAndReport()); 13: } finally { 14: service.shutdown(); 15: } } } ``` It may output in a **different order**. Worse yet, it may **print some numbers twice** and **not print some numbers at all**! The following are possible outputs of this program: 1 2 3 4 5 6 7 8 9 10 1 9 8 7 3 **6 6** 2 4 5 1 8 7 3 **2** 6 5 4 **2** 9

Question 37

**race condition**

Answer 37

the unexpected result of two tasks executing at the same time is referred to as a **race condition**.

Question 38

**Accessing Data with volatile**

Answer 38

The ***`volatile`*** keyword is used to **guarantee that access to data within memory is consistent.** The ***`volatile`*** attribute ensures that **only one thread is modifying a variable at one time and that data read among multiple threads is consistent.** ``` 3: private volatile int sheepCount = 0; 4: private void incrementAndReport() { 5: System.out.print((++sheepCount)+" "); 6: } ``` Unfortunately, this code is not thread-safe and could still result in numbers being missed: **2** 6 1 7 5 3 **2** 9 4 8 The reason this code is not thread-safe is that ***`++sheepCount`*** is still two distinct operations. Put another way, if the increment operator represents the expression ***`sheepCount = sheepCount + 1`***, then each read and write operation is thread-safe, but the combined operation is not. Referring back to our sheep example, we don't interrupt the employee while running, but we could still have multiple people in the field at the same time.

Question 39

> [!NOTE] > In practice, **volatile** is rarely used. > We only cover it because it has been known to show up on the exam from time to time.

Answer 39

**volatile**

Question 40

**Protecting Data with Atomic Classes**

Answer 40

**Atomic is the property of an operation to be carried out as a single unit of execution without any interference from another thread.** **A thread-safe atomic version of the increment operator would perform the read and write of the variable as a single operation, not allowing any other threads to access the variable during the operation.** **Figure 13.5 shows the result of making the sheepCount variable atomic.**

Question 41

**Atomic classes**

Answer 41

* ***`AtomicBoolean`*** A boolean value that may be updated atomically * ***`AtomicInteger`*** An int value that may be updated atomically * ***`AtomicLong`*** A long value that may be updated atomically

Question 42

``` 3: private AtomicInteger sheepCount = new AtomicInteger(0); 4: private void incrementAndReport() { 5: System.out.print(sheepCount.incrementAndGet()+" "); 6: } ```

Answer 42

``` 2 3 1 4 5 6 7 8 9 10 1 4 3 2 5 6 7 8 9 10 1 4 3 5 6 2 7 8 10 9 ```

Question 43

**Common atomic methods**

Answer 43

* ***get()*** Retrieves current value * ***`set(type newValue)`*** Sets given value, equivalent to assignment = operator * ***`getAndSet(type newValue)`*** Atomically sets new value and returns old value * ***`incrementAndGet()`*** For numeric classes, atomic pre-increment operation equivalent to ++value * ***`getAndIncrement()`*** For numeric classes, atomic post-increment operation equivalent to value++ * ***`decrementAndGet()`*** For numeric classes, atomic pre-decrement operation equivalent to --value * ***`getAndDecrement()`*** For numeric classes, atomic post-decrement operation equivalent to value--

Question 44

**Improving Access with synchronized Blocks**

Answer 44

* The most common technique is to **use a monitor to synchronize access**. * A **monitor**, also called a **lock**, is a structure that supports **mutual exclusion**, which is the property that **at most one thread is executing a particular segment of code at a given time.** In **Java**, any **Object** can be used as a **monitor**, along with the **synchronized** keyword, as shown in the following example: ``` var manager = new SheepManager(); synchronized(manager) { // Work to be completed by one thread at a time } ``` This example is referred to as a **synchronized block**. * Each thread that arrives will **first check if any threads are already running the block.** * If the **lock is not available**, the thread will transition to a **BLOCKED** state until it can **“acquire the lock.”** * If the **lock is available** (or the thread already holds the lock), the single thread will **enter the block, preventing all other threads from entering.** * Once the **thread finishes executing the block**, it will **release the lock, allowing one of the waiting threads to proceed.**

Question 45

**synchronized block**

Answer 45

**synchronized block**. * Each thread that arrives will **first check if any threads are already running the block.** * If the **lock is not available**, the thread will transition to a **BLOCKED** state until it can **“acquire the lock.”** * If the **lock is available** (or the thread already holds the lock), the single thread will **enter the block, preventing all other threads from entering.** * Once the **thread finishes executing the block**, it will **release the lock, allowing one of the waiting threads to proceed.**

Question 46

>[!NOTE] >To **synchronize** access across multiple threads, each thread must have access to the **same Object.** >**If each thread synchronizes on different objects, the code is not thread-safe.**

Answer 46

**synchronize**

Question 47

**Does this solution fix the problem?** ``` 11: for(int i = 0; i < 10; i++) { 12: synchronized(manager) { 13: service.submit(() -> manager.incrementAndReport()); 14: } 15: } ```

Answer 47

No, it does not! Can you spot the problem? We've synchronized the creation of the threads but not the execution of the threads. In this example, the threads would be created one at a time, but they might all still execute and perform their work simultaneously, resulting in the same type of output that you saw earlier. We did say diagnosing and resolving thread problems is difficult in practice!

Question 48

``` 5: synchronized(this) { 6: System.out.print((++sheepCount)+" "); 7: } ```

Answer 48

synchronized

Question 49

We could have synchronized on any object, as long as it was the same object. For example, the following code snippet would also work: ``` 4: private final Object herd = new Object(); 5: private void incrementAndReport() { 6: synchronized(herd) { 7: System.out.print((++sheepCount)+" "); 8: } 9: } ``` Although we didn't need to make the herd variable final, doing so ensures that it is not reassigned after threads start using it.

Answer 49

When this code executes, it will consistently output the following: 1 2 3 4 5 6 7 8 9 10 Although all threads are still created and executed at the same time, they each wait at the synchronized block for the worker to increment and report the result before entering.

Question 50

Synchronizing on Methods

Answer 50

We can add the synchronized modifier to any instance method to synchronize automatically on the object itself. For example, the following two method definitions are equivalent: ``` void sing() { synchronized(this) { System.out.print("La la la!"); } } synchronized void sing() { System.out.print("La la la!"); } ``` The first uses a synchronized block, whereas the second uses the synchronized method modifier. Which you use is completely up to you.

Question 51

We can also apply the synchronized modifier to static methods. What object is used as the monitor when we synchronize on a static method?

Answer 51

The class object, of course! For example, the following two methods are equivalent for static synchronization inside our SheepManager class: ``` static void dance() { synchronized(SheepManager.class) { System.out.print("Time to dance!"); } } static synchronized void dance() { System.out.print("Time to dance!"); } ``` As before, the first uses a synchronized block, with the second example using the synchronized modifier. You can use static synchronization if you need to order thread access across all instances rather than a single instance.

Question 52

Understanding the Lock Framework

Answer 52

**A synchronized block supports only a limited set of functionality.** For example, what if we want to check whether a lock is available and, if it is not, perform some other task? Furthermore, if the lock is never available and we synchronize on it, we might wait forever. The **Concurrency API** includes the **Lock interface**, which is conceptually similar to using the synchronized keyword but with a lot more bells and whistles. Instead of synchronizing on any Object, though, **we can “lock” only on an object that implements the Lock interface.**

Question 53

**Applying a ReentrantLock**

Answer 53

The **Lock interface** is pretty easy to use. When you need to protect a piece of code from multithreaded processing, **create an instance of Lock that all threads have access to.** **Each thread then calls *`lock()`* before it enters the protected code and calls *`unlock()`* before it exits the protected code.** ``` // Implementation #1 with a synchronized block Object object = new Object(); synchronized(object) { // Protected code } // Implementation #2 with a Lock Lock lock = new ReentrantLock(); try { lock.lock(); // Protected code } finally { lock.unlock(); } ``` The ***`ReentrantLock`*** class is a simple monitor that implements the Lock interface and supports mutual exclusion. In other words, at most one thread is allowed to hold a lock at any given time. The ReentrantLock class ensures that once a thread has called **lock()** and obtained the lock, all other threads that call **lock()** will wait until the first thread calls **unlock()**. **Which thread gets the lock next depends on the parameters used to create the Lock object.** The **ReentrantLock** class includes a **constructor** that takes a **single boolean and sets a “fairness” parameter**. If the parameter is set to **true**, the **lock will usually be granted to each thread in the order in which it was requested**. It is **false** by default when using the **no-argument constructor**. In practice, you should **enable fairness only when ordering is absolutely required, as it could lead to a *`significant slowdown`*.** Besides always making sure to release a lock, you also need to be sure that you only release a lock that you have. If you attempt to release a lock that you do not have, you will get an exception at runtime. ``` Lock lock = new ReentrantLock(); lock.unlock(); // IllegalMonitorStateException ```

Question 54

>[!TIPS] >While certainly not required, it is a good practice to use a try/finally block with Lock instances. >Doing so ensures that any acquired locks are properly released.

Answer 54

lock

Question 55

Attempting to Acquire a Lock

Answer 55

While the ReentrantLock class allows you to wait for a lock, it so far suffers from the same problem as a synchronized block. A thread could end up waiting forever to obtain a lock. Luckily, Table 13.8 includes two additional methods that make the Lock interface a lot safer to use than a synchronized block. * ***`void lock()`*** Requests lock and blocks until lock is acquired. * ***`void unlock()`*** Releases lock. * ***`boolean tryLock()`*** Requests lock and returns immediately. Returns boolean indicating whether lock was successfully acquired. * ***`boolean tryLock(long timeout, TimeUnit unit)`*** Requests lock and blocks for specified time or until lock is acquired. Returns boolean indicating whether lock was successfully acquired.

Question 56

***`tryLock()`***

Answer 56

The **tryLock()** method will **attempt to acquire a lock and *immediately* return a *boolean* result indicating whether the lock was obtained.** Unlike the **lock()** method, it **does not wait if another thread already holds the lock.** It returns immediately, regardless of whether a lock is available. The following is a sample implementation using the tryLock() method: ``` Lock lock = new ReentrantLock(); new Thread(() -> printHello(lock)).start(); if(lock.tryLock()) { try { System.out.println("Lock obtained, entering protected code"); } finally { lock.unlock(); } } else { System.out.println("Unable to acquire lock, doing something else"); } ``` Like lock(), the tryLock() method **should be used with a try/finally block.** Fortunately, you need to release the lock only if it was successfully acquired. For this reason, it is common to use the output of tryLock() in an if statement, so that unlock() is called only when the lock is obtained.

Question 57

>[!TIPS] >It is imperative that your program always check the return value of the tryLock() method. >It tells your program whether it is safe to proceed with the operation and whether the lock needs to be released later.

Answer 57

tryLock()

Question 58

**tryLock(long,TimeUnit)**

Answer 58

The Lock interface includes an overloaded version of **tryLock(long,TimeUnit)** that acts **like a hybrid of lock() and tryLock().** Like the other two methods, **if a lock is available, it will immediately return with it. If a lock is unavailable, though, it will wait up to the specified time limit** for the lock. The following code snippet uses the overloaded version of tryLock(long,TimeUnit): ``` Lock lock = new ReentrantLock(); new Thread(() -> printHello(lock)).start(); if(lock.tryLock(10,TimeUnit.SECONDS)) { try { System.out.println("Lock obtained, entering protected code"); } finally { lock.unlock(); } } else { System.out.println("Unable to acquire lock, doing something else"); } ``` The code is the same as before, except this time, one of the threads **waits up to 10 seconds to acquire the lock.**

Question 59

Acquiring the Same Lock Twice

Answer 59

The ReentrantLock class maintains a counter of the number of times a lock has been successfully granted to a thread. To release the lock for other threads to use, unlock() must be called the same number of times the lock was granted. It is critical that you release a lock the same number of times it is acquired! **For calls with tryLock(), you need to call unlock() only if the method returned true.**

Question 60

The following code snippet contains an error. Can you spot it? ``` Lock lock = new ReentrantLock(); if(lock.tryLock()) { try { lock.lock(); System.out.println("Lock obtained, entering protected code"); } finally { lock.unlock(); } } ```

Answer 60

The thread obtains the lock twice but releases it only once. You can verify this by spawning a new thread after this code runs that attempts to obtain a lock. The following prints false: `new Thread(() -> System.out.print(lock.tryLock())).start(); // false` It is critical that you release a lock the same number of times it is acquired! **For calls with tryLock(), you need to call unlock() only if the method returned true.**

Question 61

Reviewing the Lock Framework

Answer 61

To review, the **ReentrantLock class** supports the same features as a **synchronized block** while adding a number of improvements: * **Ability to request a lock without blocking.** * **Ability to request a lock while blocking for a specified amount of time**. * **A lock can be created with a fairness property, in which the lock is granted to threads in the order in which it was requested.**

Question 62

>[!TIPS] >While not on the exam, ReentrantReadWriteLock is a really useful class. It includes separate locks for reading and writing data and is useful on data structures where reads are far more common than writes. For example, if you have a thousand threads reading data but only one thread writing data, this class can help you maximize concurrent access.

Answer 62

ReentrantReadWriteLock

Question 63

Orchestrating Tasks with a CyclicBarrier

Answer 63

how to orchestrate complex tasks with many steps. The CyclicBarrier takes in its constructors a limit value, indicating the number of threads to wait for. As each thread finishes, it calls the await() method on the cyclic barrier. Once the specified number of threads have each called await(), the barrier is released, and all threads can continue.

Question 64

To coordinate these tasks, we can use the CyclicBarrier class: import java.util.concurrent.*; public class LionPenManager { private void removeLions() { System.out.println("Removing lions"); } private void cleanPen() { System.out.println("Cleaning the pen"); } private void addLions() { System.out.println("Adding lions"); } public void performTask() { removeLions(); cleanPen(); addLions(); } public static void main(String[] args) { var service = Executors.newFixedThreadPool(4); try { var manager = new LionPenManager(); for (int i = 0; i < 4; i++) service.submit(() -> manager.performTask()); } finally { service.shutdown(); } } }

Answer 64

The following is sample output based on this implementation: Removing lions Removing lions Cleaning the pen Adding lions Removing lions Cleaning the pen Adding lions Removing lions Cleaning the pen Adding lions Cleaning the pen Adding lions

Question 65

``` import java.util.concurrent.*; public class LionPenManager { private void removeLions() { System.out.println("Removing lions"); } private void cleanPen() { System.out.println("Cleaning the pen"); } private void addLions() { System.out.println("Adding lions"); } public void performTask(CyclicBarrier c1, CyclicBarrier c2) { try { removeLions(); c1.await(); cleanPen(); c2.await(); addLions(); } catch (InterruptedException | BrokenBarrierException e) { // Handle checked exceptions here } } public static void main(String[] args) { var service = Executors.newFixedThreadPool(4); try { var manager = new LionPenManager(); var c1 = new CyclicBarrier(4); var c2 = new CyclicBarrier(4, () -> System.out.println("*** Pen Cleaned!")); for (int i = 0; i < 4; i++) service.submit(() -> manager.performTask(c1, c2)); } finally { service.shutdown(); } } } ```

Answer 65

***`var c1 = new CyclicBarrier(4);`*** executes a Runnable instance upon completion. ***`var c2 = new CyclicBarrier(4, () -> System.out.println(" Pen Cleaned!"));`*** The following is sample output based on this revised implementation of our LionPenManager class: ``` Removing lions Removing lions Removing lions Removing lions Cleaning the pen Cleaning the pen Cleaning the pen Cleaning the pen *** Pen Cleaned! Adding lions Adding lions Adding lions Adding lions ```

Question 66

**Reusing CyclicBarrier**

Answer 66

**After a CyclicBarrier limit is reached (aka the barrier is broken), all threads are released**, and the number of threads waiting on the CyclicBarrier goes back to zero. At this point, the CyclicBarrier may be used again for a new set of waiting threads. For example, if our CyclicBarrier limit is 5 and we have 15 threads that call await(), the CyclicBarrier will be activated a total of three times.

Question 67

Using Concurrent Collections

Answer 67

The concurrent classes were created to help avoid common issues in which multiple threads are adding and removing objects from the same collections. At any given instance, all threads should have the same consistent view of the structure of the collection.

Question 68

Understanding Memory Consistency Errors

Answer 68

A ***`memory consistency error`*** occurs when two threads have **inconsistent views** of what should be the same data. Conceptually, we want writes on one thread to be available to another thread if it accesses the concurrent collection after the write has occurred. When two threads try to modify the same nonconcurrent collection, the JVM may throw a ***`ConcurrentModificationException`*** at runtime. In fact, it can happen with a single thread.

Question 69

``` 11: var foodData = new HashMap(); 12: foodData.put("penguin", 1); 13: foodData.put("flamingo", 2); 14: for(String key: foodData.keySet()) 15: foodData.remove(key); ```

Answer 69

This snippet will throw a **ConcurrentModificationException** during the second iteration of the loop, since the iterator on keySet() is not properly updated after the first element is removed. Changing the first line to use a **ConcurrentHashMap** will prevent the code from throwing an exception at runtime. ``` 11: var foodData = new ConcurrentHashMap(); ``` ConcurrentHashMap is ordering read/write access such that all access to the class is consistent. In this code snippet, the iterator created by keySet() is updated as soon as an object is removed from the Map.

Question 70

Working with Concurrent Classes

Answer 70

Working with Concurrent Classes

Question 71

> [!TIPS] > If the collection is immutable (and contains immutable objects), the concurrent collections are not necessary. > Immutable objects can be accessed by any number of threads and do not require synchronization. > By definition, they do not change, so there is no chance of a memory consistency error.

Answer 71

concurrent collections

Question 72

When passing around a concurrent collection, a caller may need to know the particular implementation class. That said, it is considered a good practice to pass around a nonconcurrent interface reference when possible, similar to how we instantiate a HashMap but often pass around a Map reference: ***`Map map = new ConcurrentHashMap<>();`***

Answer 72

***`Map map = new ConcurrentHashMap<>();`***

Question 73

lists the common concurrent classes with which you should be familiar for the exam.

Answer 73

* ***`ConcurrentHashMap`***, interfaces : Map, ConcurrentMap * ***`ConcurrentLinkedQueue`***, interfaces : Queue * ***`ConcurrentSkipListMap`***, interfaces : Map, SortedMap, NavigableMap, ConcurrentMap, ConcurrentNavigableMap * ***`ConcurrentSkipListSet`***, interfaces : Set, SortedSet, NavigableSet * ***`CopyOnWriteArrayList`***, interfaces : List * ***`CopyOnWriteArraySet`***, interfaces : Set * ***`LinkedBlockingQueue`***, interfaces : Queue, BlockingQueue

Question 74

Working with Concurrent Classes

Answer 74

***`ConcurrentHashMap`*** vs. ***`Map`***, or ***`ConcurrentLinkedQueue`*** vs. ***`Queue`***. For the exam, you don't need to know any class-specific concurrent methods. You just **need to know the inherited methods, such as get() and set() for List instances.** The **Skip** classes might sound strange, but they are just **“sorted”** versions of the associated **concurrent collections**. When you see a class with **Skip** in the name, just think **“sorted concurrent” collections**, and the rest should follow naturally. The **CopyOnWrite** classes behave a little differently than the other concurrent examples you have seen. These classes **create a copy of the collection any time a reference is added, removed, or changed in the collection and then update the original collection reference to point to the copy.** These classes are commonly used to ensure an iterator doesn't see modifications to the collection. LinkedBlockingQueue, which implements the concurrent BlockingQueue interface. This class is just like a regular Queue, except that it includes overloaded versions of offer() and poll() that take a timeout. These methods wait (or block) up to a specific amount of time to complete an operation.

Question 75

``` List favNumbers = new CopyOnWriteArrayList<>(List.of(4, 3, 42)); for (var n : favNumbers) { System.out.print(n + " "); // 4 3 42 favNumbers.add(n+1); } System.out.println(); System.out.println("Size: " + favNumbers.size()); // Size: 6 ```

Answer 75

Despite adding elements, the iterator is not modified, and the loop executes exactly three times. Alternatively, if we had used a regular ArrayList object, a ConcurrentModificationException would have been thrown at runtime. The CopyOnWrite classes can use a lot of memory, since a new collection structure is created any time the collection is modified. Therefore, they are commonly used in multithreaded environment situations where reads are far more common than writes.

Question 76

> [!NOTE] > A CopyOnWrite instance is similar to an immutable object, as a new underlying structure is created every time the collection is modified. > Unlike a true immutable object, though, the reference to the object stays the same even while the underlying data is changed.

Answer 76

CopyOnWrite

Question 77

Synchronized Collections methods

Answer 77

These synchronized methods are defined in the **Collections** class. They operate on the inputted collection and return a reference that is the same type as the underlying collection. * synchronizedCollection(Collection c) * synchronizedList(List list) * synchronizedMap(Map m) * synchronizedNavigableMap(NavigableMap m) * synchronizedNavigableSet(NavigableSet s) * synchronizedSet(Set s) * synchronizedSortedMap(SortedMap m) * synchronizedSortedSet(SortedSet s)

Question 78

If you're writing code to create a collection and it requires synchronization, you should use the classes defined in Table 13.9. On the other hand, if you are passed a nonconcurrent collection and need synchronization, use the methods in Table 13.10.

Answer 78

collection

Question 79

Identifying Threading Problems

Answer 79

A threading problem can occur in multithreaded applications when two or more threads interact in an unexpected and undesirable way.

Question 80

Understanding Liveness

Answer 80

Many thread operations can be performed independently, but some **require coordination**. For example, **synchronizing on a method** requires all threads that call the method to **wait for other threads to finish before continuing.** You also saw earlier in the chapter that threads in a **CyclicBarrier** will each **wait for the barrier limit to be reached before continuing.** ***`Liveness`*** is the ability of an application to be able to execute in a timely manner. **Liveness problems**, then, are those in which the application becomes unresponsive or is in some kind of “stuck” state. More precisely, liveness problems are often the result of a thread entering a **BLOCKING** or **WAITING** state **forever**, or **repeatedly** entering/exiting these states. For the exam, there are three types of liveness issues with which you should be familiar: **deadlock**, **starvation**, and **livelock**.

Question 81

Deadlock

Answer 81

Deadlock occurs when two or more threads are blocked forever, each waiting on the other. ``` import java.util.concurrent.*; class Food {} class Water {} public record Fox(String name) { public void eatAndDrink(Food food, Water water) { synchronized(food) { System.out.println(name() + " Got Food!"); move(); synchronized(water) { System.out.println(name() + " Got Water!"); } } } public void drinkAndEat(Food food, Water water) { synchronized(water) { System.out.println(name() + " Got Water!"); move(); synchronized(food) { System.out.println(name() + " Got Food!"); } } } public void move() { try { Thread.sleep(100); } catch (InterruptedException e) {} } public static void main(String[] args) { // Create participants and resources var foxy = new Fox("Foxy"); var tails = new Fox("Tails"); var food = new Food(); var water = new Water(); // Process data var service = Executors.newScheduledThreadPool(10); try { service.submit(() -> foxy.eatAndDrink(food,water)); service.submit(() -> tails.drinkAndEat(food,water)); } finally { service.shutdown(); } } } ``` The result is that our program outputs the following, and it hangs indefinitely: ``` Foxy Got Food! Tails Got Water! ```

Question 82

Starvation

Answer 82

Starvation occurs when a single thread is perpetually denied access to a shared resource or lock. The thread is still active, but it is unable to complete its work as a result of other threads constantly taking the resource that it is trying to access.

Question 83

Livelock

Answer 83

Livelock occurs when two or more threads are conceptually blocked forever, although they are each still active and trying to complete their task. Livelock is a special case of resource starvation in which two or more threads actively try to acquire a set of locks, are unable to do so, and restart part of the process. Livelock is often a result of two threads trying to resolve a deadlock. In practice, livelock is often a difficult issue to detect. Threads in a livelock state appear active and able to respond to requests, even when they are stuck in an endless cycle.

Question 84

Managing Race Conditions

Answer 84

A race condition is an undesirable result that occurs when two tasks that should be completed sequentially are completed at the same time. We encountered examples of race conditions earlier in the chapter when we introduced synchronization. Possible Outcomes for This Race Condition * Both users are able to create accounts with the username ZooFan. * Neither user is able to create an account with the username ZooFan, and an error message is returned to both users. * One user is able to create an account with the username ZooFan, while the other user receives an error message. **For the exam, you should understand that race conditions lead to invalid data if they are not properly handled. Even the solution where both participants fail to proceed is preferable to one in which invalid data is permitted to enter the system.**

Question 85

Working with Parallel Streams

Answer 85

A parallel stream is capable of processing results concurrently, using multiple threads. For example, you can use a parallel stream and the map() operation to operate concurrently on the elements in the stream, vastly improving performance over processing a single element at a time. Using a parallel stream can change not only the performance of your application but also the expected results. As you shall see, some operations also require special handling to be able to be processed in a parallel manner.

Question 86

serial stream

Answer 86

A serial stream is a stream in which the results are ordered, with only one entry being processed at a time.

Question 87

> [!TIPS] > The number of threads available in a parallel stream is proportional to the number of available CPUs in your environment.

Answer 87

parallel stream

Question 88

**Creating Parallel Streams**

Answer 88

For the exam, you should be familiar with **two ways** of creating a parallel stream. ***`Collection collection = List.of(1,2);`*** ***`Stream p1 = collection.stream().parallel();`*** ***`Stream p2 = collection.parallelStream();`*** The first way to create a parallel stream is from an existing stream. Isn't this cool? Any stream can be made parallel! ***`parallel()`*** The second way to create a parallel stream is from a Java Collection class. We use both of these methods throughout this section. ***`parallelStream()`***

Question 89

Performing a Parallel Decomposition

Answer 89

A parallel decomposition is the process of taking a task, breaking it into smaller pieces that can be performed concurrently, and then reassembling the results. The more concurrent a decomposition, the greater the performance improvement of using parallel streams. Let's try it out. First, let's define a reusable function that “does work” just by waiting for five seconds. ``` private static int doWork(int input) { try { Thread.sleep(5000); } catch (InterruptedException e) {} return input; } ``` We can pretend that in a real application, this work might involve calling a database or reading a file. Now let's use this method with a serial stream. ``` 10: long start = System.currentTimeMillis(); 11: List.of(1,2,3,4,5) 12: .stream() 13: .map(w -> doWork(w)) 14: .forEach(s -> System.out.print(s + " ")); 15: 16: System.out.println(); 17: var timeTaken = (System.currentTimeMillis()-start)/1000; 18: System.out.println("Time: "+timeTaken+" seconds"); ``` What do you think this code will output when executed as part of a main() method? Let's take a look: ``` 1 2 3 4 5 Time: 25 seconds ``` As you might expect, the results are ordered and predictable because we are using a serial stream. It also took around 25 seconds to process all five results, one at a time. What happens if we replace line 12 with one that uses a parallelStream()? The following is some sample output: ``` 3 2 1 5 4 Time: 5 seconds ``` What about the time required? In this case, our system had enough CPUs for all of the tasks to be run concurrently. If you ran this same code on a computer with fewer processors, it might output 10 seconds, 15 seconds, or some other value. The key is that we've written our code to take advantage of parallel processing when available, so our job is done.

Question 90

> [!NOTE] > The Stream interface includes a method ***`isParallel()`*** that can be used to test whether the instance of a stream supports parallel processing. > Some operations on streams preserve the parallel attribute, while others do not.

Answer 90

isParallel()

Question 91

Ordering Results

Answer 91

If your stream operation needs to guarantee ordering and you're not sure if it is serial or parallel, you can replace line 14 with one that uses ***`forEachOrdered()`***: ***`14: .forEachOrdered(s -> System.out.print(s + " "));`*** This outputs the results in the order in which they are defined in the stream: ``` 1 2 3 4 5 Time: 5 seconds ``` While we've lost some of the performance gains of using a parallel stream, our map() operation can still take advantage of the parallel stream.

Question 92

Processing Parallel Reductions

Answer 92

A ***`parallel reduction`*** is a reduction operation applied to a parallel stream. The results for parallel reductions can differ from what you expect when working with serial streams.

Question 93

Performing Order-Based Tasks

Answer 93

Since order is not guaranteed with parallel streams, methods such as findAny() on parallel streams may result in unexpected behavior.

Question 94

Consider the following example: ``` System.out.print(List.of(1,2,3,4,5,6) .parallelStream() .findAny() .get()); ```

Answer 94

The JVM allocates a number of threads and returns the value of the first one to return a result, which could be 4, 2, and so on. While neither the serial nor the parallel stream is guaranteed to return the first value, the serial stream often does. With a parallel stream, the results are likely to be more random. What about operations that consider order, such as **findFirst()**, **limit()**, and **skip()**? **Order is still preserved, but performance may suffer on a parallel stream as a result of a parallel processing task being forced to coordinate all of its threads in a synchronized-like fashion.** On the plus side, the results of ordered operations on a parallel stream will be consistent with a serial stream. For example, calling skip(5).limit(2).findFirst() will return the same result on ordered serial and parallel streams.

Question 95

Creating Unordered Streams

Answer 95

All of the streams you have been working with are considered ordered by default. It is possible to create an unordered stream from an ordered stream, similar to how you create a parallel stream from a serial stream. ***`List.of(1,2,3,4,5,6).stream().unordered();`*** This method does not reorder the elements; it just tells the JVM that if an order-based stream operation is applied, the order can be ignored. For example, calling skip(5) on an unordered stream will skip any 5 elements, not necessarily the first 5 required on an ordered stream. For serial streams, using an unordered version has no effect. But **on parallel streams, the results can greatly improve performance.** ***`List.of(1,2,3,4,5,6).stream().unordered().parallel();`*** Even though **unordered streams will not be on the exam**, if you are developing applications with parallel streams, you should know when to apply an unordered stream to improve performance.

Question 96

Combining Results with reduce()

Answer 96

the **stream** operation **reduce()** combines a stream into a single object. * Recall that the **first parameter** to the reduce() method is called the ***`identity`***, * the **second parameter** is called the ***`accumulator`***, * and the **third parameter** is called the ***`combiner`***. ``` U reduce(U identity, BiFunction accumulator, BinaryOperator combiner) ```

Question 97

>[!NOTES] > While this is not in scope for the exam, the accumulator and combiner must be associative, non-interfering, and stateless. > Don't panic; you don't need to know advanced math terms for the exam!

Answer 97

accumulator and combiner

Question 98

We can concatenate a list of char values using the reduce() method, as shown in the following example: ``` System.out.println(List.of('w', 'o', 'l', 'f') .parallelStream() .reduce("", (s1,c) -> s1 + c, (s2,s3) -> s2 + s3)); // wolf ```

Answer 98

The naming of the variables in this stream example is not accidental. We used c for char, whereas s1, s2, and s3 are String values. On parallel streams, the reduce() method works by applying the reduction to pairs of elements within the stream to create intermediate values and then combining those intermediate values to produce a final result. Put another way, in a serial stream, wolf is built one character at a time. In a parallel stream, the intermediate values wo and lf are created and then combined. With parallel streams, we now have to be concerned about order. What if the elements of a string are combined in the wrong order to produce wlfo or flwo? **The Stream API prevents this problem while still allowing streams to be processed in parallel, as long as you follow one simple rule: make sure that the *`accumulator`* and *`combiner`* produce the same result regardless of the order they are called in.** With parallel streams, though, order is no longer guaranteed, and any argument that violates these rules is much more likely to produce side effects or unpredictable results.

Question 99

``` System.out.println(List.of(1,2,3,4,5,6) .parallelStream() .reduce(0, (a, b) -> (a - b))); // PROBLEMATIC ACCUMULATOR ``` It may output -21, 3, or some other value.

Answer 99

We can omit a combiner parameter in these examples, as the accumulator can be used when the intermediate data types are the same. It may output -21, 3, or some other value.

Question 100

Selecting a reduce() Method

Answer 100

Although the one- and two-argument versions of reduce() support parallel processing, it is recommended that you use the three-argument version of reduce() when working with parallel streams. Providing an explicit combiner method allows the JVM to partition the operations in the stream more efficiently.

Question 101

Combining Results with collect()

Answer 101

Stream API includes a three-argument version of ***`collect()`*** that takes ***`accumulator`*** and ***`combiner`*** operators along with a ***`supplier`*** operator instead of an identity. ``` R collect(Supplier supplier, BiConsumer accumulator, BiConsumer combiner) ``` the ***`accumulator`*** and ***`combiner`*** operations must be able to process results ***`in any order`***. ``` Stream stream = Stream .of("w", "o", "l","f") .parallel(); SortedSet set = stream .collect(ConcurrentSkipListSet::new, Set::add, Set::addAll); System.out.println(set); // [f, l, o, w] ``` elements in a ***`ConcurrentSkipListSet`*** are ***`sorted`*** according to their natural ordering. You **should use a concurrent collection** to combine the results, ensuring that the results of concurrent threads do **not cause a ConcurrentModificationException**. Performing parallel reductions with a collector requires additional considerations. For example, if the collection into which you are inserting is an **ordered data** set, such as a **List**, **the elements in the resulting collection must be in the same order**, regardless of whether you use a serial or parallel stream. This **may reduce performance**, though, as **some operations cannot be completed in parallel**.

Question 102

Performing a Parallel Reduction on a Collector

Answer 102

Every **Collector** instance defines a **characteristics()** method that returns a set of **Collector.Characteristics** attributes. When using a **Collector** to perform a **parallel reduction**, a number of properties must hold **true**. Otherwise, the **collect()** operation will execute in a **single-threaded** fashion.

Question 103

Requirements for Parallel Reduction with collect()

Answer 103

* The **stream is parallel**. * The parameter of the **collect()** operation has the **Characteristics.CONCURRENT characteristic**. * Either the **stream is unordered** or the **collector has the characteristic Characteristics.UNORDERED**. For example, while **Collectors.toSet()** does have the **UNORDERED** characteristic, it does not have the **CONCURRENT** characteristic. Therefore, the following is not a parallel reduction even with a parallel stream: ***`parallelStream.collect(Collectors.toSet()); // Not a parallel reduction`*** The ***`Collectors`*** class includes **two sets of *`static methods `*for retrieving collectors**, ***`toConcurrentMap()`*** and ***`groupingByConcurrent()`***, both of which are **UNORDERED** and **CONCURRENT**. These methods produce **Collector** instances capable of performing **parallel reductions** efficiently. Like their nonconcurrent counterparts, there are overloaded versions that take additional arguments. ``` Stream ohMy = Stream.of("lions", "tigers", "bears").parallel(); ConcurrentMap map = ohMy .collect(Collectors.toConcurrentMap(String::length, k -> k, (s1, s2) -> s1 + "," + s2)); System.out.println(map); // {5=lions,bears,6=tigers} System.out.println(map.getClass()); // java.util.concurrent.ConcurrentHashMap ``` groupingBy() example ``` var ohMy = Stream.of("lions", "tigers", "bears").parallel(); ConcurrentMap> map = ohMy.collect(Collectors.groupingByConcurrent(String::length)); System.out.println(map); // {5=[lions, bears],6=[tigers]} ```

Question 104

Avoiding Stateful Streams

Answer 104

***`Side effects`*** can appear in **parallel streams** if your **lambda expressions are stateful**. A ***`stateful lambda expression`*** is one whose result depends on any state that might change during the execution of a pipeline. For example, the following method that filters out even numbers is stateful: ``` public List addValues(IntStream source) { var data = Collections.synchronizedList(new ArrayList()); source.filter(s -> s % 2 == 0) .forEach(i -> { data.add(i); }); // STATEFUL: DON'T DO THIS! return data; } ``` Let's say this method is executed with a serial stream: ``` var list = addValues(IntStream.range(1, 11)); System.out.print(list); // [2, 4, 6, 8, 10] ``` But what if someone else passes in a parallel stream? ``` var list = addValues(IntStream.range(1, 11).parallel()); System.out.print(list); // [6, 8, 10, 2, 4] ``` Oh, no: our results no longer match our input order! The problem is that our lambda expression is stateful and modifies a list that is outside our stream. We can fix this solution by rewriting our stream operation to be stateless: ``` public List addValuesBetter(IntStream source) { return source.filter(s -> s % 2 == 0) .boxed() .collect(Collectors.toList()); } ``` This method processes the stream and then collects all the results into a new list. It produces the same ordered result on both serial and parallel streams. It is strongly recommended that you avoid stateful operations when using parallel streams, to remove any potential data side effects. In fact, they should be avoided in serial streams since doing so limits the code's ability to someday take advantage of parallelization.

Question 105

**Summary**

Answer 105

* how to create and **define the *`thread`*'s work using a *`Runnable`* instance** * how to **pause** and **interrupt** the **thread**. * When working with the **Concurrency API,** you should also know how to create threads using **Callable lambda expressions.** * how to concurrently execute tasks using **ExecutorService** like a pro. * know which **ExecutorService** instances are available, including **scheduled** and **pooled** services. * **Thread-safety** is about protecting data from being corrupted by multiple threads modifying it at the same time. * Java offers many tools to keep data safe, including atomic classes, **synchronized methods/blocks**, the **Lock framework**, and **CyclicBarrier**. * Concurrency API also includes numerous collection classes that handle multithreaded access for you. You should be familiar with the **concurrent collections**, including the **CopyOnWrite** classes, which create a new underlying structure any time the underlying collection is modified. * **potential threading issues** can arise. **(For the exam, you need to know only the basic theory behind these concepts.)** * **Deadlock, starvation, and livelock** can result in programs that appear stuck, * **race conditions** can result in unpredictable data * **parallel streams** * how to use them to perform **parallel** **decompositions** and **reductions**. * can greatly improve the performance * can also cause unexpected results since the processing is no longer ordered. * Remember to avoid stateful lambda expressions, especially when working with parallel streams.

Question 106

**Exam Essentials**

Answer 106

* **Be able to write thread-safe code.** Thread-safety is about protecting shared data from concurrent access. A monitor can be used to ensure that only one thread processes a particular section of code at a time. In Java, monitors can be implemented with a **synchronized block or method** or using an instance of **Lock**. **ReentrantLock** has a number of advantages over using a synchronized block, including the ability to **check whether a lock is available without blocking it**, as well as supporting the **fair acquisition of locks**. To achieve synchronization, two or more threads must coordinate on the same shared object. * **Be able to apply the atomic classes.** An **atomic** operation is one that occurs without interference from another thread. The Concurrency API includes a set of **atomic classes** that are similar to the primitive classes, except that they ensure that operations on them are performed atomically. Know the **difference between an atomic variable and one marked with the volatile modifier.** * **Create concurrent tasks with a thread executor service using Runnable and Callable.** An ExecutorService creates and manages a single thread or a pool of threads. Instances of Runnable and Callable can both be submitted to a thread executor and will be completed using the available threads in the service. Callable differs from Runnable in that Callable returns a generic data type and can throw a checked exception. A ScheduledExecutorService can be used to schedule tasks at a fixed rate or with a fixed interval between executions. * **Be able to use the concurrent collection classes.** The Concurrency API includes numerous collection classes that include built-in support for multithreaded processing, such as ConcurrentHashMap. It also includes a class CopyOnWriteArrayList that creates a copy of its underlying list structure every time it is modified and is useful in highly concurrent environments. * **Identify potential threading problems.** Deadlock, starvation, and livelock are three threading problems that can occur and result in threads never completing their task. Deadlock occurs when two or more threads are blocked forever. Starvation occurs when a single thread is perpetually denied access to a shared resource. Livelock is a form of starvation where two or more threads are active but conceptually blocked forever. Finally, race conditions occur when two threads execute at the same time, resulting in an unexpected outcome. * **Understand the impact of using parallel streams.** The Stream API allows for the easy creation of parallel streams. Using a parallel stream can cause unexpected results, since the order of operations may no longer be predictable. Some operations, such as reduce() and collect(), require special consideration to achieve optimal performance when applied to a parallel stream.

Answer 107

public final void notify() Wakes up a single thread that is waiting on this object's monitor. If any threads are waiting on this object, one of them is chosen to be awakened. The choice is arbitrary and occurs at the discretion of the implementation. A thread waits on an object's monitor by calling one of the wait methods. The awakened thread will not be able to proceed until the current thread relinquishes the lock on this object. The awakened thread will compete in the usual manner with any other threads that might be actively competing to synchronize on this object; for example, the awakened thread enjoys no reliable privilege or disadvantage in being the next thread to lock this object. This method should only be called by a thread that is the owner of this object's monitor. A thread becomes the owner of the object's monitor in one of three ways: By executing a synchronized instance method of that object. By executing the body of a synchronized statement that synchronizes on the object. For objects of type Class, by executing a synchronized static method of that class. Only one thread at a time can own an object's monitor. public final void notifyAll() Wakes up all threads that are waiting on this object's monitor. A thread waits on an object's monitor by calling one of the wait methods. The awakened threads will not be able to proceed until the current thread relinquishes the lock on this object. The awakened threads will compete in the usual manner with any other threads that might be actively competing to synchronize on this object; for example, the awakened threads enjoy no reliable privilege or disadvantage in being the next thread to lock this object. This method should only be called by a thread that is the owner of this object's monitor. See the notify method for a description of the ways in which a thread can become the owner of a monitor. public final void wait() throws InterruptedException Causes the current thread to wait until it is awakened, typically by being notified or interrupted. In all respects, this method behaves as if wait(0L, 0) had been called. See the specification of the wait(long, int) method for details.