<repo_name>RyanZhu1024/java_concurrency<file_sep>/src/main/java/producerconsumers/Main.java
package producerconsumers;

/**
 * Created by shuxuan on 1/22/15.
 */
public class Main {
    public static void main(String[] args) {
        FileMock fileMock=new FileMock(100,10);
        Buffer buffer=new Buffer(20);
        Producer producer=new Producer(fileMock,buffer);
        Consumer[] consumers=new Consumer[3];
        Thread threadProducer=new Thread(producer, "producer");
        Thread[] threadConsumers=new Thread[3];

        for(int i=0;i<3;i++){
            consumers[i]=new Consumer(buffer);
            threadConsumers[i]=new Thread(consumers[i],"consumer"+i);
        }

        threadProducer.start();
        for(int i=0;i<3;i++){
            threadConsumers[i].start();
        }
    }
}
<file_sep>/src/main/java/forkjoin/DocumentTask.java
package forkjoin;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

/**
 * Created by shuxuan on 1/30/15.
 */
public class DocumentTask extends RecursiveTask<Integer> {
    private String[][] document;
    private int start;
    private int end;
    private String word;

    public DocumentTask(String[][] document, int start, int end, String word) {
        this.document = document;
        this.start = start;
        this.end = end;
        this.word = word;
    }

