Java Multi-Thread

1. 2种实现方法： thread类和runnable接口

extends Thead

package demo1;

class Runner extends Thread {

    @Override
    public void run() {
        for (int i=0;i<10;i++){
            System.out.println("hello"+i);
            
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
    
}

public class App {

    public static void main(String[] args) {
        Runner r1 = new Runner();
        Runner r2 = new Runner();
        r1.start();
        r2.start();
        
    }

}

implement runnable

package demo2;


class Runner implements Runnable{

    @Override
    public void run() {
        for (int i=0;i<10;i++){
            System.out.println("hello "+i);
            
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        
    }
    
}

public class App2 {

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


    }

}

 2. volatile

A variable’s value will be modified by different thread. Guaranteed it will not be cached, and different thread will see the update value.

 http://stackoverflow.com/questions/17748078/simplest-and-understandable-example-of-volatile-keyword-in-java

Simplest and understandable example of volatile keyword in java

http://www.javamex.com/tutorials/synchronization_volatile.shtml

The volatile keyword in Java

 3. How to wait until a thread finish running

because start() method return immediately, so main() thread can execute before a thread finish.

package demo4;

public class App {

    private int count = 0;
    public static void main(String[] args) {
        App app = new App();
        app.doWork();
    }
    public void doWork(){
        Thread t1 = new Thread(new Runnable(){
            public void run(){
                for (int i=0;i<10000;i++){
                    count++;
                }
            }
        });
        Thread t2 = new Thread(new Runnable(){
            public void run(){
                for (int i=0;i<10000;i++){
                    count++;
                }
            }
        });
        t1.start();
        t2.start();
        
        System.out.println(count);// output before t1 & t2 finish
    }

}

use join(), join() will return until thread finish run().

　　　　　　　t1.start();
        t2.start();
        
        try {
            t1.join();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        try {
            t2.join();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
        System.out.println(count);// output before t1 & t2 finish

4. synchronized: prevent to access variable at the same time

synchronized method:

package demo4;

public class App {

    private int count = 0;
    public synchronized void increment(){
        this.count++;
    }
    public static void main(String[] args) {
        App app = new App();
        app.doWork();
    }
    public void doWork(){
        Thread t1 = new Thread(new Runnable(){
            public void run(){
                for (int i=0;i<10000;i++){
                    increment();
                }
            }
        });
        Thread t2 = new Thread(new Runnable(){
            public void run(){
                for (int i=0;i<10000;i++){
                    increment();
                }
            }
        });
        t1.start();
        t2.start();
        
        try {
            t1.join();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        try {
            t2.join();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
        System.out.println(count);// output before t1 & t2 finish
    }

}

 synchronized block:

package demo4_2;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

public class Worker {
    
    private Random ran = new Random();

    private List<Integer> list1 = new ArrayList<Integer>();
    private List<Integer> list2 = new ArrayList<Integer>();
    

    
    public synchronized void stageOne(){
        try {
            Thread.sleep(1);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        list1.add(ran.nextInt(100));
    }
    public synchronized void stageTwo(){
        try {
            Thread.sleep(1);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        list2.add(ran.nextInt(100));
    }
    public void process(){
        for (int i=0;i<1000;i++){
            this.stageOne();
            this.stageTwo();
        }
    }
    public  void main() {
        System.out.println("worker");
        long start = System.currentTimeMillis();
        Thread t1 = new Thread(new Runnable(){
            public void run(){
                process();
            }
        });
        Thread t2 = new Thread(new Runnable(){
            public void run(){
                process();
            }
        });
        t1.start();
        t2.start();
        try {
            t1.join();
        } catch (InterruptedException e1) {
            e1.printStackTrace();
        }
        try {
            t2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        long end = System.currentTimeMillis();
        System.out.println("time taken "+ (end-start));
        System.out.println("list1 size: "+list1.size());
        System.out.println("list2 size: "+list2.size());


        

    }

}

 5.  Thread pool: Executor Service

http://tutorials.jenkov.com/java-util-concurrent/executorservice.html

ExecutorService

package demo5;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

class Processor implements Runnable{
    private int id;
    public Processor (int id){
        this.id = id;
    }

    @Override
    public void run() {
        System.out.println("starting :" +id);
        
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.out.println("complete :" +id);
        
        
    }
    
}
public class App {

    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        ExecutorService es =  Executors.newFixedThreadPool(2);
        // create 2 thread in the thread pool
        for (int i=0;i<5;i++){
            es.submit(new Processor(i));
            // the submitted task will be picked up by any available thread
        }
        es.shutdown();// you cannot submit new task after shut down
        long end = System.currentTimeMillis();

        System.out.println("all task submitted :"+(end-start));
        try {
            es.awaitTermination(10, TimeUnit.SECONDS);
            // wait for a certain time, it will wait less or equal to this time, before doing thing after this line
            // but it will not terminate anything, it just wait for this time
            // if you do not shutdown, the wait time will equal to this time
            // this method is similar to join, but with time constrain
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        end = System.currentTimeMillis();


        System.out.println("all task complete :"+(end-start));

    }

}

 add Future:

future.get();  //returns null if the task has finished correctly.

package demo5;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

class Processor implements Runnable{
    private int id;
    public Processor (int id){
        this.id = id;
    }

    @Override
    public void run() {
        System.out.println("starting :" +id);
        
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.out.println("complete :" +id);
        
        
    }
    
}
public class App {

    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        ExecutorService es =  Executors.newFixedThreadPool(2);
        // create 2 thread in the thread pool
        Future future[] = new Future[5];
        for (int i=0;i<5;i++){
            future[i] = es.submit(new Processor(i));
            // the submitted task will be picked up by any available thread
        }
        es.shutdown();// you cannot submit new task after shut down
        long end = System.currentTimeMillis();

        System.out.println("all task submitted :"+(end-start));
        try {
            es.awaitTermination(10, TimeUnit.SECONDS);
            // wait for a certain time, it will wait less or equal to this time, before doing thing after this line
            // but it will not terminate anything, it just wait for this time
            // if you do not shutdown, the wait time will equal to this time
            // similar to join, but with time constrain
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        end = System.currentTimeMillis();

        for (int i=0;i<5;i++){
            try {
                System.out.println("future :" + future[i].get());
            } catch (InterruptedException | ExecutionException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

        System.out.println("all task complete :"+(end-start));

    }

}

 6. CountDownLatch

Use CountDownLatch when one thread like main thread, require to wait for one or more thread to complete, before it can start processing.

http://stackoverflow.com/questions/17827022/what-is-countdown-latch-in-java-multithreading

What is Countdown Latch in Java MultiThreading?

package demo6;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

class Processor implements Runnable {

    private CountDownLatch latch;
    private int id;
    
    public Processor(CountDownLatch la, int id){
        this.latch = la;
        this.id = id;
    }
    @Override
    public void run() {
        System.out.println("start:"+this.id);
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.out.println("end:"+this.id);
        this.latch.countDown();

        
    }
    
}
public class App {

    public static void main(String[] args) {

        CountDownLatch latch = new CountDownLatch(3);
        ExecutorService  es = Executors.newFixedThreadPool(3);
        long start = System.currentTimeMillis();

        for (int i=0;i<4;i++){
            es.submit(new Processor(latch,i));
        }
        try {
            latch.await();
            // wait until latch count down to 0
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        long end = System.currentTimeMillis();

        System.out.println("complete count down :"+(end-start));
        // we have 4 task to submit, and 3 threads in thread pool
        // it will print after first 3 task finish, the 4th task will be executed after print
        // if count down from 3, time is 3000; if count 4, time is 6000

    }

}

 7. BlockingQueue

 Items inserted in a particular order are retrieved in that same order — but with the added guarantee that any attempt to retrieve an item from an empty queue will block the calling thread until the item is ready to be retrieved. Likewise, any attempt to insert an item into a queue that is full will block the calling thread until space becomes available in the queue's storage.

http://www.ibm.com/developerworks/java/library/j-5things4/index.html?ca=drs-

5 things you didn't know about ... java.util.concurrent, Part 1

package demo7;

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

public class App {
    private static BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(10);

    public static void main(String[] args) throws InterruptedException {
        
        Thread t1 = new Thread(new Runnable(){
            public void run(){
                try {
                    producer();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                
            }
        });
        
        Thread t2 = new Thread(new Runnable(){
            public void run(){
                try {
                    consumer();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                
            }
        });
        
        t1.start();
        t2.start();
        
        t1.join();
        t2.join();
    }
    
    private static void producer() throws InterruptedException{
        Random ran = new Random();
        while(true){
            queue.put(ran.nextInt(100));
        }
    }
    private static void consumer() throws InterruptedException{
        Random ran = new Random();

        while(true){
            Thread.sleep(100);
            if (ran.nextInt(10)==0){
                Integer va = queue.take();
                System.out.println("take value: "+va+" queu size: "+queue.size());

            }
        }
    }

}