Laboratory Assignments:

• 1 Working with I/O Streams
• 2 Working with Java Collections
• 3 Working with XML Documents
• 4 Development of Graphical User Interface Applications
• 5 Working with Java FX Graphics
• 6 Working with Stream API and Threads

Laboratory Training 6

Working with Stream API and Threads

1 Training Assignment

1.1 Individual Task

It is necessary to implement the individual task of the previous laboratory training, adding the search for its roots (with some precision). The algorithm for finding the roots is to sequentially examine with a certain step of the interval points, find the intervals in which the function changes the sign and deduces the arithmetic mean arithmetic and the ends of the intervals that are the roots of the equation. The calculation should be done in a separate thread.

The user should be able to enter the parameters a and b, range of search and accuracy, and get the graph and roots. Implement the ability to pause, restore the thread, and also complete termination and re-calculation with new data.

1.2 Calculating π in a Separate Thread (Additional Task)

Implement the program of calculation π with accuracy up to a given ε as the sum of a sequence:

The calculation should be done in a separate thread. When performing the calculation, you should give the user the opportunity to enter a request for the number of calculated elements of the sum.

1.3 Working with BlockingQueue

Create a console program in which one thread adds integers to the BlockingQueue queue, and the other calculates their arithmetic mean.

1.4 Working with Java 8 Stream API

Create a console program that displays all positive integers whose sums of digits are equal to the specified value. Use streams of elements.

1.5 Creating a GUI Application for Getting Prime Factors of Numbers (Additional Task)

Implement JavaFX GUI application, in which the user inputs a range of numbers (from and to), and in the window the numbers and their prime factors are displayed. Realize the ability to pause, resume the stream, and also complete termination and re-calculation with new data.

2 Instructions

2.1 Working with Threads

2.1.1 Overview

A process is a copy of a computer program that is loaded in memory is executed. In modern operating systems, a process can be executed in parallel with other processes. A process has its own address space, and this space is physically unavailable for other processes.

A thread (execution thread) is a separate subtask that can be executed in parallel with other subtasks (threads) within a single process. Each process contains at least one thread called the main thread. All threads created by the process are executed in the address space of this process and have access to process resources. Creating threads is essentially less resource-intensive operation than creating new processes. Threads are sometimes referred to as lightweight processes.

If the process has created multiple threads, then all of them are executed in parallel, while the time of the central processor (or several central processors in multiprocessor systems) is distributed among these threads. Distribution of the CPU time is performed by a special module of the operating system, the so-called scheduler. The scheduler in turn handles the management of individual thread, so even in the one-processor system an illusion of parallel running of threads is created. Time distribution is performed by interruptions of the system timer.

Threads are used to implement independent subtasks within a single process in order to create background processes, simulate the parallel execution of certain actions, or enhance the user interface's convenience.

2.1.2 Low-level Thread Tools

There are two ways to create a thread object using the Thread class:

• creation of a new class derived from the java.lang.Thread class and creating an object of a new class;
• creation of a class that implements the java.lang.Runnable interface; an object of this class is passed to the java.lang.Thread class constructor as an argument.

When creating a derived class from Thread, it is necessary to override its run() method. After creating the thread object, it should be started using the start() method. This method performs some initialization and then calls run(). The following example creates a separate execution thread that outputs the next integer from 1 to 40:

package ua.in.iwanoff.java.sixth;

public class ThreadTest extends Thread {
  
    @Override
    public void run() {
        for (int counter = 1; counter <= 40; counter++) {
            try {
                Thread.sleep(1000); // Delay time in milliseconds
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(counter);
        }
    }

    public static void main(String[] args) {
        new ThreadTest().start();
        // You can add actions that run in parallel with the run() method.
    }

}

Calling the sleep() method causes the thread to be suspended at the specified number of milliseconds. Calling the sleep() method requires catching of the InterruptedException exception that is thrown if this thread is interrupted by another thread. A similar example is with the creation of a class object that implements the Runnable interface. In this case, you also need to run the thread using the start() method:

package ua.in.iwanoff.java.sixth;

public class RunnableTest implements Runnable {
  
    @Override
    public void run() {
        for (int counter = 1; counter <= 40; counter++) {
            try {
                Thread.sleep(1000);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(counter);
        }
    }  

    public static void main(String[] args) {
        new Thread(new RunnableTest()).start();
        // You can add actions that run in parallel with the run () method
    }

}

The second approach is better, since in this case we are free to choose the base class.

You can assign names to individual threads (the second parameter of the constructor) and get these names using the getName() function.

Any thread can be in several standard states. The thread receives the state "new" (Thread.State.NEW) when the thread object is created. Calling the start() method moves the thread to the state "runnable" (Thread.State.RUNNABLE). There are states "blocked" (Thread.State.BLOCKED), "blocked in time", or "in the standby mode" (Thread.State.TIMED_WAITING), "waiting", or "not working" (Thread.State.WAITING) and "complete" (Thread.State.TERMINATED). When creating a thread, it gets a "new" state and is not executed. You can call the getState() method to get the state of a thread. The following example shows some states of the thread within its life cycle:

package ua.in.iwanoff.java.sixth;

public class StateTest {

    public static void main(String[] args) throws InterruptedException {
        Thread testThread = new Thread();
        System.out.println(testThread.getState()); // NEW
        testThread.start();
        System.out.println(testThread.getState()); // RUNNABLE
        testThread.interrupt(); // interrupt the thread
        Thread.sleep(100);      // it takes time to complete the thread
        System.out.println(testThread.getState()); // TERMINATED
    }
}

The wait(long timeout) method, like sleep(long timeout) method, allows you to pause the thread for a specified number of milliseconds. When executing this method, InterruptedException can also be thrown. The use of these methods allows you to switch the thread to standby (TIMED_WAITING). Unlike sleep(), the ability to work after calling the wait() method can be restored using notify() or notifyAll() methods.

The wait() method can be called without parameters. The thread thus gets into the "disable" state (WAITING).

A less reliable alternative way to stop a thread for a while is calling the yield() method, which should stop the thread for a certain amount of time, so that other threads can perform their actions.

After the run() method is completed, the thread finishes its work and passes to the passive state (TERMINATED).

In earlier versions of Java, for the forced stop, as well and temporary suspension of the thread and subsequent resumption, the use class Thread methods stop(), suspend() and resume() was provided. The stop(), suspend() and resume() methods are considered undesirable to use (deprecated methods), since they provoke the creation of unreliable and difficult to debug code. Furthermore, the use of suspend() and resume() can trigger deadlocks.

You can interrupt the thread that is in the standby state (Thread.State.TIMED_WAITING) by calling the interrupt() method. In this case, the InterruptedException is thrown. We can modify the previous example so that it allows us to interrupt the thread:

package ua.in.iwanoff.java.sixth;

import java.util.Scanner;

public class InterruptTest implements Runnable {

