ThreadingTest.java

gistfile1.txt

package eu.mmacphail.threads;

import org.junit.jupiter.api.Test;

import java.time.LocalDateTime;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadingTest {

    @Test
    public void currentThread() throws InterruptedException {
        System.out.println(LocalDateTime.now() + " - Before sleeping");
        Thread.sleep(2000);
        System.out.println(LocalDateTime.now() + " - After sleeping");
        Thread thread = Thread.currentThread();
        System.out.println("id " + thread.getId());
        System.out.println("name " + thread.getName());
        System.out.println("thread group " + thread.getThreadGroup().getName());
        System.out.println("thread group count " + thread.getThreadGroup().activeCount());
    }

    @Test
    public void threadPool() {
        ExecutorService threadPool = Executors.newFixedThreadPool(5, getThreadFactory());
        Runnable sayHello = () -> System.out.println("Hello from thread " + Thread.currentThread().getName() + "-" + Thread.currentThread().getId());
        for (int i = 0; i < 100; i++) {
            threadPool.submit(sayHello);
        }
    }

    private ThreadFactory getThreadFactory() {
        return r -> {
            Thread t = new Thread(r);
            t.setName("MyThread");
            t.setDaemon(false);
            return t;
        };
    }

    @Test
    public void threadPoolUnboundedQueue() throws InterruptedException {
        final int numberOfTasks = 100;
        CountDownLatch latch = new CountDownLatch(numberOfTasks);

        ExecutorService threadPool = Executors.newFixedThreadPool(5, getThreadFactory());
        Runnable doStuff = runAndSleep(latch, 5000);
        for (int i = 0; i < numberOfTasks; i++) {
            threadPool.submit(doStuff);
        }
        latch.await();
    }

    @Test
    public void threadPoolOverflowUnboundedQueue() throws InterruptedException {
        final int numberOfTasks = 100000000;
        CountDownLatch latch = new CountDownLatch(numberOfTasks);

        ExecutorService threadPool = Executors.newFixedThreadPool(5, getThreadFactory());
        Runnable doStuff = runAndSleep(latch, 5000);
        for (int i = 0; i < numberOfTasks; i++) {
            threadPool.submit(doStuff);
        }
        latch.await();
    }

    @Test
    public void threadPoolAbortPolicy() throws InterruptedException {
        ExecutorService threadPool = new ThreadPoolExecutor(10, 10, 30,
                TimeUnit.SECONDS, new LinkedBlockingDeque<>(5),
                new ThreadPoolExecutor.AbortPolicy());

        final int numberOfTasks = 100;
        final int taskDuration = 5000;
        CountDownLatch latch = new CountDownLatch(numberOfTasks);

        Runnable doStuff = runAndSleep(latch, taskDuration);
        for (int i = 0; i < numberOfTasks; i++) {
            threadPool.submit(doStuff);
        }
        latch.await();
    }

    private AtomicInteger count = new AtomicInteger(0);

    @Test
    public void threadPoolDiscardPolicy() throws InterruptedException {
        count.set(0);
        ExecutorService threadPool = new ThreadPoolExecutor(10, 10, 30,
                TimeUnit.SECONDS, new LinkedBlockingDeque<>(5),
                new ThreadPoolExecutor.DiscardPolicy());

        final int numberOfTasks = 100;
        final int taskDuration = 5000;
        CountDownLatch latch = new CountDownLatch(numberOfTasks);

        Runnable doStuff = runAndSleep(latch, taskDuration);
        for (int i = 0; i < numberOfTasks; i++) {
            threadPool.submit(doStuff);
        }
        boolean completed = latch.await(30L, TimeUnit.SECONDS);
        System.out.println("Latch completed: " + completed);
        System.out.println("count : " + count.get());
    }

    @Test
    public void threadPoolDiscardOldestPolicy() throws InterruptedException {
        ExecutorService threadPool = new ThreadPoolExecutor(10, 10, 30,
                TimeUnit.SECONDS, new LinkedBlockingDeque<>(5),
                new ThreadPoolExecutor.DiscardOldestPolicy());

        final int numberOfTasks = 100;
        final int taskDuration = 5000;
        CountDownLatch latch = new CountDownLatch(numberOfTasks);

        Runnable doStuff = runAndSleep(latch, taskDuration);
        for (int i = 0; i < numberOfTasks; i++) {
            threadPool.submit(doStuff);
        }
        boolean completed = latch.await(30L, TimeUnit.SECONDS);
        System.out.println("Latch completed: " + completed);
    }

    @Test
    public void threadPoolCallerRunsPolicy() throws InterruptedException {
        ExecutorService threadPool = new ThreadPoolExecutor(10, 10, 30,
                TimeUnit.SECONDS, new LinkedBlockingDeque<>(5),
                new ThreadPoolExecutor.CallerRunsPolicy());

        final int numberOfTasks = 100;
        final int taskDuration = 5000;
        CountDownLatch latch = new CountDownLatch(numberOfTasks);

        Runnable doStuff = runAndSleep(latch, taskDuration);
        for (int i = 0; i < numberOfTasks; i++) {
            System.out.println("Submitting task " + i);
            threadPool.submit(doStuff);
        }
        boolean completed = latch.await(30L, TimeUnit.SECONDS);
        System.out.println("Latch completed: " + completed);
    }

    // Atomics like AtomicInteger, AtomicBoolean

    private Runnable runAndSleep(CountDownLatch latch, int sleepDuration) {
        return () -> {
            try {
                System.out.println(LocalDateTime.now() + "- Thread " + Thread.currentThread().getId() + " is busy");
                Thread.sleep(sleepDuration);
                System.out.println(LocalDateTime.now() + "- Thread " + Thread.currentThread().getId() + " is released");
                count.incrementAndGet();
                latch.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
                latch.countDown();
            }
        };
    }
}