    @Override
    protected Integer compute() {
        int result = 0;
        if (end - start < 10) {
            result = processLine(document, start, end, word);
        } else {
            int mid = (end + start) / 2;
            DocumentTask task1 = new DocumentTask(document, start, mid, word);
            DocumentTask task2 = new DocumentTask(document, mid, end, word);
            invokeAll(task1, task2);
            try {
                result = groupResults(task1.get(), task2.get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        return result;
    }

    private int groupResults(Integer number1, Integer number2) {
        return number1 + number2;
    }

    private int processLine(String[][] document, int start, int end, String word) {
        List<LineTask> tasks = new ArrayList<LineTask>();
        for (int i = start; i < end; i++) {
            String[] line = document[i];
            LineTask lineTask = new LineTask(0, line.length, word, line);
            tasks.add(lineTask);
        }
        invokeAll(tasks);// synchronous call
        int result = 0;
        for (int i = 0; i < tasks.size(); i++) {
            try {
                result += tasks.get(i).get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        return result;
    }
}
<file_sep>/src/main/java/executor/Task.java
package executor;

import java.util.concurrent.TimeUnit;

/**
 * Created by shuxuan on 1/27/15.
 */
public class Task implements Runnable {
    private String name;

    public Task(String name) {
        this.name = name;
    }

    @Override
    public void run() {
        System.out.println(this.name+" is running");
        try {
            TimeUnit.SECONDS.sleep(5L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        throw new RuntimeException();
    }
}
<file_sep>/README.md
# java_concurrency
a java concurrency repo
<file_sep>/src/main/java/phaser/StudentMain.java
package phaser;

/**
 * Created by shuxuan on 1/26/15.
 */
public class StudentMain {
    public static void main(String[] args) throws InterruptedException {
        MyPhaser myPhaser=new MyPhaser();
        Student[] students=new Student[5];
        for(int i=0;i<5;i++){
            students[i]=new Student(myPhaser);
            myPhaser.register();
        }
        Thread[] threads=new Thread[5];
        for(int i=0;i<5;i++){
            threads[i]=new Thread(students[i]);
            threads[i].start();
        }
        for(int i=0;i<5;i++){
            threads[i].join();
        }

        System.out.println("All threads have been terminated");
    }
}
<file_sep>/src/main/java/exchanger/Consumer.java
package exchanger;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Exchanger;

/**
 * Created by shuxuan on 1/27/15.
 */
public class Consumer implements Runnable {

    private List<String> buffer;
    private final Exchanger<List<String>> exchanger;

    public Consumer(List<String> buffer, Exchanger<List<String>> exchanger) {
        this.exchanger = exchanger;
        this.buffer = buffer;
    }

    @Override
    public void run() {
        int cycle = 1;
        for (int i = 0; i < 10; i++) {
            System.out.println("Consumer: " + cycle);
            try {
                this.buffer = this.exchanger.exchange(buffer);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            Iterator<String> iterator = this.buffer.iterator();
            while (iterator.hasNext()) {
                System.out.println("Consumer message: " + iterator.next());
                iterator.remove();
            }
            cycle++;
        }
    }
}
<file_sep>/src/main/java/exchanger/Main.java
package exchanger;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Exchanger;

/**
 * Created by shuxuan on 1/27/15.
 */
public class Main {
    public static void main(String[] args) {
        List<String> consumerList=new ArrayList<String>();
        List<String> producerList=new ArrayList<String>();
        Exchanger<List<String>> exchanger=new Exchanger<List<String>>();

        Producer producer=new Producer(producerList,exchanger);
        Consumer consumer=new Consumer(consumerList,exchanger);

        Thread proTh=new Thread(producer);
        Thread conTh=new Thread(consumer);

        proTh.start();
        conTh.start();
    }
}
<file_sep>/src/main/java/executor/FactorialMain.java
package executor;

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

/**
 * Created by shuxuan on 1/27/15.
 */
public class FactorialMain {
    public static void main(String[] args) {
        ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);
        List<Future<Integer>> resultList = new ArrayList();
        Random random = new Random();
        for (int i = 0; i < 10; i++) {
            Integer num = random.nextInt(10);
            FactorialCalculator factorialCalculator = new FactorialCalculator(num);
            Future<Integer> result = threadPoolExecutor.submit(factorialCalculator);
            resultList.add(result);
        }
        do {
            System.out.println("Main: currently done tasks: " + threadPoolExecutor.getCompletedTaskCount());
            for (int i = 0; i < resultList.size(); i++) {
                Future<Integer> future = resultList.get(i);
                System.out.println("Task: " + i + " is done? " + future.isDone());
            }
//            try {
//                TimeUnit.MILLISECONDS.sleep(50);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
        } while (threadPoolExecutor.getCompletedTaskCount() < resultList.size());// similar to the cyclicbarrier
        for (int i = 0; i < resultList.size(); i++) {
            Future<Integer> future = resultList.get(i);
            try {
                System.out.println("Task: " + i + " has value: " + future.get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }

            threadPoolExecutor.shutdown();
        }
    }
}
<file_sep>/src/main/java/executor/InvokeMain.java
package executor;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * Created by shuxuan on 1/28/15.
 */
public class InvokeMain {
    public static void main(String[] args) {
//        ScheduledThreadPoolExecutor executorService = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(1000);
        ScheduledThreadPoolExecutor executorService = new MyExecutor(1);
        List<InvokeTask> taskList = new ArrayList<InvokeTask>();
        for (int i = 0; i < 3; i++) {
            InvokeTask task = new InvokeTask(String.valueOf(i));
            taskList.add(task);
        }
        List<Future<Result>> futureList = new ArrayList();
        Task task = new Task("asd");
//        executorService.submit(new FutureTask(task,String.class));
        executorService.scheduleAtFixedRate(task, 1, 1, TimeUnit.MILLISECONDS);
//        executorService.scheduleWithFixedDelay(task,1,100,TimeUnit.MILLISECONDS);
//            futureList = executorService.scheduleAtFixedRate(task,1,1,TimeUnit.SECONDS);
        executorService.setContinueExistingPeriodicTasksAfterShutdownPolicy(true);
        executorService.shutdown();
        System.out.println("Main: printing all results");
        for (Future<Result> resultFuture : futureList) {
            try {
                Result result = resultFuture.get();
                System.out.println("Name: " + result.getName() + " and value: " + result.getValue());
//                executorService.awaitTermination(1, TimeUnit.DAYS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }

        }
    }

    public void test() {
        ScheduledThreadPoolExecutor executorService = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(1);
        executorService.setExecuteExistingDelayedTasksAfterShutdownPolicy(true);
        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);

    }
}
<file_sep>/src/main/java/completionService/ReportProcessor.java
package completionService;

import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * Created by shuxuan on 1/29/15.
 */
public class ReportProcessor implements Runnable {
    private CompletionService<String> service;

    public void setEnd(boolean end) {
        this.end = end;
    }

    private boolean end;

    public ReportProcessor(CompletionService<String> service) {
        this.service = service;
        this.end = false;
    }

    @Override
    public void run() {
        while(!end){
            Future<String> future=service.poll();
            if(future!=null){
                try {
                    String result=future.get();
                    System.out.println("Report Received: "+result);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }

            }
        }
    }
}
<file_sep>/src/main/java/countDownlatch/Main.java
package countDownlatch;

/**
 * Created by shuxuan on 1/23/15.
 */
public class Main {
    public static void main(String[] args) {
        VideoConference conference=new VideoConference(10);
        Thread conferenceThread=new Thread(conference);
        conferenceThread.start();
        for(int i=0;i<10;i++){
            Thread participantTh=new Thread(new Participant(conference,"Participant"+i));
            participantTh.start();
        }
    }
}
<file_sep>/src/main/java/executor/Server.java
package executor;

import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * Created by shuxuan on 1/27/15.
 */
public class Server {
    private ThreadPoolExecutor threadPoolExecutor;

    public Server() {
//        this.threadPoolExecutor= (ThreadPoolExecutor) Executors.newCachedThreadPool();
        this.threadPoolExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(5, Executors.defaultThreadFactory());
    }

    public void executeTask(Task task) {
        System.out.println("Server: a new Task arrived");
        this.threadPoolExecutor.execute(task);
//        System.out.println("Pool size: "+this.threadPoolExecutor.getPoolSize());
//        System.out.println("Active count: "+this.threadPoolExecutor.getActiveCount());
//        System.out.println("Server: completed tasks: "+this.threadPoolExecutor.getCompletedTaskCount());
    }

    public void endServer() {
        this.threadPoolExecutor.shutdown();
    }
}
<file_sep>/src/main/java/semaphoreControlling/Job.java
package semaphoreControlling;

/**
 * Created by shuxuan on 1/23/15.
 */
public class Job implements Runnable {

    private PrintQueue printQueue;

    public Job(PrintQueue queue) {
        this.printQueue=queue;
    }

    @Override
    public void run() {
        System.out.println("Ready for printing job for thread: "+Thread.currentThread().getName());
        this.printQueue.printJob(new Object());
        System.out.println("Finish printing job for thread: "+Thread.currentThread().getName());
    }
}
<file_sep>/src/main/java/forkjoin/Main3.java
package forkjoin;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;

/**
 * Created by shuxuan on 1/30/15.
 */
public class Main3 {
    public static void main(String[] args) {
        ForkJoinPool pool = new ForkJoinPool();
        FolderProcessor processor1 = new FolderProcessor("log", "/var/log");
        FolderProcessor processor2 = new FolderProcessor("log", "/var/log");
        FolderProcessor processor3 = new FolderProcessor("log", "/var/log");
        pool.execute(processor2);
        pool.execute(processor3);
        pool.execute(processor1);
        do {
            System.out.printf("***************************************** *\n");
            System.out.printf("Main: Parallelism: %d\n", pool.
                    getParallelism());
            System.out.printf("Main: Active Threads: %d\n", pool.
                    getActiveThreadCount());
            System.out.printf("Main: Task Count: %d\n", pool.
                    getQueuedTaskCount());
            System.out.printf("Main: Steal Count: %d\n", pool.
                    getStealCount());
            System.out.printf("***************************************** *\n");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } while (!processor1.isDone() || !processor2.isDone() || !processor3.isDone());
        System.out.println("Processor1: "+processor1.join().size());
        System.out.println("Processor2: "+processor2.join().size());
        System.out.println("Processor3: "+processor3.join().size());
    }
}
<file_sep>/src/main/java/exchanger/Producer.java
package exchanger;

import java.util.List;
import java.util.concurrent.Exchanger;

/**
 * Created by shuxuan on 1/27/15.
 */
public class Producer implements Runnable {
    private List<String> buffer;
    private final Exchanger<List<String>> exchanger;

    public Producer(List<String> buffer, Exchanger<List<String>> exchanger) {
        this.buffer = buffer;
        this.exchanger = exchanger;
    }

    @Override
    public void run() {
        int cycle=1;
        for(int i=0;i<10;i++){
            System.out.println("Producer: Cycle "+cycle);
            for(int j=0;j<10;j++){
                String message="Event: "+((i*10)+j);
                System.out.println("Producer: message "+message);
                this.buffer.add(message);
            }
            try {
                /*
                transfer this buffer to another thread as the parameter
                receive another thread's buffer as a return
                 */
                this.buffer=this.exchanger.exchange(buffer);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            cycle++;
        }
    }
}
<file_sep>/src/main/java/cyclicBarrier/Grouper.java
package cyclicBarrier;

/**
 * Created by shuxuan on 1/23/15.
 */
public class Grouper implements Runnable{

    private Result result;

    public Grouper(Result result) {
        this.result = result;
    }

    @Override
    public void run() {
        int finalResult=0;
        System.out.println("processing results---------");
        int data[]=result.getData();
        for(int number:data){
            finalResult+=number;
        }
        System.out.println("printing result: "+finalResult);
    }
}
<file_sep>/src/main/java/forkjoin/ProductGenerator.java
package forkjoin;

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

/**
 * Created by shuxuan on 1/29/15.
 */
public class ProductGenerator {
    public List<Product> generate(int size){
        List<Product> products=new ArrayList<Product>(size);
        for(int i=0;i<size;i++){
            Product product=new Product();
            product.setName("Product"+i);
            product.setPrice(10);
            products.add(product);
        }
        return products;
    }
}