    @Override
    public void run() {
        for (int counter = 1; counter <= 40; counter++) {
            try {
                Thread.sleep(1000);
            }
            catch (InterruptedException e) {
                System.out.println("Thread interrupted.");
                break;
            }
            System.out.println(counter);
        }
    }

    @SuppressWarnings("resource")
    public static void main(String[] args) {
        Thread thread = new Thread(new InterruptTest());
        thread.start();
        System.out.println("Press Enter to interrupt");
        new Scanner(System.in).nextLine();
        thread.interrupt();
    }

}

The join() method allows one thread to wait for the completion of another. Calling this method inside thread t1 for thread t2 will stop the current thread (t1) until the completion of t2, as shown in the following example:

package ua.in.iwanoff.java.sixth;

public class JoinTest {
    static Thread t1, t2;
  
    static class FirstThread implements Runnable {
        @Override
        public void run() {
            try {
                System.out.println("The first thread started.");
                Thread.sleep(1000);
                System.out.println("The primary work of the first thread has been completed.");
                t2.join();
                System.out.println("The first thread completed.");
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class SecondThread implements Runnable {
        @Override
        public void run() {
            try {
                System.out.println("The second thread started.");
                Thread.sleep(3000);
                System.out.println("The second thread completed.");
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        t1 = new Thread(new FirstThread());
        t2 = new Thread(new SecondThread());
        t1.start();
        t2.start();
    }

}

The order of message output will be as follows:

The first thread started.
The second thread started.
The primary work of the first thread has been completed.
The second thread completed.
The first thread completed.

Like sleep(), join() responds to interrupt, therefore, it is necessary to catch InterruptedException.

You can create a thread whose work is interrupted when the thread that created it terminates. In order to create such a thread, the setDaemon() method is called with the true parameter. The setDaemon() method must be called after the thread was created, but before the start() method is called. You can check whether there is a thread by the daemon or not using isDaemon() method. If the daemon thread creates other threads, they will also receive the status of the daemon thread.

package ua.in.iwanoff.java.sixth;

public class Th extends Thread {
    public void run() {
        try {
            if (isDaemon()) {
                System.out.println("start of the daemon thread");
                sleep(1000); 
            }
            else {
                System.out.println("start of the regular thread");
            }
        }
        catch (InterruptedException e) {
            System.err.print("Error" + e);
        }
        finally {
            if (!isDaemon())
                System.out.println("regular thread termination");
            else
                System.out.println("daemon thread termination");
        }
    }

    public static void main(String[] args) {
        Th usual = new Th();
        Th daemon = new Th();
        daemon.setDaemon(true);
        daemon.start();
        usual.start();
        System.out.println("the last line of the main() method");
    }
}

After compiling and running, it is possible that the following message output sequence will be obtained:

start of the daemon thread
the last line of the main() method
start of the regular thread
regular thread termination

The daemon thread (due to the call to the sleep(10000) method) did not have time to complete the execution of its code before the main application thread associated with the main() method was completed. The basic property of daemon threads is the ability of the main application thread to complete execution with the end of the code of the main() method, ignoring the fact that the daemon thread is still running. If you set the delay time also for the main() thread, then the daemon thread can have time to complete its execution before the main thread ends.

start of the daemon thread
start of the regular thread
regular thread termination
daemon thread termination
the last line of the main() method

In the case of multiple threads, it is important to eliminate the possibility of using one resource by two threads. If the data members of a class are declared as private and access to this memory area is possible only through methods, then collisions can be avoided by declaring these methods as synchronized. At the same time, only one thread can call a synchronized method for a specific object. Such a method or code fragment is referred to as a critical section.

The concept of the monitor is used to provide access to such code. A monitor is a certain object that blocks a code during execution by a certain thread. If the synchronized keyword is located before of a function name, the object for which this method is called (this) considered as monitor. In this case, after at least one synchronized method is called, all methods with the synchronized modifier are blocked as well.

Consider an example. The Adder class allows you to add integers to some accumulator. The AdderThread class, which implements the execution thread, provides the addition of consecutive five integer values. In the main() function of the AdderTest class, we create two threads and run them:

package ua.in.iwanoff.java.sixth;

class Adder {
    long sum = 0;

    public void add(long value) {
        this.sum += value;
        System.out.println(Thread.currentThread().getName() + " " + sum);
    }
}

class AdderThread extends Thread {
    private Adder counter = null;

    public AdderThread(String name, Adder counter) {
        super(name);
        this.counter = counter;
    }

    public void run() {
        try {
            for (int i = 0; i < 5; i++) {
                counter.add(i);
                Thread.sleep(10);
            }
        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class AdderTest {

    public static void main(String[] args) {
        Adder adder = new Adder();
        Thread threadA = new AdderThread("A", adder);
        Thread threadB = new AdderThread("B", adder);
        threadA.start();
        threadB.start();
    }

}

Tthe program that was started several times will produce various intermediate results due to the incorrect operation of two threads with the same object. To fix the situation, the synchronized modifier should be added before the add() method:

  synchronized public void add(long value) {
      this.sum += value;
      System.out.println(Thread.currentThread().getName() + " " + sum);
  }

Sometimes declaring the entire method as synchronized is inconvenient, since in this case other threads could execute other blocking methods that are not related to this one. To solve this problem, use a block that starts with the word synchronized and contains in brackets the name of the object responsible for the lock. Any object derived from java.lang.Object (that supports the necessary methods) can act as a monitor. The following example creates a special object for blocking:

  public void add(long value) {
      Object lock = new Object();
      synchronized (lock) {
          this.sum += value;
          System.out.println(Thread.currentThread().getName() + " " + sum);
      }
  }

If the amount of memory occupied by the object does not exceed 32 bits, writing and reading do not require synchronization. Such objects are called atomic.

When exchanging data through fields, the volatile keyword should be used before describing the corresponding field. This is due to the presence of a data caching mechanism for each thread separately, and the value from this cache may differ for each of them. Declaring a field with the volatile keyword disables such caching for it:

volatile long sum = 0;

The wait() method, called inside a synchronized block or method, stops the execution of the current thread and releases the captured object, including the lock object, from the lock. You can return blocking of a thread object by calling the notify() method for a specific thread or by calling notifyAll() for all threads. A call can only be made from another thread that blocked, in turn, the specified object.

The setPriority() function allows you to modify priority of a thread. It is not recommended to set highest priority (Thread.MAX_PRIORITY constant).

2.1.3 Use Thread Execution in JavaFX Applications

The work of graphical user interface applications is usually associated with threads. After starting the JavaFX application, the application thread is automatically created to handle the events. Only this thread can interact with windows and visual components. Attempting two threads simultaneously to control the visualization of components and information on the window's surface can lead to chaotic appearance and unpredictable behavior of the visual components. However, thanks to the multithreading, you can provide better controllability of the application. In addition, it is advisable to take in a separate thread some actions that are performed for a long time. This will provide an opportunity to monitor the process, pause and continue it, change settings, etc.

JavaFX tools allow you to execute code in the thread, which is responsible for receiving and processing events. The javafx.application.Platform.runLater() method receives an object that implements the Runnable functional interface. This object enters the event processing thread, where the object becomes in the queue and, when the opportunity arises, its run() method is executed. In this method, it is advisable to implement interaction with visual components. Example 3.6 demonstrates the possibility of calling this method from a child thread.

There are also high-level mechanisms for working with threads in JavaFX, presented in the javafx.concurrent package: the abstract Task class, derived from java.util.concurrent.FutureTask, in turn provides for the creation of a derived class that implements the call() method. This method implements interaction with visual components. In addition, this class provides the ability to directly bind some components (for example, ProgressBar) to the task.

2.2 Use of Thread-Safe Collections

2.2.1 Use of Synchronized Analogues of Standard Collections

The java.util.Collections class can be used to obtain thread-safe clones of existing collections. These are static functions: synchronizedCollection(), synchronizedList(), synchronizedMap(), synchronizedSet(), synchronizedSortedMap(), and synchronizedSortedSet(). Each of these functions takes as a parameter the corresponding non-synchronized collection and returns a collection of the same type, all methods of which (except for iterators) are defined as synchronized. Accordingly, all operations, including reading, use a lock that reduces performance and restricts the use of the appropriate collections.

Iterators of standard collections implemented in java.util implement a so-called fail-fast behavior: if some object was changed bypassing the collection iterator throws the ConcurrentModificationException exception, and in terms of multithreading, the behavior of the iterator may be unpredictable: an exception may not be thrown immediately or not thrown at all. Reliable and work with collection elements are provided by special container classes defined in java.util.concurrent. These are types such as CopyOnWriteArrayList, CopyOnWriteArraySet, ConcurrentMap (interface), ConcurrentHashMap (implementation), ConcurrentNavigableMap (interface), and ConcurrentSkipListMap (implementation).

The CopyOnWriteArrayList class that implements the List interface, creates a copy of the array for all operations that change the value of items. The iterator also works with its copy. Such behavior of the iterator is called fail-safe behavior.

The ConcurrentHashMap and ConcurrentSkipListMap classes provide associative arrays that do not require blocking of all operations. The ConcurrentHashMap class organizes a data structure divided into segments. Blocking is done at the level of one segment. Due to the improved hash function, the probability of applying two threads to one segment is significantly reduced. The ConcurrentSkipListMap class implements storage using a special data structure, the skip list. The elements of the associative array are sorted by the keys. Example 2.2 demonstrates work with ConcurrentSkipListMap.

2.2.2 Work with Special Collections that Use Locks

The java.util.concurrent package provides a collection of multi-threaded security collections. These are generic types such as BlockingQueue (interface) and its standard implementations (ArrayBlockingQueue, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue, DelayQueue), as well as derived from it TransferQueue interface (LinkedTransferQueue implementation) and LinkedBlockingDeque interface (standard implementation is LinkedBlockingDeque).

The BlockingQueue interface represents a queue that allows you to add and get items safely from the perspective of multithreading. A typical use of the BlockingQueue is adding objects in one thread, and getting objects in another thread.

The thread that adds objects (first thread) continues to add them to the queue until some upper limit of the permissible number of elements is reached. In this case, when trying to add new objects, the first thread will be blocked until the thread that receives the elements (second thread) does not remove at least one item from the queue. The second thread can remove and use objects from the queue. If, however, the second thread tries to get an empty queue object, this thread will be blocked until the first thread adds a new object.

The BlockingQueue interface inherits java.util.Queue, so it supports add(), remove(), element() (with exception throwing), plus offer(), poll() and peek() (with special returning value). In addition, blocking methods are put() (for adding) and take() (for extraction) are declared. There are also methods that use time-locking: offer() and poll() with additional parameters (time interval and time unit).

You cannot add null to BlockingQueue. This attempt leads to the throw of NullPointerException.

The most popular implementation of BlockingQueue is ArrayBlockingQueue. In order to store references to objects in this implementation, a conventional array of limited length is used. This length cannot be changed after creating a queue. All class constructors receive this array length as the first parameter. In the example below, two threads are created - Producer and Consumer. The producer adds integers with a delay between additions, and the consumer withdraws them

package ua.in.iwanoff.java.sixth;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

class Producer implements Runnable {
    private BlockingQueue<Integer> queue;
    int countToAdd;

    public Producer(BlockingQueue<Integer> queue, int countToAdd) {
        this.queue = queue;
        this.countToAdd = countToAdd;
    }

    public void run() {
        // We try to add numbers:
        try {
            for (int i = 1; i <= countToAdd; i++) {
                queue.put(i);
                System.out.printf("Added: %d%n", i);
                Thread.sleep(100);
            }
        }
        catch (InterruptedException e) {
            System.out.println("Producer interrupted");
        }
    }
}

class Consumer implements Runnable {
    private BlockingQueue<Integer> queue;
    int countToTake;

    public Consumer(BlockingQueue<Integer> queue, int countToTake) {
        this.queue = queue;
        this.countToTake = countToTake;
    }

    public void run() {
        // We withdraw numbers:
        try {
            for (int i = 1; i <= countToTake; i++) {
                System.out.printf("Taken by customer: %d%n", queue.take());
            }
        }
        catch (InterruptedException e) {
            System.out.println("Consumer interrupted");
        }
    }
}

public class BlockingQueueDemo {

    public static void main(String[] args) throws InterruptedException {
        // Create a queue of 10 items:
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
    
        // Create and launch two threads - for writing and reading:
        Thread producerThread = new Thread(new Producer(queue, 100));
        Thread consumerThread = new Thread(new Consumer(queue, 4));
        producerThread.start();
        consumerThread.start();
    
        // Wait 10 seconds and interrupt the first thread:
        Thread.sleep(10000);
        producerThread.interrupt();
    }
  
}

When you start the program, you can see that the producer thread consistently adds numbers to the queue. The first four numbers are withdrawn by the consumer thread, while the consumer thread is blocked each time, waiting for the next number. Since only 10 numbers can fit in the queue, the queue quickly becomes full after the consumer thread stops withdrawing items. After ten seconds, the main thread interrupts the producer thread.

The implementation of LinkedBlockingQueue differs from ArrayBlockingQueue in its internal representation: a linked list is used. The maximum size is no longer significant (Integer.MAX_VALUE by default), but it can also be specified in the constructor.

The PriorityBlockingQueue class uses the same ordering rules as java.util.PriorityQueue.

The SynchronousQueue class represents a queue that can contain only one element. The thread that added the item is blocked until another thread receives it from the queue. If the second thread cannot get an element (the queue is empty), this thread is also blocked until a new element is added by the first thread.

The implementation of DelayQueue allows you to remove only those elements that have been expired for some time (delay). The elements of this queue should implement the interface java.util.concurrent.Delayed. This interface defines the getDelay() method, which returns the delay remaining for the object. From the queue, the element for which the delay time has passed before the other is removed in the first place. If none of the elements is delayed, the poll() method returns null.

The BlockingDeque interface, derived from the BlockingQueue, additionally supports the aaddFirst(), addLast(), takeFirst(), and takeLast() operations, which are inherent in double-ended queues. The LinkedBlockingDeque class offers a standard implementation of this interface.

2.3 Working with Containers and Stream API in Java 8

2.3.1 Use of Optional Class

Optional values are containers for values that can sometimes be empty. Traditionally, for uncertain values, the value null was used. Using the null constant can be inconvenient because it can lead to the throwing of the NullPointerException exception, which complicates debugging and maintenance of the program.

A generic Optional class allows you to save the reference value to some object, and also to check if the value is set. For example, some method returns a numeric value, but for some arguments the argument cannot return something definite. This method can return an Optional type object and this value can then be used in the calling function. Suppose some function calculates and returns the reciprocal value and returns an "empty" object if the argument is equal to 0. The main() function performs the calculation for the reciprocal numbers for array items:

package ua.in.iwanoff.java.sixth;

import java.util.Optional;

public class OptionalDemo {

    static Optional<Double> reciprocal(double x) {
        Optional<Double> result = Optional.empty();
        if (x != 0) {
            result = Optional.of(1 / x);
        }
        return result;
    }
  
    public static void main(String[] args) {
        double[] arr = { -2, 0, 10 };
        Optional<Double> y;
        for (double x : arr) {
            System.out.printf("x = %6.3f  ", x);
            y = reciprocal(x);
            if (y.isPresent()) {
                System.out.printf("y = %6.3f%n", y.get());
            }
            else {
                System.out.printf("The value cannot be calculated%n");
            }
        }
    }

}

If you do not check for the presence of a value (isPresent()), a java.util.NoSuchElementException will be thrown when you try to call the get() function for the "empty" value. It can be caught instead of calling the isPresent() function.

In some cases, the null value should not be considered as possible. In this case, use ofNullable() to store the value. For example:

Integer k = null;
Optional<Integer> opt = Optional.ofNullable(k);
System.out.println(opt.isPresent()); // false

Assume that if the reciprocal() function described before does not return a value in the case of division by zero, the y variable should be assigned to 0. Traditionally, in this case, the if ... else statement or conditional operation is used:

y = reciprocal(x);
double z = y.isPresent() ? y.get() : 0;

The orElse() method allows you to make the code more compact:

double z = reciprocal(x).orElse(0.0);

In addition to the generic Optional class, you can also use classes optimized for primitive types: OptionalInt, OptionalLong, OptionalBoolean, etc. The previous example (calculation of the reciprocal value) could be realized in this way (using OptionalDouble):

package ua.in.iwanoff.java.sixth;

import java.util.OptionalDouble;

public class OptionalDoubleDemo {

    static OptionalDouble reciprocal(double x) {
        OptionalDouble result = OptionalDouble.empty();
        if (x != 0) {
            result = OptionalDouble.of(1 / x);
        }
        return result;
    }

    public static void main(String[] args) {
        double[] arr = {-2, 0, 10};
        OptionalDouble y;
        for (double x : arr) {
            System.out.printf("x = %6.3f  ", x);
            y = reciprocal(x);
            if (y.isPresent()) {
                System.out.printf("y = %6.3f%n", y.getAsDouble());
            }
            else {
                System.out.printf("The value cannot be calculated%n");
            }
        }
    }

}

As can be seen from the example, the result can be obtained using the getAsDouble() method instead of get().

2.3.2 Additional Features for Standard Containers

Standard interfaces of the java.util package are supplemented with methods that focus on using lambda expressions and references to methods. To ensure compatibility with previous versions, new interfaces provide the default implementation of the new methods. In particular, the Iterable interface defines the forEach() method, which allows you to perform some actions in the loop that do not change the elements of the collection. You can specify an action using a lambda expression or a reference to a method. For example:

public class ForEachDemo {
    static int sum = 0;
    
    public static void main(String[] args) {
        Iterable<Integer> numbers = new ArrayList(Arrays.asList(2, 3, 4));
        numbers.forEach(n -> sum += n);
        System.out.println(sum);
    }
}

In the above example, the sum of collection elements is calculated. The variable that holds the sum is described as a static class field, since lambda expressions cannot change local variables.

The Collection interface defines the removeIf() method, which allows you to remove items from the collection if items match a certain filter rule. In the following example, odd items are removed from the collection of integers. The forEach() method is used for columnwise output the collection items:

Collection<Integer> c = new ArrayList(Arrays.asList(2, 4, 11, 8, 12, 3));
c.removeIf(k -> k % 2 != 0);
// The rest of the items are displayed columnwise:
c.forEach(System.out::println);

The List interface provides methods replaceAll() and sort(). The second one can be used instead of the analogous static method of the Collections class, but the definition of the sorting feature is obligatory:

List<Integer> list = new ArrayList(Arrays.asList(2, 4, 11, 8, 12, 3));
list.replaceAll(k -> k * k); // replace the numbers with their second powers
System.out.println(list);    // [4, 16, 121, 64, 144, 9]
list.sort(Integer::compare);
System.out.println(list);    // [4, 9, 16, 64, 121, 144]
list.sort((i1, i2) -> Integer.compare(i2, i1));
System.out.println(list);    // [144, 121, 64, 16, 9, 4]

The most significant changes have affected the Map interface. The added methods listed in the table:

The following example demonstrates the use of some of these methods:

package ua.in.iwanoff.java.sixth;

import java.util.HashMap;
import java.util.Map;

public class MapDemo {

    static void print(Integer i, String s) {
        System.out.printf("%3d %10s %n", i, s);
    }

    public static void main(String[] args) {
        Map<Integer, String> map = new HashMap<>();
        map.put(1, "one");
        map.put(2, "two");
        map.put(7, "seven");
        map.forEach(MapDemo::print); // columnwise output
        System.out.println(map.putIfAbsent(7, "eight")); // seven
        System.out.println(map.putIfAbsent(8, "eight")); // null
        System.out.println(map.getOrDefault(2, "zero")); // two
        System.out.println(map.getOrDefault(3, "zero")); // zero
        map.replaceAll((i, s) -> i > 1 ? s.toUpperCase() : s);
        System.out.println(map); // {1=one, 2=TWO, 7=SEVEN, 8=EIGHT}
        map.compute(7, (i, s) -> s.toLowerCase());
        System.out.println(map); // {1=one, 2=TWO, 7=seven, 8=EIGHT}
        map.computeIfAbsent(2, (i) -> i + "");
        System.out.println(map); // nothing changed
        map.computeIfAbsent(4, (i) -> i + "");
        System.out.println(map); // {1=one, 2=TWO, 4=4, 7=seven, 8=EIGHT}
        map.computeIfPresent(5, (i, s) -> s.toLowerCase());
        System.out.println(map); // nothing changed
        map.computeIfPresent(2, (i, s) -> s.toLowerCase());
        System.out.println(map); // {1=one, 2=two, 4=4, 7=seven, 8=EIGHT}
        // Adding a new pair:
        map.merge(9, "nine", (value, newValue) -> value.concat(newValue));
        System.out.println(map.get(9));                  // nine
        // The text is concatenated with the previous one:
        map.merge(9, " as well", (value, newValue) -> value.concat(newValue));
        System.out.println(map.get(9));                  // nine as well
    }
}

2.3.3 Using the Java 8 Stream API

Streams for work with collections, or streams of elements, data streams (Stream API) are designed for high-level processing of data stored in containers. They should not be confused with input / output streams.

The Stream API is used to search, filter, transform, find the minimum and maximum values, as well as other data manipulation. An important advantage of the Stream API is the ability to work reliably and efficiently in a multithreading environment.

Streams should be understood not as a new kind of collections, but as a channel for transmission and processing of data. The stream of elements works with some data source, such as an array or collection. The stream does not store data directly, but performs transferring, filtering, sorting, etc. The actions performed by the stream do not change the source data. For example, sorting data in a stream does not change their order in the source, but creates a separate resulting collection.

You can create sequential and parallel streams of elements. Parallel streams are secure in terms of multithreading. From the available parallel stream you can get sequential one and vice versa.

To work with streams Java 8 java.util.stream package provides a set of interfaces and classes that provide operations on a stream of elements in the style of functional programming. The stream is represented by an object that implements the java.util.stream.Stream interface. In turn, this interface inherits the methods of the general interface java.util.stream.BaseStream.

Stream operations (methods) defined in the BaseStream, Stream, and other derived interfaces are divided into intermediate and terminal. Intermediate operations receive and generate data streams and serve to create so-called pipelines, in which a sequence of actions is performed over a sequence. Terminal operations give the final result and thus "consume" the output stream. This means that the output stream cannot be reused and, if necessary, must be re-created.

The most significant methods of the generic java.util.stream.BaseStream interface are given in the table (S - type of the stream, E - type of the element, R - container type):

The use of stream iterators will be discussed later.

The Stream interface extends the set of methods for working with streaming elements. It is also a generic interface and is suitable for working with any reference types. The following are the most commonly used Stream interface methods:

There are several ways to create a stream. You can use the factory methods added to the Collection interface (with default implementations), respectively stream() (for synchronous work) and parallelStream() (for asynchronous work):

List<Integer> intList = Arrays.asList(3, 4, 1, 2);
Stream<Integer> sequential = intList.stream();
Stream<Integer> parallel = intList.parallelStream();

You can create a stream from an array:

Integer[] a = { 1, 2, 3 };
Stream<Integer> fromArray = Arrays.stream(a);

You can create a data source with the specified items. To do this, use the "factory" method of():

Stream<Integer> newStream = Stream.of(4, 5, 6);

Streams of items can be created from input streams (BufferedReader.lines()), filled with random values (Random.ints()), and also obtained from archives, bit sets, etc.

Intermediate operations are characterized by so-called lazy behavior: they are performed not instantaneously, but as the need arises - when the final operation is working with a new data stream. Lazy behavior increases the efficiency of work with the stream of elements.

Most operations are implemented in such a way that actions on individual elements do not depend on other elements. Such operations are called stateless operations. Other operations that require working on all elements at once (for example, sorted()) are called stateful operations.

You can get an array from a stream using the toArray() method. The following example creates a stream and then outputs to the console by creating an array and obtaining a string representation using the static Arrays.toString() method:

s = Stream.of (1, -2, 3);
System.out.println(Arrays.toString(s.toArray())); // [1, -2, 3]

Streams provide iterators. The iterator() method of the Stream interface returns an object that implements the java.util.Iterator interface. The iterator can be used explicitly:

s = Stream.of(11, -2, 3);
Iterator<Integer> it = s.iterator();
while (it.hasNext()) {
    System.out.println(it.next());
}

There is also a special type of iterator, a split iterator (implemented by the Spliterator interface). It, in particular, allows you to bypass some part of the elements.

Streams provide iteration over data elements using the forEach() method. The function parameter is the standard Consumer functional interface, which defines a method with a single parameter and a void result type. For example:

Stream<Integer> s = Stream.of(4, 5, 6, 1, 2, 3);
s.forEach(System.out::println);

The simplest stream operation is filtering. The intermediate filter() operation returns a filtered stream, taking a parameter of Predicate type. The Predicate type is a functional interface that describes a method with a single parameter and boolean result type. For example, you can filter out only even numbers from the stream s:

s.filter(k -> k % 2 == 0).forEach(System.out::println);

The previous example illustrates the use of lambda expressions when working with streams, as well as a small conveyor that includes one intermediate operation.

The intermediate sorted() operation returns the sorted representation of the stream. Elements are ordered in the natural order (if it is defined). In other cases, the Comparator interface should be implemented, for example, using the lambda expression:

// Sort ascending:
Stream<Integer> s = Stream.of(4, 5, 6, 1, 2, 3);
s.sorted().forEach(System.out::println);
// Sort descending:
s = Stream.of(4, 5, 6, 1, 2, 3);
s.sorted((k1, k2) -> Integer.compare(k2, k1)).forEach(System.out::println);

The last example shows that after each call to the terminal operation, the stream should be recreated.

The intermediate operation map() receives a functional interface that defines a certain function for transforming and forming a new stream from the resulting transformed elements. For example, we calculate the squares of numbers:

s = Stream.of(1, 2, 3);
s.map(x -> x * x).forEach(System.out::println);

Using the distinct() method, you can get a stream containing only different elements of the collection. For example:

s = Stream.of(1, 1, -2, 3, 3);
System.out.println(Arrays.toString(s.distinct().toArray())); // [1, -2, 3]

The end operation count() with the resulting type long returns the number of elements in the stream:

s = Stream.of(4, 5, 6, 1, 2, 3);
System.out.println(s.count()); // 6

The terminal operations min() and max() return Optional objects with a minimum and maximum value, respectively. A Comparator type parameter is used for comparison. For example:

s = Stream.of(11, -2, 3);
System.out.println(s.min(Integer::compare).get()); // -2

Using a terminal reduce() operation, we can calculate a scalar value. The reduce() operation in its simplest form performs the specified action with two operands, the first of which is the result of performing the action on the previous elements, and the second is the current element. In the following example, we find the sum of the elements of the data stream:

s = Stream.of(1, 1, -2, 3, 3);
Optional<Integer> sum = s.reduce((s1, s2) -> s1 + s2);
sum.ifPresent(System.out::println); // 6

The min(), max(), and reduce() operations get a scalar value from the stream, so they are called reduction operations.

There are also streams for working with primitive types: IntStream, LongStream and DoubleStream.

3 Sample Programs

3.1 Working with the ConcurrentSkipListMap Container

In the following program, one thread fills ConcurrentSkipListMap with a number / list of simple factors pairs, and another thread finds prime numbers (numbers with the only simple factor) and lists them. Numbers are checked in a given range.

PrimeFactorization class (getting simple factors):

package ua.in.iwanoff.java.sixth;

import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;

public class PrimeFactorization implements Runnable {
    static volatile Map<Long, List<Long>> table = new ConcurrentSkipListMap<>();
    private long from, to;

    public long getFrom() {
        return from;
    }

    public long getTo() {
        return to;
    }

    public void setRange(long from, long to) {
        this.from = from;
        this.to = to;
    }

    // Getting a list of dividers of a given number:
    private List<Long> factors(long n) {
        String number = n + "\t";
        List<Long> result = new ArrayList<>();
        for (long k = 2; k <= n; k++) {
            while (n % k == 0) {
                result.add(k);
                n /= k;
            }
        }
        System.out.println(number + result);
        return result;
    }

    @Override
    public void run() {
        try {
            for (long n = from; n <= to; n++) {
                table.put(n, factors(n));
                Thread.sleep(1);
            }
        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String toString() {
        String result = table.entrySet().size() + " numbers\n";
        for (Map.Entry<?, ?> e : table.entrySet()) {
            result += e + "\n";
        }
        return result;
    }

}

Primes class (prime numbers):

package ua.in.iwanoff.java.sixth;

import java.util.Arrays;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;

public class Primes implements Runnable {
    PrimeFactorization pf;
    Set<Long> result;

    public Primes(PrimeFactorization pf) {
        this.pf = pf;
    }

    @Override
    public void run() {
        result = new LinkedHashSet<>();
        try {
            for (long last = pf.getFrom(); last <= pf.getTo(); last++) {
                List<Long> factors;
                do { // trying to get the next set of numbers:
                    factors = pf.table.get(last);
                    Thread.sleep(1);
                }
                while (factors == null);
                // found a prime number:
                if (factors.size() == 1) {
                    result.add(last);
                    System.out.println(this);
                }
            }
        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String toString() {
        return " Prime numbers: " + result;
    }
}

PrimesTest class:

package ua.in.iwanoff.java.sixth;

public class PrimesTest {
    public static void main(String[] args) throws InterruptedException {
        PrimeFactorization pf = new PrimeFactorization();
        Primes primes = new Primes(pf);
        pf.setRange(100000L, 100100L);
        new Thread(pf).start();
        Thread primesThread = new Thread(primes);
        primesThread.start();
        primesThread.join();
        System.out.println(pf);
        System.out.println(primes);
    }
}

3.2 Getting a Table of Prime Numbers using Data Streams

The following program allows getting a table of primes in a given range. For prime numbers, it is advisable to use IntStream:

package ua.in.iwanoff.java.sixth;

import java.util.stream.IntStream;

public class PrimeFinder {
    private static boolean isPrime(int n) {
        return n > 1 && IntStream.range(2, n - 1).noneMatch(k -> n % k == 0);
    }

    public static void printAllPrimes(int from, int to) {
        IntStream primes = IntStream.range(from, to + 1).filter(PrimeFinder::isPrime);
        primes.forEach(System.out::println);
    }

    public static void main(String[] args) {
        printAllPrimes(6, 199);
    }
}

The isPrime() method checks whether n is a prime number. For numbers greater than 1, a sequence of consecutive integers is formed, for each of which is checked, whether n is divided by this number. In the printAllPrimes() method, we generate a stream of prime numbers using a filter and output the numbers using the forEach() method.

3.3 Creating a GUI Application for Calculating and Displaying Prime Numbers

Suppose it is necessary to develop a program for obtaining primes in the range from 1 to a certain value, which can be quite large. To find primes we will use the simplest algorithm for sequential checking of all numbers from 2 to the square root of the verifiable number. Such a check may take a long time. To create a user-friendly interface, it's advisable to use a separate execution thread for checking numbers. This will allow user to pause and resume searches, perform standard window manipulations, including closing window before the search will be finished.

The root pane of our application is BorderPane. The source code of the JavaFX application will be as follows:

package ua.in.iwanoff.java.sixth;

import javafx.application.Application;
import javafx.fxml.FXMLLoader;
import javafx.scene.Scene;
import javafx.scene.layout.BorderPane;
import javafx.stage.Stage;
import java.io.IOException;

public class PrimeApp extends Application {

    public static void main(String[] args) {
        launch(args);
    }

    @Override
    public void start(Stage primaryStage) {
        FXMLLoader loader = new FXMLLoader();
        try {
            BorderPane root = loader.load(getClass().getResource("PrimeAppWindow.fxml"));
            Scene scene = new Scene(root, 700, 500);
            primaryStage.setScene(scene);
            primaryStage.setTitle("Prime Numbers");
            primaryStage.show();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

The user interface will include the top panel (HBox), on which will be successively allocated a label with the text "To:", a textFieldTo input field, and four buttons: buttonStart, buttonSuspend, buttonResume and buttonStop with the text "Start", "Suspend", "Resume" and "Stop". After starting a program, only the buttonStart button is available. The middle part will be occupied by the textAreaResults input area, for which editing is disabled. The lower part will be occupied by the progressBar component, in which the percentage of the search process for primes will be displayed. The FXML document file will look like this:

<?xml version="1.0" encoding="UTF-8"?>

<?import javafx.geometry.Insets?>
<?import javafx.scene.control.Button?>
<?import javafx.scene.control.Label?>
<?import javafx.scene.control.TextArea?>
<?import javafx.scene.control.TextField?>
<?import javafx.scene.layout.BorderPane?>
<?import javafx.scene.layout.HBox?>

<?import javafx.scene.control.ProgressBar?>
<BorderPane prefHeight="400.0" prefWidth="600.0" xmlns="http://javafx.com/javafx/8.0.121"
            xmlns:fx="http://javafx.com/fxml/1" fx:controller="ua.in.iwanoff.java.sixth.PrimeAppController">
    <top>
        <HBox prefHeight="3.0" prefWidth="600.0" spacing="10.0" BorderPane.alignment="CENTER">
            <padding>
                <Insets bottom="10.0" left="10.0" right="10.0" top="10.0"/>
            </padding>
            <Label text="To:"/>
            <TextField fx:id="textFieldTo" text="1000"/>
            <Button fx:id="buttonStart" minWidth="80.0" text="Start" onAction="#startClick"/>
            <Button fx:id="buttonSuspend" minWidth="80.0" disable="true" text="Suspend" onAction="#suspendClick"/>
            <Button fx:id="buttonResume" minWidth="80.0" disable="true" text="Resume" onAction="#resumeClick"/>
            <Button fx:id="buttonStop" minWidth="80.0" disable="true" text="Stop" onAction="#stopClick"/>
        </HBox>
    </top>
    <center>
        <TextArea fx:id="textAreaResults" editable="false" wrapText="true" BorderPane.alignment="CENTER"/>
    </center>
    <bottom>
        <ProgressBar fx:id="progressBar" maxWidth="Infinity" progress="0"/>
    </bottom>
</BorderPane>

We can generate an empty controller class, add fields associated with visual components and methods of handling events:

package ua.in.iwanoff.java.sixth;

import javafx.event.ActionEvent;
import javafx.fxml.FXML;
import javafx.scene.control.*;

public class PrimeAppController {
    @FXML private TextField textFieldTo;
    @FXML private Button buttonStart;
    @FXML private Button buttonSuspend;
    @FXML private Button buttonResume;
    @FXML private Button buttonStop;
    @FXML private ProgressBar progressBar;
    @FXML private TextArea textAreaResults;

    @FXML private void startClick(ActionEvent actionEvent) {
        
    }

    @FXML private void suspendClick(ActionEvent actionEvent) {
        
    }

    @FXML private void resumeClick(ActionEvent actionEvent) {
        
    }

    @FXML private void stopClick(ActionEvent actionEvent) {
        
    }

}

After starting the program an application window will be the following:

Now, separately from the GUIs, you can create a class that will be responsible for calculating primes in a separate thread. The PrimeNumbers class will implement the Runnable interface. The main functions (start(), suspend(), resume() and stop()), which will be called from the controller, change the state of the object (suspended and stopped fields). To provide flexibility, the program updates data about simple numbers and percentage of process execution through the callback mechanism. The fields of type Runnable store appropriate objects. To launch run() methods in another thread, these functions are called using Platform.runLater() method. The PrimeNumbers class cheat code will look like this:

package ua.in.iwanoff.java.sixth;

import javafx.application.Platform;

public class PrimeNumbers implements Runnable {
    private Thread primesThread;     // the thread of the calculation of primes
    private int to;                  // the end of the range of the calculation of primes
    private int lastFound;           // last found prime
    private Runnable displayFunc;    // the function that is called to output the found number
    private Runnable percentageFunc; // a function that updates the percentage of the completed process
    private Runnable finishFunc;     // a function that is called after the finishing
    private double percentage;
    private boolean suspended;
    private boolean stopped;

    public PrimeNumbers(Runnable addFunc, Runnable percentageFunc, Runnable finishFunc) {
        this.displayFunc = addFunc;
        this.percentageFunc = percentageFunc;
        this.finishFunc = finishFunc;
    }

    public int getTo() {
        return to;
    }

    public void setTo(int to) {
        this.to = to;
    }

    public synchronized int getLastFound() {
        return lastFound;
    }

    private synchronized void setLastFound(int lastFound) {
        this.lastFound = lastFound;
    }

    public synchronized double getPercentage() {
        return percentage;
    }

    private synchronized void setPercentage(double percentage) {
        this.percentage = percentage;
    }

    public synchronized boolean isSuspended() {
        return suspended;
    }

    private synchronized void setSuspended(boolean suspended) {
        this.suspended = suspended;
    }

    public synchronized boolean isStopped() {
        return stopped;
    }

    private synchronized void setStopped(boolean stopped) {
        this.stopped = stopped;
    }

    @Override
    public void run() {
        for (int n = 2; n <= to; n++)
        {
            try {
                setPercentage(n * 1.0 / to);
                // Update the percentage:
                if (percentageFunc != null) {
                    Platform.runLater(percentageFunc);
                }
                boolean prime = true;
                for (int i = 2; i * i <= n; i++) {
                    if (n % i == 0) {
                        prime = false;
                        break;
                    }
                }
                Thread.sleep(20);
                if (prime) {
                    setLastFound(n);
                    // Display prime number found:
                    if (displayFunc != null) {
                        displayFunc.run();
                    }
                }
            }
            catch (InterruptedException e) {
                // Depending on the state of the object, 
                // wait for the continuation or complete the search:
                while (isSuspended()) {
                    try {
                        Thread.sleep(100);
                    }
                    catch (InterruptedException e1) {
                        // Interrupted by waiting:
                        if (isStopped()) {
                            break;
                        }
                    }
                }
                if (isStopped()) {
                    break;
                }
            }
        }
        if (finishFunc != null) {
            Platform.runLater(finishFunc);
        }
    }

    public void start() {
        primesThread = new Thread(this);
        setSuspended(false);
        setStopped(false);
        primesThread.start();
    }

    public void suspend() {
        setSuspended(true);
        primesThread.interrupt();
    }

    public void resume() {
        setSuspended(false);
        primesThread.interrupt();
    }

    public void stop() {
        setStopped(true);
        primesThread.interrupt();
    }
}

As you can see from the code, data access methods are implemented with the synchronized modifier, which prevents simultaneous writing and reading of partially completed data from different threads. Therefore, all references to class fields from its methods are carried out only through the access functions.

Now we can implement the controller code:

package ua.in.iwanoff.java.sixth;

import javafx.event.ActionEvent;
import javafx.fxml.FXML;
import javafx.scene.control.*;

public class PrimeAppController {
    @FXML private TextField textFieldTo;
    @FXML private Button buttonStart;
    @FXML private Button buttonSuspend;
    @FXML private Button buttonResume;
    @FXML private Button buttonStop;
    @FXML private ProgressBar progressBar;
    @FXML private TextArea textAreaResults;

    private PrimeNumbers primeNumbers = new PrimeNumbers(this::addToTextArea, this::setProgress, this::finish);

    @FXML private void startClick(ActionEvent actionEvent) {
        try {
            primeNumbers.setTo(Integer.parseInt(textFieldTo.getText()));
            textAreaResults.setText("");
            progressBar.setProgress(0);
            buttonStart.setDisable(true);
            buttonSuspend.setDisable(false);
            buttonResume.setDisable(true);
            buttonStop.setDisable(false);
            primeNumbers.start();
        } 
        catch (NumberFormatException e) {
            Alert alert = new Alert(Alert.AlertType.ERROR);
            alert.setTitle("Error");
            alert.setHeaderText("Wrong range!");
            alert.showAndWait();
        }
    }

    @FXML private void suspendClick(ActionEvent actionEvent) {
        primeNumbers.suspend();
        buttonStart.setDisable(true);
        buttonSuspend.setDisable(true);
        buttonResume.setDisable(false);
        buttonStop.setDisable(false);
    }

    @FXML private void resumeClick(ActionEvent actionEvent) {
        primeNumbers.resume();
        buttonStart.setDisable(true);
        buttonSuspend.setDisable(false);
        buttonResume.setDisable(true);
        buttonStop.setDisable(false);
    }

    @FXML private void stopClick(ActionEvent actionEvent) {
        primeNumbers.stop();
    }

    private void addToTextArea() {
        textAreaResults.setText(textAreaResults.getText() + primeNumbers.getLastFound() + " ");
    }

    private void setProgress() {
        progressBar.setProgress(primeNumbers.getPercentage());
    }

    private void finish() {
        buttonStart.setDisable(false);
        buttonSuspend.setDisable(true);
        buttonResume.setDisable(true);
        buttonStop.setDisable(true);
    }

}

The addToTextArea(), setProgress() and finish() methods are not intended ffor direct call from the controller, but for a callback. References to these functions are passed to the PrimeNumbers constructor. Event handlers call up the appropriate PrimeNumbers class methods and change the accessibility of the buttons.

4 Exercises

1. Create a function that calculates the square root, if possible, and returns an OptionalDouble object.
2. Create a program in which all positive integers of the specified range are printed, if the sum of digits is equal to the specified value. Use data streams. Do not use explicit loops.

5 Quiz

1. Give a definition of the process and the thread.
2. What are the benefits of using threads?
3. What are the ways to create threads in Java?
4. What are the states of the thread object?
5. When threads are terminated or suspended?
6. What means are used for threads synchronization?
7. In which cases is the thread blocked?
8. What is the usage of the synchronized modifier?
9. Which container classes are thread-safe?
10. What is the usage of Optional class?
11. Is it possible to use the Optional class with primitive types?
12. What are the additional capabilities of working with standard containers provided in Java 8?
13. What are the benefits and features of the Stream API?
14. How to get a stream from the collection?
15. How to get a stream from an array?
16. What is the difference between intermediate and terminal operations?