Thread pool and executors

In order to prevent overhead of thread creation and reuse existed threads Java supports thread pools since version 1.5. The mechanism of thread pool is based on the equally called pattern. Roughly speaking, thread pool is a queue of initialized threads which makes its usage less expensive neither than using classic single Thread approach.
Java Virtual Machine provides fast access to the queue and every thread has its own instruction to be performed; instructions are customized with objects that implement Callable or Runnable.

Instead of creating threads directly with operations like Thread th = new Thread(); you may use instance of Executor interface.

Executor executor = new Executor() {
    public void execute(Runnable command) {
        command.run();
     }
};
executor.execute(someRunnableInstance);

Executor uses already existed thread and makes the thread to perform operations defined at someRunnableInstance.

ExecutorService (http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorService.html) extends Executor interface within new very useful methods. If you’d like to have custom class which implements ExecutorService you must override about 13 methods (jdk 1.8.0.31)

Let’s take a look on a few of them

•  Future<?> submit(Runnable task), <T> Future<T> submit(Callable<T> task) – submits task to be executed.
• void shutdown() – initiates the shutdown of the submitted task (no guarantee that the task will be stopped)

Deadlock

Chip and Dale must save Zipper and beat Fat Cat then but they cannot decide who is responsible for saving Zipper and who is responsible for beating their enemy.

public class Deadlock {
  
 public void init() {
  
  RescueRangers chip = new RescueRangers();
  RescueRangers dale = new RescueRangers();
  chip.setMate(dale);
  dale.setMate(chip);
  
  Thread t1 = new Thread(chip, "Chip");
  Thread t2 = new Thread(dale, "Dale");
  t1.start();  
  t2.start();
 }  
 
 class RescueRangers implements Runnable {
  
  private RescueRangers mate;
  
  public void setMate(RescueRangers mate) {
   this.mate = mate;
  }
  
  public synchronized void saveZipper() {
   System.out.println("Saving Zipper " + Thread.currentThread().getName());
   mate.attackFatCat();   
  }
  
  public synchronized void attackFatCat() {
   System.out.println("Attack Fat Cat " + Thread.currentThread().getName());   
  }
  
  public void run() {
   saveZipper();
  }
 } 
}

Once you launch this code you will see that Chip and Dale cannot attack Fat Cat because they are stuck saving Zipper – it is Fat Cat’s trap.

Output:

Saving Zipper Chip

Saving Zipper Dale

Chip and Dale wait for each other saving Zipper, neither can proceed attacking Fat Cat until the other does first, so both are stuck.

Synchronization block and atomic operations

When two or more threads modify some single object state we can get unexpected result. For example, we expect to get the value of object modified by Thread-A, but we are getting some other unexpected value because object was also modified by Thread-B.

class MyObject {
 private String val;

 public String getVal() {
  return val;
 }

 public void setVal(String val) {
  this.val = val;
  for (int iter = 0; iter < 1000; ++iter) {
   String str = "Useless Operation";
   str += iter;
  }
  System.out.println(this.val);  
 }
  
}

Thread-A changed the state of the object using the setVal, then Thread-B also changed the object state and Thread-A will output the value that has been set by Thread-B. That’s the code snippet of threads initialization

Thread thread_a = new Thread(new Runnable() {

 @Override
 public void run() {
  obj.setVal("1");

 }
});

Thread thread_b = new Thread(new Runnable() {

 @Override
 public void run() {
  obj.setVal("9");

 }
});
thread_a.start();
thread_b.start();

The output is going to be “9”, “9”. Surely, if you removed the useless cycle from setVal then you would get proper results because the value would be printed by Thread-A faster than the Thread-B changes it. Another option to prevent this behavior is using synchronized keyword.

What is Thread?

Thread in java is:

1) an instance of java.lang.Thread

2) a thread of execution – some kind of “lightweight” process which has its own call stack

To initialize the instance of Thread it’s enough to use the default Thread constructor

Thread t = new Thread();

