#Java MultiThreading Part III
##Lesson 8 : Example on using (low level synchronization)
Processor.java
import java.util.LinkedList;
import java.util.Random;
public class Processor {
private LinkedList <Integer> list = new LinkedList<Integer>();
private final int LIMIT = 10;
private Object lock = new Object(); //LockOn
public void produce() throws InterruptedException{
int value = 0;
while(true){ //Infinite Loop
synchronized(lock){
while(list.size() == LIMIT){
//wait until the current thread invoke notify()
lock.wait(); //No1
}
list.add(value ++);
lock.notify(); // Will wake up the consume() @ No2
}
}
}//Endof Produce()
public void consume() throws InterruptedException{
Random random = new Random();
while(true){
synchronized(lock){
while(list.size() == 0){
lock.wait(); //No2
}
System.out.print("List size: " + list.size());
int value = list.removeFirst();
System.out.println("; value: " + value);
lock.notify(); // Will wake up thread @ No1
}//EndOf Synchronized()
Thread.sleep(random.nextInt(1000)); //0-999
}//Endof While
}//Endof Consume()
}App.java
/*
* Low Level Synchronization (Example)
* Run Concurently(Simultaneously)
*/
public class App {
public static void main(String[] args) throws InterruptedException {
final Processor processor = new Processor();
Thread t1 = new Thread(new Runnable(){
@Override
public void run() {
try {
processor.produce();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
Thread t2 = new Thread(new Runnable(){
@Override
public void run() {
try {
processor.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
//Start
t1.start();
t2.start();
t1.join();
t2.join();
}//ENDof Main
}OUTPUT
List size: 10; value: 0
List size: 10; value: 1
List size: 10; value: 2
List size: 10; value: 3
List size: 10; value: 4
List size: 10; value: 5
List size: 10; value: 6
List size: 10; value: 7
List size: 10; value: 8
List size: 10; value: 9
List size: 10; value: 10
List size: 10; value: 11
List size: 10; value: 12
List size: 10; value: 13
List size: 10; value: 14
List size: 10; value: 15
List size: 10; value: 16
List size: 10; value: 17
##Lesson 9 : Re-entrants Locks Class (Al-ternative using Synchronize Keywords)
Runner.java
import java.util.Scanner;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class Runner {
private int count = 0;
private Lock lock = new ReentrantLock(); //Alternative to Synchronized
private Condition condition = lock.newCondition();
private void increment(){
for (int i = 0; i < 1000; i++){
count++;
}
}
public void firstThread() throws InterruptedException{
lock.lock();
System.out.println("Waiting...");
condition.await(); //Wait until signal() is invoked (No1)
System.out.println("Woken Up!");
try {
increment(); //Use try-catch => prevent Exception
} catch (Exception e) {
e.printStackTrace();
}finally{
lock.unlock();
}
}
public void secondThread() throws InterruptedException{
Thread.sleep(1000);
lock.lock();
System.out.println("Press The return key!");
new Scanner(System.in).nextLine();
System.out.println("Got return key!");
condition.signal(); //No1
try {
increment(); //Use try-catch => prevent Exception
} catch (Exception e) {
e.printStackTrace();
}finally{
lock.unlock();
}
}
public void finished(){
System.out.println("Count: " + count);
}
}App.java
/*
* Re-entrants Locks
* Run Concurently(Simultaneously)
*/
public class App {
public static void main(String[] args) throws InterruptedException {
final Runner runner = new Runner();
Thread t1 = new Thread(new Runnable(){
@Override
public void run() {
try {
runner.firstThread();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
Thread t2 = new Thread(new Runnable(){
@Override
public void run() {
try {
runner.secondThread();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
//Start
t1.start();
t2.start();
t1.join();
t2.join();
runner.finished();
}//ENDof Main
}OUTPUT
Waiting...
Press The return key!
Got return key!
Woken Up!
Count: 2000
##Lesson 10 : DeadLock Occured .... How to Solved it ?
Runner.java
import java.util.Random;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class Runner {
private Account acc1 = new Account();
private Account acc2 = new Account();
private Lock lock1 = new ReentrantLock();
private Lock lock2 = new ReentrantLock();
//Solution --> To DeadLock
private void acquireLocks(Lock lock1, Lock lock2) throws InterruptedException{
while(true){
//Acquire Lock
Boolean gotFirstLock = false;
Boolean gotSecondLock = false;
try{
gotFirstLock = lock1.tryLock(); //Return true if have
gotSecondLock = lock2.tryLock();
}finally{
if (gotFirstLock && gotSecondLock){
return;
}
if (gotFirstLock){
lock1.unlock();
}
if (gotSecondLock){
lock2.unlock();
}
}
//Lock not acquired
Thread.sleep(1);
}
}
public void firstThread() throws InterruptedException{
Random random = new Random();
for (int i = 0; i < 1000; i++){
acquireLocks(lock1,lock2);
try{
Account.transfer(acc1, acc2, random.nextInt(100)); //Transfer range(0-99)
}finally{
lock1.unlock();
lock2.unlock();
}
}
}//EndOf firstThread()
public void secondThread() throws InterruptedException{
Random random = new Random();
for (int i = 0; i < 1000; i++){
/*
* Dead Lock (Did not lock in the same Order)
*
* lock2.lock();
* lock1.lock();
*/
acquireLocks(lock2,lock1);
try{
Account.transfer(acc2, acc1, random.nextInt(100)); //Transfer range(0-99)
}finally{
lock1.unlock();
lock2.unlock();
}
}
}//EndOf secondThread()
public void finished(){
System.out.println("Account 1 Balance: " + acc1.getBalance());
System.out.println("Account 2 Balance: " + acc2.getBalance());
System.out.println("Total Account Balance: " + (acc1.getBalance() + acc2.getBalance()) );
}//EndOf Finished()
}Account.java
public class Account {
private int balance = 1000;
public void deposit(int amount){
balance += amount;
}
public void withdraw(int amount){
balance -= amount;
}
public int getBalance(){
return balance;
}
public static void transfer(Account acc1, Account acc2, int amount){
acc1.withdraw(amount);
acc2.deposit(amount);
}
}App.java
/*
* DeadLock
* Run Concurently(Simultaneously)
*/
public class App {
public static void main(String[] args) throws InterruptedException {
final Runner runner = new Runner();
Thread t1 = new Thread(new Runnable(){
@Override
public void run() {
try {
runner.firstThread();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
Thread t2 = new Thread(new Runnable(){
@Override
public void run() {
try {
runner.secondThread();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
//Start
t1.start();
t2.start();
t1.join();
t2.join();
runner.finished();
}//ENDof Main
}OUTPUT
Account 1 Balance: 2266
Account 2 Balance: -266
Total Account Balance: 2000
##Lesson 11 : Semaphores
###1. Introduce Semaphores Class(object)
Semaphore :
Mainly used to limit the number of simultaneous threads that can access a resources, but you can also use them to implement deadlock recovery systems since a semaphore with one permit is basically a lock that you can unlock from other threads.
App.java
import java.util.concurrent.Semaphore;
/*
* Semaphores
* Run Concurently(Simultaneously)
*/
public class App {
public static void main(String[] args) throws InterruptedException {
Semaphore semaphore = new Semaphore(1);
semaphore.release(); // +1 permits
semaphore.acquire(); // -1 permits
System.out.println(semaphore.availablePermits()); //1
}//ENDof Main
}OUTPUT
1
###2. How to use Semaphores Class(object)
Connection.java
import java.util.concurrent.Semaphore;
public class Connection {
private static Connection instance = new Connection();
private int connections = 0;
//Semaphore
private Semaphore sempahore = new Semaphore(10);
private Connection(){
}
public static Connection getConnection(){
return instance;
}
public void connect(){
try {
sempahore.acquire(); // -1
} catch (InterruptedException e1) {
e1.printStackTrace();
}
try{
doConnect();
}finally{
sempahore.release(); // +1
}
}
public void doConnect(){
synchronized(this){
connections++;
System.out.println("Current Connections: " + connections);
}
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
synchronized(this){
connections--;
}
}//ENDof Connect()
}App.java
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/*
* Semaphores
* Run Concurently(Simultaneously)
*/
public class App {
public static void main(String[] args) throws InterruptedException {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 200; i++) {
// submit => Create a new thread auto
executor.submit(new Runnable() {
@Override
public void run() {
Connection.getConnection().connect(); //Invoke
}
});
}//ENDof For loop
executor.shutdown();
executor.awaitTermination(1, TimeUnit.DAYS); //Wait for 1 day then ONLY terminate
}// ENDof Main
}OUTPUT
Current Connections: 1
Current Connections: 2
Current Connections: 3
Current Connections: 4
Current Connections: 5
Current Connections: 6
Current Connections: 7
Current Connections: 8
Current Connections: 9
Current Connections: 10
Current Connections: 1
Current Connections: 2
Current Connections: 3
Current Connections: 4
Current Connections: 5
Current Connections: 6
Current Connections: 7
Current Connections: 8
Current Connections: 9
Current Connections: 10
Current Connections: 1
Current Connections: 2
Current Connections: 3
Current Connections: 4
Current Connections: 5
Current Connections: 6
Current Connections: 7
Current Connections: 8
Current Connections: 9
Current Connections: 10
Current Connections: 1
Current Connections: 2
Current Connections: 3
Current Connections: 4
Current Connections: 5
Current Connections: 6
Current Connections: 7
Current Connections: 8
Current Connections: 9
Current Connections: 10
##Resources