This instruction is useless, because thread does not have any functionality. Another Thread’s constructor which takes an instance of java.lang.Runnable interface allows us to bring some functionality to our thread.

Thread t = new Thread(new Runnable() {

 @Override
 public void run() {
   System.out.println("Hey from the thread");
 }
});

Note: as long as any instance of java.lang.Thread implements java.lang.Runnable we can pass any thread instance as a constructor parameter of other thread. To run our thread we need to invoke start 

method:

t.start();

Then we will see ‘Hey from the thread’ console message. 
Don’t be fooled by Thread’s run() method – it doesn’t launch your thread instructions for asynchronous execution but it in the same thread where this method has been invoked.
There are other Threads frequently used methods such as: sleep [static], yield [static], join. In order to remember what do they all need for it’d be better to take a look on a figure with possible thread states: 

Apache Spark checkpoint issue on windows

"To keep track of the log statistics for all of time, state must be maintained between processing RDD's in a DStream.

To maintain state for key-pair values, the data may be too big to fit in memory on one machine - Spark Streaming can maintain the state for you. To do that, call the updateStateByKey function of the Spark Streaming library.

First, in order to use updateStateByKey, checkpointing must be enabled on the streaming context. To do that, just call checkpoint on the streaming context with a directory to write the checkpoint data." (from http://databricks.gitbooks.io/databricks-spark-reference-applications/content/logs_analyzer/chapter1/total.html)

When you enable checkpointing for your streaming context with ssc.checkpoint(<PATH_TO_DIRECTORY>); you may get the error messages

java.io.IOException: Could not locate executable null\bin\winutils.exe in the Hadoop binaries.
...
Exception in thread "pool-8-thread-1" java.lang.NullPointerException
at java.lang.ProcessBuilder.start(ProcessBuilder.java:1010)
at org.apache.hadoop.util.Shell.runCommand(Shell.java:404)


Also, looking inside checkpoint directory you can find some files created at the moment of the stream processing, however these files are empty. These files store state information (act as checkpoint) without this data we cannot use updateStateByKey properly.

To solve this issue you need:

Regular Expression for Timestamp value

Given string: 2012-02-10 00:00:00
Task: Check if it matches java.sql.Timestamp format.
Solution: use the regular expression
((1\d{3})|(20\d{2}))-((0[1-9])|(1[0-2]))-((0[1-9])|([1-2]\d)|(3[0-1])) (([0-1]\d)|(2[0-3])):([0-5]\d):([0-5]\d)|(^$)

Java way:
String pattern = "(((1\\d{3})|(20\\d{2}))-((0[1-9])|(1[0-2]))-((0[1-9])|([1-2]\\d)|(3[0-1])) (([0-1]\\d)|(2[0-3])):([0-5]\\d):([0-5]\\d))|(^$)";

NOTE: the regular expression above allows empty values as well, remove |(^$) from the end of the expression to get rid of it.

PS: if you'd like to know more about regular expressions and play with it go to regexpal.com

Make a List

Just to summarize all the ways of list creation in Java.
Taken from: http://stackoverflow.com/questions/858572/how-to-make-a-new-list-in-java


JDK

1. List listA = new ArrayList<string>();
2. List listB = Arrays.asList("A", "B", "C")

Guava

1. List names = Lists.newArrayList("Mike", "John", "Lesly");
2. List chars = Lists.asList("A","B", new String [] {"C", "D"});

Immutable List

1. Collections.unmodifiableList(new ArrayList<string>(Arrays.asList("A","B")));
2. ImmutableList.builder()                                      // Guava
            .add("A")
            .add("B").build();
3. ImmutableList.of("A", "B");                                  // Guava
4. ImmutableList.copyOf(Lists.newArrayList("A", "B", "C"));     // Guava

Empty immutable List

1. Collections.emptyList();
2. Collections.EMPTY_LIST;

List of Characters

1. Lists.charactersOf("String")                                 // Guava
2. Lists.newArrayList(Splitter.fixedLength(1).split("String"))  // Guava

List of Integers

Ints.asList(1,2,3);                                             // Guava

A little java collections memo: