mindol1004/spring-boot-rest
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

java concurrency (#2502)

Browse Source 
* fix spring config

* fix spring config

* fix spring config

* minor fix

* fix spring-boot module

* fix pom

* upgrade jackson

* minor fix

* java concurrency

* cleanup
This commit is contained in:
Doha2012
2017-08-27 12:14:26 +02:00
committed by Grzegorz Piwowarek
parent 09d10ac85f
commit 25263f1d6f
93 changed files with 362 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
core-java-concurrency/.gitignore vendored Normal file
View File

@@ -0,0 +1,26 @@
*.class
0.*
#folders#
/target
/neoDb*
/data
/src/main/webapp/WEB-INF/classes
*/META-INF/*
.resourceCache
# Packaged files #
*.jar
*.war
*.ear
# Files generated by integration tests
*.txt
backup-pom.xml
/bin/
/temp
#IntelliJ specific
.idea/
*.iml

32
core-java-concurrency/README.md Normal file
View File

@@ -0,0 +1,32 @@
=========
## Core Java Concurrency Examples
### Relevant Articles:
- [Guide To CompletableFuture](http://www.baeldung.com/java-completablefuture)
- [A Guide to the Java ExecutorService](http://www.baeldung.com/java-executor-service-tutorial)
- [Introduction to Thread Pools in Java](http://www.baeldung.com/thread-pool-java-and-guava)
- [Guide to java.util.concurrent.Future](http://www.baeldung.com/java-future)
- [Guide to java.util.concurrent.BlockingQueue](http://www.baeldung.com/java-blocking-queue)
- [Guide to CountDownLatch in Java](http://www.baeldung.com/java-countdown-latch)
- [A Guide to ConcurrentMap](http://www.baeldung.com/java-concurrent-map)
- [Guide to PriorityBlockingQueue in Java](http://www.baeldung.com/java-priority-blocking-queue)
- [Avoiding the ConcurrentModificationException in Java](http://www.baeldung.com/java-concurrentmodificationexception)
- [Custom Thread Pools In Java 8 Parallel Streams](http://www.baeldung.com/java-8-parallel-streams-custom-threadpool)
- [Guide to java.util.concurrent.Locks](http://www.baeldung.com/java-concurrent-locks)
- [An Introduction to ThreadLocal in Java](http://www.baeldung.com/java-threadlocal)
- [Guide to DelayQueue](http://www.baeldung.com/java-delay-queue)
- [A Guide to Java SynchronousQueue](http://www.baeldung.com/java-synchronous-queue)
- [Guide to the Java TransferQueue](http://www.baeldung.com/java-transfer-queue)
- [Guide to the ConcurrentSkipListMap](http://www.baeldung.com/java-concurrent-skip-list-map)
- [Difference Between Wait and Sleep in Java](http://www.baeldung.com/java-wait-and-sleep)
- [LongAdder and LongAccumulator in Java](http://www.baeldung.com/java-longadder-and-longaccumulator)
- [The Dining Philosophers Problem in Java](http://www.baeldung.com/java-dining-philoshophers)
- [Guide to CopyOnWriteArrayList](http://www.baeldung.com/java-copy-on-write-arraylist)
- [Guide to the Java Phaser](http://www.baeldung.com/java-phaser)
- [Guide to Synchronized Keyword in Java](http://www.baeldung.com/java-synchronized)
- [An Introduction to Atomic Variables in Java](http://www.baeldung.com/java-atomic-variables)
- [CyclicBarrier in Java](http://www.baeldung.com/java-cyclic-barrier)
- [Guide to Volatile Keyword in Java](http://www.baeldung.com/java-volatile)
- [Overview of the java.util.concurrent](http://www.baeldung.com/java-util-concurrent)
- [Semaphores in Java](http://www.baeldung.com/java-semaphore)

237
core-java-concurrency/pom.xml Normal file
View File

@@ -0,0 +1,237 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.baeldung</groupId>
<artifactId>core-java-concurrency</artifactId>
<version>0.1.0-SNAPSHOT</version>
<packaging>jar</packaging>
<name>core-java-concurrency</name>
<parent>
<groupId>com.baeldung</groupId>
<artifactId>parent-modules</artifactId>
<version>1.0.0-SNAPSHOT</version>
</parent>
<dependencies>
<!-- utils -->
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>${guava.version}</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-collections4</artifactId>
<version>${commons-collections4.version}</version>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>${commons-io.version}</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>${commons-lang3.version}</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-math3</artifactId>
<version>${commons-math3.version}</version>
</dependency>
<!-- test scoped -->
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<version>${assertj.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.jayway.awaitility</groupId>
<artifactId>awaitility</artifactId>
<version>${avaitility.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<finalName>core-java-concurrency</finalName>
<resources>
<resource>
<directory>src/main/resources</directory>
<filtering>true</filtering>
</resource>
</resources>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
<executions>
<execution>
<id>copy-dependencies</id>
<phase>prepare-package</phase>
<goals>
<goal>copy-dependencies</goal>
</goals>
<configuration>
<outputDirectory>${project.build.directory}/libs</outputDirectory>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<configuration>
<archive>
<manifest>
<addClasspath>true</addClasspath>
<classpathPrefix>libs/</classpathPrefix>
<mainClass>org.baeldung.executable.ExecutableMavenJar</mainClass>
</manifest>
</archive>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
<configuration>
<archiveBaseDirectory>${project.basedir}</archiveBaseDirectory>
<archive>
<manifest>
<mainClass>org.baeldung.executable.ExecutableMavenJar</mainClass>
</manifest>
</archive>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<shadedArtifactAttached>true</shadedArtifactAttached>
<transformers>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>org.baeldung.executable.ExecutableMavenJar</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>com.jolira</groupId>
<artifactId>onejar-maven-plugin</artifactId>
<executions>
<execution>
<configuration>
<mainClass>org.baeldung.executable.ExecutableMavenJar</mainClass>
<attachToBuild>true</attachToBuild>
<filename>${project.build.finalName}-onejar.${project.packaging}</filename>
</configuration>
<goals>
<goal>one-jar</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
<configuration>
<classifier>spring-boot</classifier>
<mainClass>org.baeldung.executable.ExecutableMavenJar</mainClass>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<profiles>
<profile>
<id>integration</id>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<executions>
<execution>
<phase>integration-test</phase>
<goals>
<goal>test</goal>
</goals>
<configuration>
<excludes>
<exclude>**/*ManualTest.java</exclude>
</excludes>
<includes>
<include>**/*IntegrationTest.java</include>
</includes>
</configuration>
</execution>
</executions>
<configuration>
<systemPropertyVariables>
<test.mime>json</test.mime>
</systemPropertyVariables>
</configuration>
</plugin>
</plugins>
</build>
</profile>
</profiles>
<properties>
<!-- util -->
<guava.version>21.0</guava.version>
<commons-lang3.version>3.5</commons-lang3.version>
<commons-math3.version>3.6.1</commons-math3.version>
<commons-io.version>2.5</commons-io.version>
<commons-collections4.version>4.1</commons-collections4.version>
<collections-generic.version>4.01</collections-generic.version>
<!-- testing -->
<assertj.version>3.6.1</assertj.version>
<avaitility.version>1.7.0</avaitility.version>
</properties>
</project>

34
core-java-concurrency/src/main/java/com/baeldung/concurrent/Scheduledexecutorservice/ScheduledExecutorServiceDemo.java Normal file
View File

@@ -0,0 +1,34 @@
package com.baeldung.concurrent.Scheduledexecutorservice;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
public class ScheduledExecutorServiceDemo {
public void execute() {
ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
ScheduledFuture<?> scheduledFuture = executorService.schedule(() -> {
// Task
}, 1, TimeUnit.SECONDS);
executorService.scheduleAtFixedRate(() -> {
// Task
}, 1, 10, TimeUnit.SECONDS);
executorService.scheduleWithFixedDelay(() -> {
// Task
}, 1, 10, TimeUnit.SECONDS);
Future<String> future = executorService.schedule(() -> {
// Task
return "Hellow world";
}, 1, TimeUnit.SECONDS);
executorService.shutdown();
}
}

13
core-java-concurrency/src/main/java/com/baeldung/concurrent/atomic/SafeCounterWithLock.java Normal file
View File

@@ -0,0 +1,13 @@
package com.baeldung.concurrent.atomic;
public class SafeCounterWithLock {
private volatile int counter;
public int getValue() {
return counter;
}
public synchronized void increment() {
counter++;
}
}

21
core-java-concurrency/src/main/java/com/baeldung/concurrent/atomic/SafeCounterWithoutLock.java Normal file
View File

@@ -0,0 +1,21 @@
package com.baeldung.concurrent.atomic;
import java.util.concurrent.atomic.AtomicInteger;
public class SafeCounterWithoutLock {
private final AtomicInteger counter = new AtomicInteger(0);
public int getValue() {
return counter.get();
}
public void increment() {
while(true) {
int existingValue = getValue();
int newValue = existingValue + 1;
if(counter.compareAndSet(existingValue, newValue)) {
return;
}
}
}
}

13
core-java-concurrency/src/main/java/com/baeldung/concurrent/atomic/UnsafeCounter.java Normal file
View File

@@ -0,0 +1,13 @@
package com.baeldung.concurrent.atomic;
public class UnsafeCounter {
int counter;
public int getValue() {
return counter;
}
public void increment() {
counter++;
}
}

24
core-java-concurrency/src/main/java/com/baeldung/concurrent/blockingqueue/BlockingQueueUsage.java Normal file
View File

@@ -0,0 +1,24 @@
package com.baeldung.concurrent.blockingqueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class BlockingQueueUsage {
public static void main(String[] args) {
int BOUND = 10;
int N_PRODUCERS = 4;
int N_CONSUMERS = Runtime.getRuntime().availableProcessors();
int poisonPill = Integer.MAX_VALUE;
int poisonPillPerProducer = N_CONSUMERS / N_PRODUCERS;
BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(BOUND);
for (int i = 0; i < N_PRODUCERS; i++) {
new Thread(new NumbersProducer(queue, poisonPill, poisonPillPerProducer)).start();
}
for (int j = 0; j < N_CONSUMERS; j++) {
new Thread(new NumbersConsumer(queue, poisonPill)).start();
}
}
}

28
core-java-concurrency/src/main/java/com/baeldung/concurrent/blockingqueue/NumbersConsumer.java Normal file
View File

@@ -0,0 +1,28 @@
package com.baeldung.concurrent.blockingqueue;
import java.util.concurrent.BlockingQueue;
public class NumbersConsumer implements Runnable {
private final BlockingQueue<Integer> queue;
private final int poisonPill;
NumbersConsumer(BlockingQueue<Integer> queue, int poisonPill) {
this.queue = queue;
this.poisonPill = poisonPill;
}
public void run() {
try {
while (true) {
Integer number = queue.take();
if (number.equals(poisonPill)) {
return;
}
String result = number.toString();
System.out.println(Thread.currentThread().getName() + " result: " + result);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}

36
core-java-concurrency/src/main/java/com/baeldung/concurrent/blockingqueue/NumbersProducer.java Normal file
View File

@@ -0,0 +1,36 @@
package com.baeldung.concurrent.blockingqueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadLocalRandom;
public class NumbersProducer implements Runnable {
private final BlockingQueue<Integer> numbersQueue;
private final int poisonPill;
private final int poisonPillPerProducer;
NumbersProducer(BlockingQueue<Integer> numbersQueue, int poisonPill, int poisonPillPerProducer) {
this.numbersQueue = numbersQueue;
this.poisonPill = poisonPill;
this.poisonPillPerProducer = poisonPillPerProducer;
}
public void run() {
try {
generateNumbers();
} catch (InterruptedException e) {
Thread.currentThread()
.interrupt();
}
}
private void generateNumbers() throws InterruptedException {
for (int i = 0; i < 100; i++) {
numbersQueue.put(ThreadLocalRandom.current()
.nextInt(100));
}
for (int j = 0; j < poisonPillPerProducer; j++) {
numbersQueue.put(poisonPill);
}
}
}

23
core-java-concurrency/src/main/java/com/baeldung/concurrent/countdownlatch/BrokenWorker.java Normal file
View File

@@ -0,0 +1,23 @@
package com.baeldung.concurrent.countdownlatch;
import java.util.List;
import java.util.concurrent.CountDownLatch;
public class BrokenWorker implements Runnable {
private final List<String> outputScraper;
private final CountDownLatch countDownLatch;
BrokenWorker(final List<String> outputScraper, final CountDownLatch countDownLatch) {
this.outputScraper = outputScraper;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
if (true) {
throw new RuntimeException("Oh dear");
}
countDownLatch.countDown();
outputScraper.add("Counted down");
}
}

34
core-java-concurrency/src/main/java/com/baeldung/concurrent/countdownlatch/WaitingWorker.java Normal file
View File

@@ -0,0 +1,34 @@
package com.baeldung.concurrent.countdownlatch;
import java.util.List;
import java.util.concurrent.CountDownLatch;
public class WaitingWorker implements Runnable {
private final List<String> outputScraper;
private final CountDownLatch readyThreadCounter;
private final CountDownLatch callingThreadBlocker;
private final CountDownLatch completedThreadCounter;
WaitingWorker(final List<String> outputScraper, final CountDownLatch readyThreadCounter, final CountDownLatch callingThreadBlocker, CountDownLatch completedThreadCounter) {
this.outputScraper = outputScraper;
this.readyThreadCounter = readyThreadCounter;
this.callingThreadBlocker = callingThreadBlocker;
this.completedThreadCounter = completedThreadCounter;
}
@Override
public void run() {
// Mark this thread as read / started
readyThreadCounter.countDown();
try {
callingThreadBlocker.await();
outputScraper.add("Counted down");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
completedThreadCounter.countDown();
}
}
}

22
core-java-concurrency/src/main/java/com/baeldung/concurrent/countdownlatch/Worker.java Normal file
View File

@@ -0,0 +1,22 @@
package com.baeldung.concurrent.countdownlatch;
import java.util.List;
import java.util.concurrent.CountDownLatch;
public class Worker implements Runnable {
private final List<String> outputScraper;
private final CountDownLatch countDownLatch;
Worker(final List<String> outputScraper, final CountDownLatch countDownLatch) {
this.outputScraper = outputScraper;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
// Do some work
System.out.println("Doing some logic");
outputScraper.add("Counted down");
countDownLatch.countDown();
}
}

77
core-java-concurrency/src/main/java/com/baeldung/concurrent/cyclicbarrier/CyclicBarrierDemo.java Normal file
View File

@@ -0,0 +1,77 @@
package com.baeldung.concurrent.cyclicbarrier;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierDemo {
private CyclicBarrier cyclicBarrier;
private List<List<Integer>> partialResults = Collections.synchronizedList(new ArrayList<>());
private Random random = new Random();
private int NUM_PARTIAL_RESULTS;
private int NUM_WORKERS;
private void runSimulation(int numWorkers, int numberOfPartialResults) {
NUM_PARTIAL_RESULTS = numberOfPartialResults;
NUM_WORKERS = numWorkers;
cyclicBarrier = new CyclicBarrier(NUM_WORKERS, new AggregatorThread());
System.out.println("Spawning " + NUM_WORKERS + " worker threads to compute " + NUM_PARTIAL_RESULTS + " partial results each");
for (int i = 0; i < NUM_WORKERS; i++) {
Thread worker = new Thread(new NumberCruncherThread());
worker.setName("Thread " + i);
worker.start();
}
}
class NumberCruncherThread implements Runnable {
@Override
public void run() {
String thisThreadName = Thread.currentThread().getName();
List<Integer> partialResult = new ArrayList<>();
for (int i = 0; i < NUM_PARTIAL_RESULTS; i++) {
Integer num = random.nextInt(10);
System.out.println(thisThreadName + ": Crunching some numbers! Final result - " + num);
partialResult.add(num);
}
partialResults.add(partialResult);
try {
System.out.println(thisThreadName + " waiting for others to reach barrier.");
cyclicBarrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}
}
class AggregatorThread implements Runnable {
@Override
public void run() {
String thisThreadName = Thread.currentThread().getName();
System.out.println(thisThreadName + ": Computing final sum of " + NUM_WORKERS + " workers, having " + NUM_PARTIAL_RESULTS + " results each.");
int sum = 0;
for (List<Integer> threadResult : partialResults) {
System.out.print("Adding ");
for (Integer partialResult : threadResult) {
System.out.print(partialResult + " ");
sum += partialResult;
}
System.out.println();
}
System.out.println(Thread.currentThread().getName() + ": Final result = " + sum);
}
}
public static void main(String[] args) {
CyclicBarrierDemo play = new CyclicBarrierDemo();
play.runSimulation(5, 3);
}
}

24
core-java-concurrency/src/main/java/com/baeldung/concurrent/cyclicbarrier/CyclicBarrierExample.java Normal file
View File

@@ -0,0 +1,24 @@
package com.baeldung.concurrent.cyclicbarrier;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierExample {
public void start() {
CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {
// Task
System.out.println("All previous tasks are completed");
});
Thread t1 = new Thread(new Task(cyclicBarrier), "T1");
Thread t2 = new Thread(new Task(cyclicBarrier), "T2");
Thread t3 = new Thread(new Task(cyclicBarrier), "T3");
if (!cyclicBarrier.isBroken()) {
t1.start();
t2.start();
t3.start();
}
}
}

24
core-java-concurrency/src/main/java/com/baeldung/concurrent/cyclicbarrier/Task.java Normal file
View File

@@ -0,0 +1,24 @@
package com.baeldung.concurrent.cyclicbarrier;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class Task implements Runnable {
private CyclicBarrier barrier;
public Task(CyclicBarrier barrier) {
this.barrier = barrier;
}
@Override
public void run() {
try {
System.out.println("Thread : " + Thread.currentThread().getName() + " is waiting");
barrier.await();
System.out.println("Thread : " + Thread.currentThread().getName() + " is released");
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}
}

32
core-java-concurrency/src/main/java/com/baeldung/concurrent/delayqueue/DelayObject.java Normal file
View File

@@ -0,0 +1,32 @@
package com.baeldung.concurrent.delayqueue;
import com.google.common.primitives.Ints;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class DelayObject implements Delayed {
private String data;
private long startTime;
DelayObject(String data, long delayInMilliseconds) {
this.data = data;
this.startTime = System.currentTimeMillis() + delayInMilliseconds;
}
@Override
public long getDelay(TimeUnit unit) {
long diff = startTime - System.currentTimeMillis();
return unit.convert(diff, TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(Delayed o) {
return Ints.saturatedCast(this.startTime - ((DelayObject) o).startTime);
}
@Override
public String toString() {
return "{" + "data='" + data + '\'' + ", startTime=" + startTime + '}';
}
}

30
core-java-concurrency/src/main/java/com/baeldung/concurrent/delayqueue/DelayQueueConsumer.java Normal file
View File

@@ -0,0 +1,30 @@
package com.baeldung.concurrent.delayqueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
public class DelayQueueConsumer implements Runnable {
private BlockingQueue<DelayObject> queue;
private final Integer numberOfElementsToTake;
final AtomicInteger numberOfConsumedElements = new AtomicInteger();
DelayQueueConsumer(BlockingQueue<DelayObject> queue, Integer numberOfElementsToTake) {
this.queue = queue;
this.numberOfElementsToTake = numberOfElementsToTake;
}
@Override
public void run() {
for (int i = 0; i < numberOfElementsToTake; i++) {
try {
DelayObject object = queue.take();
numberOfConsumedElements.incrementAndGet();
System.out.println("Consumer take: " + object);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

34
core-java-concurrency/src/main/java/com/baeldung/concurrent/delayqueue/DelayQueueProducer.java Normal file
View File

@@ -0,0 +1,34 @@
package com.baeldung.concurrent.delayqueue;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
public class DelayQueueProducer implements Runnable {
private BlockingQueue<DelayObject> queue;
private final Integer numberOfElementsToProduce;
private final Integer delayOfEachProducedMessageMilliseconds;
DelayQueueProducer(BlockingQueue<DelayObject> queue,
Integer numberOfElementsToProduce,
Integer delayOfEachProducedMessageMilliseconds) {
this.queue = queue;
this.numberOfElementsToProduce = numberOfElementsToProduce;
this.delayOfEachProducedMessageMilliseconds = delayOfEachProducedMessageMilliseconds;
}
@Override
public void run() {
for (int i = 0; i < numberOfElementsToProduce; i++) {
DelayObject object
= new DelayObject(UUID.randomUUID().toString(), delayOfEachProducedMessageMilliseconds);
System.out.println("Put object = " + object);
try {
queue.put(object);
Thread.sleep(500);
} catch (InterruptedException ie) {
ie.printStackTrace();
}
}
}
}

29
core-java-concurrency/src/main/java/com/baeldung/concurrent/diningphilosophers/DiningPhilosophers.java Normal file
View File

@@ -0,0 +1,29 @@
package com.baeldung.concurrent.diningphilosophers;
public class DiningPhilosophers {
public static void main(String[] args) throws Exception {
Philosopher[] philosophers = new Philosopher[5];
Object[] forks = new Object[philosophers.length];
for (int i = 0; i < forks.length; i++) {
forks[i] = new Object();
}
for (int i = 0; i < philosophers.length; i++) {
Object leftFork = forks[i];
Object rightFork = forks[(i + 1) % forks.length];
if (i == philosophers.length - 1) {
philosophers[i] = new Philosopher(rightFork, leftFork); // The last philosopher picks up the right fork first
} else {
philosophers[i] = new Philosopher(leftFork, rightFork);
}
Thread t = new Thread(philosophers[i], "Philosopher " + (i + 1));
t.start();
}
}
}

36
core-java-concurrency/src/main/java/com/baeldung/concurrent/diningphilosophers/Philosopher.java Normal file
View File

@@ -0,0 +1,36 @@
package com.baeldung.concurrent.diningphilosophers;
public class Philosopher implements Runnable {
private final Object leftFork;
private final Object rightFork;
Philosopher(Object left, Object right) {
this.leftFork = left;
this.rightFork = right;
}
private void doAction(String action) throws InterruptedException {
System.out.println(Thread.currentThread().getName() + " " + action);
Thread.sleep(((int) (Math.random() * 100)));
}
@Override
public void run() {
try {
while (true) {
doAction(System.nanoTime() + ": Thinking"); // thinking
synchronized (leftFork) {
doAction(System.nanoTime() + ": Picked up left fork");
synchronized (rightFork) {
doAction(System.nanoTime() + ": Picked up right fork - eating"); // eating
doAction(System.nanoTime() + ": Put down right fork");
}
doAction(System.nanoTime() + ": Put down left fork. Returning to thinking");
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}

14
core-java-concurrency/src/main/java/com/baeldung/concurrent/executor/ExecutorDemo.java Normal file
View File

@@ -0,0 +1,14 @@
package com.baeldung.concurrent.executor;
import java.util.concurrent.Executor;
public class ExecutorDemo {
public void execute() {
Executor executor = new Invoker();
executor.execute(()->{
// task to be performed
});
}
}

12
core-java-concurrency/src/main/java/com/baeldung/concurrent/executor/Invoker.java Normal file
View File

@@ -0,0 +1,12 @@
package com.baeldung.concurrent.executor;
import java.util.concurrent.Executor;
public class Invoker implements Executor {
@Override
public void execute(Runnable r) {
r.run();
}
}

27
core-java-concurrency/src/main/java/com/baeldung/concurrent/executorservice/ExecutorServiceDemo.java Normal file
View File

@@ -0,0 +1,27 @@
package com.baeldung.concurrent.executorservice;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class ExecutorServiceDemo {
ExecutorService executor = Executors.newFixedThreadPool(10);
public void execute() {
executor.submit(() -> {
new Task();
});
executor.shutdown();
executor.shutdownNow();
try {
executor.awaitTermination(20l, TimeUnit.NANOSECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

10
core-java-concurrency/src/main/java/com/baeldung/concurrent/executorservice/Task.java Normal file
View File

@@ -0,0 +1,10 @@
package com.baeldung.concurrent.executorservice;
public class Task implements Runnable {
@Override
public void run() {
// task details
}
}

26
core-java-concurrency/src/main/java/com/baeldung/concurrent/future/FactorialSquareCalculator.java Normal file
View File

@@ -0,0 +1,26 @@
package com.baeldung.concurrent.future;
import java.util.concurrent.RecursiveTask;
public class FactorialSquareCalculator extends RecursiveTask<Integer> {
private static final long serialVersionUID = 1L;
final private Integer n;
FactorialSquareCalculator(Integer n) {
this.n = n;
}
@Override
protected Integer compute() {
if (n <= 1) {
return n;
}
FactorialSquareCalculator calculator = new FactorialSquareCalculator(n - 1);
calculator.fork();
return n * n + calculator.join();
}
}

44
core-java-concurrency/src/main/java/com/baeldung/concurrent/future/FutureDemo.java Normal file
View File

@@ -0,0 +1,44 @@
package com.baeldung.concurrent.future;
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;
import java.util.concurrent.TimeoutException;
public class FutureDemo {
public String invoke() {
String str = null;
ExecutorService executorService = Executors.newFixedThreadPool(10);
Future<String> future = executorService.submit(() -> {
// Task
Thread.sleep(10000l);
return "Hellow world";
});
future.cancel(false);
try {
future.get(20, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e1) {
e1.printStackTrace();
}
if (future.isDone() && !future.isCancelled()) {
try {
str = future.get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
return str;
}
}

20
core-java-concurrency/src/main/java/com/baeldung/concurrent/future/SquareCalculator.java Normal file
View File

@@ -0,0 +1,20 @@
package com.baeldung.concurrent.future;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
class SquareCalculator {
private final ExecutorService executor;
SquareCalculator(ExecutorService executor) {
this.executor = executor;
}
Future<Integer> calculate(Integer input) {
return executor.submit(() -> {
Thread.sleep(1000);
return input * input;
});
}
}

83
core-java-concurrency/src/main/java/com/baeldung/concurrent/locks/ReentrantLockWithCondition.java Normal file
View File

@@ -0,0 +1,83 @@
package com.baeldung.concurrent.locks;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Stack;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import static java.lang.Thread.sleep;
public class ReentrantLockWithCondition {
private static Logger LOG = LoggerFactory.getLogger(ReentrantLockWithCondition.class);
private Stack<String> stack = new Stack<>();
private static final int CAPACITY = 5;
private ReentrantLock lock = new ReentrantLock();
private Condition stackEmptyCondition = lock.newCondition();
private Condition stackFullCondition = lock.newCondition();
private void pushToStack(String item) throws InterruptedException {
try {
lock.lock();
if (stack.size() == CAPACITY) {
LOG.info(Thread.currentThread().getName() + " wait on stack full");
stackFullCondition.await();
}
LOG.info("Pushing the item " + item);
stack.push(item);
stackEmptyCondition.signalAll();
} finally {
lock.unlock();
}
}
private String popFromStack() throws InterruptedException {
try {
lock.lock();
if (stack.size() == 0) {
LOG.info(Thread.currentThread().getName() + " wait on stack empty");
stackEmptyCondition.await();
}
return stack.pop();
} finally {
stackFullCondition.signalAll();
lock.unlock();
}
}
public static void main(String[] args) {
final int threadCount = 2;
ReentrantLockWithCondition object = new ReentrantLockWithCondition();
final ExecutorService service = Executors.newFixedThreadPool(threadCount);
service.execute(() -> {
for (int i = 0; i < 10; i++) {
try {
object.pushToStack("Item " + i);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
service.execute(() -> {
for (int i = 0; i < 10; i++) {
try {
LOG.info("Item popped " + object.popFromStack());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
service.shutdown();
}
}

88
core-java-concurrency/src/main/java/com/baeldung/concurrent/locks/SharedObjectWithLock.java Normal file
View File

@@ -0,0 +1,88 @@
package com.baeldung.concurrent.locks;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import static java.lang.Thread.sleep;
public class SharedObjectWithLock {
private static final Logger LOG = LoggerFactory.getLogger(SharedObjectWithLock.class);
private ReentrantLock lock = new ReentrantLock(true);
private int counter = 0;
void perform() {
lock.lock();
LOG.info("Thread - " + Thread.currentThread().getName() + " acquired the lock");
try {
LOG.info("Thread - " + Thread.currentThread().getName() + " processing");
counter++;
} catch (Exception exception) {
LOG.error(" Interrupted Exception ", exception);
} finally {
lock.unlock();
LOG.info("Thread - " + Thread.currentThread().getName() + " released the lock");
}
}
private void performTryLock() {
LOG.info("Thread - " + Thread.currentThread().getName() + " attempting to acquire the lock");
try {
boolean isLockAcquired = lock.tryLock(2, TimeUnit.SECONDS);
if (isLockAcquired) {
try {
LOG.info("Thread - " + Thread.currentThread().getName() + " acquired the lock");
LOG.info("Thread - " + Thread.currentThread().getName() + " processing");
sleep(1000);
} finally {
lock.unlock();
LOG.info("Thread - " + Thread.currentThread().getName() + " released the lock");
}
}
} catch (InterruptedException exception) {
LOG.error(" Interrupted Exception ", exception);
}
LOG.info("Thread - " + Thread.currentThread().getName() + " could not acquire the lock");
}
public ReentrantLock getLock() {
return lock;
}
boolean isLocked() {
return lock.isLocked();
}
boolean hasQueuedThreads() {
return lock.hasQueuedThreads();
}
int getCounter() {
return counter;
}
public static void main(String[] args) {
final int threadCount = 2;
final ExecutorService service = Executors.newFixedThreadPool(threadCount);
final SharedObjectWithLock object = new SharedObjectWithLock();
service.execute(object::perform);
service.execute(object::performTryLock);
service.shutdown();
}
}

104
core-java-concurrency/src/main/java/com/baeldung/concurrent/locks/StampedLockDemo.java Normal file
View File

@@ -0,0 +1,104 @@
package com.baeldung.concurrent.locks;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.StampedLock;
import static java.lang.Thread.sleep;
public class StampedLockDemo {
private Map<String, String> map = new HashMap<>();
private Logger logger = LoggerFactory.getLogger(StampedLockDemo.class);
private final StampedLock lock = new StampedLock();
public void put(String key, String value) throws InterruptedException {
long stamp = lock.writeLock();
try {
logger.info(Thread.currentThread().getName() + " acquired the write lock with stamp " + stamp);
map.put(key, value);
} finally {
lock.unlockWrite(stamp);
logger.info(Thread.currentThread().getName() + " unlocked the write lock with stamp " + stamp);
}
}
public String get(String key) throws InterruptedException {
long stamp = lock.readLock();
logger.info(Thread.currentThread().getName() + " acquired the read lock with stamp " + stamp);
try {
sleep(5000);
return map.get(key);
} finally {
lock.unlockRead(stamp);
logger.info(Thread.currentThread().getName() + " unlocked the read lock with stamp " + stamp);
}
}
private String readWithOptimisticLock(String key) throws InterruptedException {
long stamp = lock.tryOptimisticRead();
String value = map.get(key);
if (!lock.validate(stamp)) {
stamp = lock.readLock();
try {
sleep(5000);
return map.get(key);
} finally {
lock.unlock(stamp);
logger.info(Thread.currentThread().getName() + " unlocked the read lock with stamp " + stamp);
}
}
return value;
}
public static void main(String[] args) {
final int threadCount = 4;
final ExecutorService service = Executors.newFixedThreadPool(threadCount);
StampedLockDemo object = new StampedLockDemo();
Runnable writeTask = () -> {
try {
object.put("key1", "value1");
} catch (InterruptedException e) {
e.printStackTrace();
}
};
Runnable readTask = () -> {
try {
object.get("key1");
} catch (InterruptedException e) {
e.printStackTrace();
}
};
Runnable readOptimisticTask = () -> {
try {
object.readWithOptimisticLock("key1");
} catch (InterruptedException e) {
e.printStackTrace();
}
};
service.submit(writeTask);
service.submit(writeTask);
service.submit(readTask);
service.submit(readOptimisticTask);
service.shutdown();
}
}

120
core-java-concurrency/src/main/java/com/baeldung/concurrent/locks/SynchronizedHashMapWithRWLock.java Normal file
View File

@@ -0,0 +1,120 @@
package com.baeldung.concurrent.locks;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import static java.lang.Thread.sleep;
public class SynchronizedHashMapWithRWLock {
private static Map<String, String> syncHashMap = new HashMap<>();
private Logger logger = LoggerFactory.getLogger(SynchronizedHashMapWithRWLock.class);
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final Lock readLock = lock.readLock();
private final Lock writeLock = lock.writeLock();
public void put(String key, String value) throws InterruptedException {
try {
writeLock.lock();
logger.info(Thread.currentThread().getName() + " writing");
syncHashMap.put(key, value);
sleep(1000);
} finally {
writeLock.unlock();
}
}
public String get(String key) {
try {
readLock.lock();
logger.info(Thread.currentThread().getName() + " reading");
return syncHashMap.get(key);
} finally {
readLock.unlock();
}
}
public String remove(String key) {
try {
writeLock.lock();
return syncHashMap.remove(key);
} finally {
writeLock.unlock();
}
}
public boolean containsKey(String key) {
try {
readLock.lock();
return syncHashMap.containsKey(key);
} finally {
readLock.unlock();
}
}
boolean isReadLockAvailable() {
return readLock.tryLock();
}
public static void main(String[] args) throws InterruptedException {
final int threadCount = 3;
final ExecutorService service = Executors.newFixedThreadPool(threadCount);
SynchronizedHashMapWithRWLock object = new SynchronizedHashMapWithRWLock();
service.execute(new Thread(new Writer(object), "Writer"));
service.execute(new Thread(new Reader(object), "Reader1"));
service.execute(new Thread(new Reader(object), "Reader2"));
service.shutdown();
}
private static class Reader implements Runnable {
SynchronizedHashMapWithRWLock object;
Reader(SynchronizedHashMapWithRWLock object) {
this.object = object;
}
@Override
public void run() {
for (int i = 0; i < 10; i++) {
object.get("key" + i);
}
}
}
private static class Writer implements Runnable {
SynchronizedHashMapWithRWLock object;
public Writer(SynchronizedHashMapWithRWLock object) {
this.object = object;
}
@Override
public void run() {
for (int i = 0; i < 10; i++) {
try {
object.put("key" + i, "value" + i);
sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}

27
core-java-concurrency/src/main/java/com/baeldung/concurrent/phaser/LongRunningAction.java Normal file
View File

@@ -0,0 +1,27 @@
package com.baeldung.concurrent.phaser;
import java.util.concurrent.Phaser;
class LongRunningAction implements Runnable {
private String threadName;
private Phaser ph;
LongRunningAction(String threadName, Phaser ph) {
this.threadName = threadName;
this.ph = ph;
ph.register();
}
@Override
public void run() {
System.out.println("This is phase " + ph.getPhase());
System.out.println("Thread " + threadName + " before long running action");
ph.arriveAndAwaitAdvance();
try {
Thread.sleep(20);
} catch (InterruptedException e) {
e.printStackTrace();
}
ph.arriveAndDeregister();
}
}

22
core-java-concurrency/src/main/java/com/baeldung/concurrent/semaphore/SemaPhoreDemo.java Normal file
View File

@@ -0,0 +1,22 @@
package com.baeldung.concurrent.semaphore;
import java.util.concurrent.Semaphore;
public class SemaPhoreDemo {
static Semaphore semaphore = new Semaphore(10);
public void execute() throws InterruptedException {
System.out.println("Available permit : " + semaphore.availablePermits());
System.out.println("Number of threads waiting to acquire: " + semaphore.getQueueLength());
if (semaphore.tryAcquire()) {
semaphore.acquire();
// perform some critical operations
semaphore.release();
}
}
}

31
core-java-concurrency/src/main/java/com/baeldung/concurrent/semaphores/CounterUsingMutex.java Normal file
View File

@@ -0,0 +1,31 @@
package com.baeldung.concurrent.semaphores;
import java.util.concurrent.Semaphore;
class CounterUsingMutex {
private final Semaphore mutex;
private int count;
CounterUsingMutex() {
mutex = new Semaphore(1);
count = 0;
}
void increase() throws InterruptedException {
mutex.acquire();
this.count = this.count + 1;
Thread.sleep(1000);
mutex.release();
}
int getCount() {
return this.count;
}
boolean hasQueuedThreads() {
return mutex.hasQueuedThreads();
}
}

23
core-java-concurrency/src/main/java/com/baeldung/concurrent/semaphores/DelayQueueUsingTimedSemaphore.java Normal file
View File

@@ -0,0 +1,23 @@
package com.baeldung.concurrent.semaphores;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.concurrent.TimedSemaphore;
class DelayQueueUsingTimedSemaphore {
private final TimedSemaphore semaphore;
DelayQueueUsingTimedSemaphore(long period, int slotLimit) {
semaphore = new TimedSemaphore(period, TimeUnit.SECONDS, slotLimit);
}
boolean tryAdd() {
return semaphore.tryAcquire();
}
int availableSlots() {
return semaphore.getAvailablePermits();
}
}

25
core-java-concurrency/src/main/java/com/baeldung/concurrent/semaphores/LoginQueueUsingSemaphore.java Normal file
View File

@@ -0,0 +1,25 @@
package com.baeldung.concurrent.semaphores;
import java.util.concurrent.Semaphore;
class LoginQueueUsingSemaphore {
private final Semaphore semaphore;
LoginQueueUsingSemaphore(int slotLimit) {
semaphore = new Semaphore(slotLimit);
}
boolean tryLogin() {
return semaphore.tryAcquire();
}
void logout() {
semaphore.release();
}
int availableSlots() {
return semaphore.availablePermits();
}
}

21
core-java-concurrency/src/main/java/com/baeldung/concurrent/skiplist/Event.java Normal file
View File

@@ -0,0 +1,21 @@
package com.baeldung.concurrent.skiplist;
import java.time.ZonedDateTime;
class Event {
private final ZonedDateTime eventTime;
private final String content;
Event(ZonedDateTime eventTime, String content) {
this.eventTime = eventTime;
this.content = content;
}
ZonedDateTime getEventTime() {
return eventTime;
}
String getContent() {
return content;
}
}

29
core-java-concurrency/src/main/java/com/baeldung/concurrent/skiplist/EventWindowSort.java Normal file
View File

@@ -0,0 +1,29 @@
package com.baeldung.concurrent.skiplist;
import java.time.ZonedDateTime;
import java.util.Comparator;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
class EventWindowSort {
private final ConcurrentSkipListMap<ZonedDateTime, String> events
= new ConcurrentSkipListMap<>(Comparator.comparingLong(value -> value.toInstant().toEpochMilli()));
void acceptEvent(Event event) {
events.put(event.getEventTime(), event.getContent());
}
ConcurrentNavigableMap<ZonedDateTime, String> getEventsFromLastMinute() {
return events.tailMap(ZonedDateTime
.now()
.minusMinutes(1));
}
ConcurrentNavigableMap<ZonedDateTime, String> getEventsOlderThatOneMinute() {
return events.headMap(ZonedDateTime
.now()
.minusMinutes(1));
}
}

27
core-java-concurrency/src/main/java/com/baeldung/concurrent/sleepwait/ThreadA.java Normal file
View File

@@ -0,0 +1,27 @@
package com.baeldung.concurrent.sleepwait;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/***
* Example of waking up a waiting thread
*/
public class ThreadA {
private static final Logger LOG = LoggerFactory.getLogger(ThreadA.class);
private static final ThreadB b = new ThreadB();
public static void main(String... args) throws InterruptedException {
b.start();
synchronized (b) {
while (b.sum == 0) {
LOG.debug("Waiting for ThreadB to complete...");
b.wait();
}
LOG.debug("ThreadB has completed. Sum from that thread is: " + b.sum);
}
}
}

20
core-java-concurrency/src/main/java/com/baeldung/concurrent/sleepwait/ThreadB.java Normal file
View File

@@ -0,0 +1,20 @@
package com.baeldung.concurrent.sleepwait;
/***
* Example of waking up a waiting thread
*/
class ThreadB extends Thread {
int sum;
@Override
public void run() {
synchronized (this) {
int i = 0;
while (i < 100000) {
sum += i;
i++;
}
notify();
}
}
}

29
core-java-concurrency/src/main/java/com/baeldung/concurrent/sleepwait/WaitSleepExample.java Normal file
View File

@@ -0,0 +1,29 @@
package com.baeldung.concurrent.sleepwait;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/***
* Example of wait() and sleep() methods
*/
public class WaitSleepExample {
private static final Logger LOG = LoggerFactory.getLogger(WaitSleepExample.class);
private static final Object LOCK = new Object();
public static void main(String... args) throws InterruptedException {
sleepWaitInSynchronizedBlocks();
}
private static void sleepWaitInSynchronizedBlocks() throws InterruptedException {
Thread.sleep(1000); // called on the thread
LOG.debug("Thread '" + Thread.currentThread().getName() + "' is woken after sleeping for 1 second");
synchronized (LOCK) {
LOCK.wait(1000); // called on the object, synchronization required
LOG.debug("Object '" + LOCK + "' is woken after waiting for 1 second");
}
}
}

35
core-java-concurrency/src/main/java/com/baeldung/concurrent/synchronize/BaeldungSynchronizedBlocks.java Normal file
View File

@@ -0,0 +1,35 @@
package com.baeldung.concurrent.synchronize;
public class BaeldungSynchronizedBlocks {
private int count = 0;
private static int staticCount = 0;
void performSynchronisedTask() {
synchronized (this) {
setCount(getCount() + 1);
}
}
static void performStaticSyncTask() {
synchronized (BaeldungSynchronizedBlocks.class) {
setStaticCount(getStaticCount() + 1);
}
}
public int getCount() {
return count;
}
public void setCount(int count) {
this.count = count;
}
static int getStaticCount() {
return staticCount;
}
private static void setStaticCount(int staticCount) {
BaeldungSynchronizedBlocks.staticCount = staticCount;
}
}

37
core-java-concurrency/src/main/java/com/baeldung/concurrent/synchronize/BaeldungSynchronizedMethods.java Normal file
View File

@@ -0,0 +1,37 @@
package com.baeldung.concurrent.synchronize;
public class BaeldungSynchronizedMethods {
private int sum = 0;
private int syncSum = 0;
static int staticSum = 0;
void calculate() {
setSum(getSum() + 1);
}
synchronized void synchronisedCalculate() {
setSyncSum(getSyncSum() + 1);
}
static synchronized void syncStaticCalculate() {
staticSum = staticSum + 1;
}
public int getSum() {
return sum;
}
public void setSum(int sum) {
this.sum = sum;
}
int getSyncSum() {
return syncSum;
}
private void setSyncSum(int syncSum) {
this.syncSum = syncSum;
}
}

23
core-java-concurrency/src/main/java/com/baeldung/concurrent/threadfactory/BaeldungThreadFactory.java Normal file
View File

@@ -0,0 +1,23 @@
package com.baeldung.concurrent.threadfactory;
import java.util.concurrent.ThreadFactory;
public class BaeldungThreadFactory implements ThreadFactory {
private int threadId;
private String name;
public BaeldungThreadFactory(String name) {
threadId = 1;
this.name = name;
}
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r, name + "-Thread_" + threadId);
System.out.println("created new thread with id : " + threadId + " and name : " + t.getName());
threadId++;
return t;
}
}

13
core-java-concurrency/src/main/java/com/baeldung/concurrent/threadfactory/Demo.java Normal file
View File

@@ -0,0 +1,13 @@
package com.baeldung.concurrent.threadfactory;
public class Demo {
public void execute() {
BaeldungThreadFactory factory = new BaeldungThreadFactory("BaeldungThreadFactory");
for (int i = 0; i < 10; i++) {
Thread t = factory.newThread(new Task());
t.start();
}
}
}

10
core-java-concurrency/src/main/java/com/baeldung/concurrent/threadfactory/Task.java Normal file
View File

@@ -0,0 +1,10 @@
package com.baeldung.concurrent.threadfactory;
public class Task implements Runnable {
@Override
public void run() {
// task details
}
}

13
core-java-concurrency/src/main/java/com/baeldung/concurrent/volatilekeyword/SharedObject.java Normal file
View File

@@ -0,0 +1,13 @@
package com.baeldung.concurrent.volatilekeyword;
public class SharedObject {
private volatile int count=0;
void increamentCount(){
count++;
}
public int getCount(){
return count;
}
}

17
core-java-concurrency/src/main/java/com/baeldung/threadlocal/Context.java Normal file
View File

@@ -0,0 +1,17 @@
package com.baeldung.threadlocal;
public class Context {
private final String userName;
public Context(String userName) {
this.userName = userName;
}
@Override
public String toString() {
return "Context{" +
"userNameSecret='" + userName + '\'' +
'}';
}
}

21
core-java-concurrency/src/main/java/com/baeldung/threadlocal/SharedMapWithUserContext.java Normal file
View File

@@ -0,0 +1,21 @@
package com.baeldung.threadlocal;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class SharedMapWithUserContext implements Runnable {
public final static Map<Integer, Context> userContextPerUserId = new ConcurrentHashMap<>();
private final Integer userId;
private UserRepository userRepository = new UserRepository();
public SharedMapWithUserContext(Integer userId) {
this.userId = userId;
}
@Override
public void run() {
String userName = userRepository.getUserNameForUserId(userId);
userContextPerUserId.put(userId, new Context(userName));
}
}

24
core-java-concurrency/src/main/java/com/baeldung/threadlocal/ThreadLocalWithUserContext.java Normal file
View File

@@ -0,0 +1,24 @@
package com.baeldung.threadlocal;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ThreadLocalWithUserContext implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(ThreadLocalWithUserContext.class);
private static final ThreadLocal<Context> userContext = new ThreadLocal<>();
private final Integer userId;
private UserRepository userRepository = new UserRepository();
public ThreadLocalWithUserContext(Integer userId) {
this.userId = userId;
}
@Override
public void run() {
String userName = userRepository.getUserNameForUserId(userId);
userContext.set(new Context(userName));
LOG.debug("thread context for given userId: " + userId + " is: " + userContext.get());
}
}

10
core-java-concurrency/src/main/java/com/baeldung/threadlocal/UserRepository.java Normal file
View File

@@ -0,0 +1,10 @@
package com.baeldung.threadlocal;
import java.util.UUID;
public class UserRepository {
public String getUserNameForUserId(Integer userId) {
return UUID.randomUUID().toString();
}
}

20
core-java-concurrency/src/main/java/com/baeldung/threadpool/CountingTask.java Normal file
View File

@@ -0,0 +1,20 @@
package com.baeldung.threadpool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.stream.Collectors;
public class CountingTask extends RecursiveTask<Integer> {
private final TreeNode node;
public CountingTask(TreeNode node) {
this.node = node;
}
@Override
protected Integer compute() {
return node.value + node.children.stream().map(childNode -> new CountingTask(childNode).fork()).collect(Collectors.summingInt(ForkJoinTask::join));
}
}

29
core-java-concurrency/src/main/java/com/baeldung/threadpool/ExitingExecutorServiceExample.java Normal file
View File

@@ -0,0 +1,29 @@
package com.baeldung.threadpool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import com.google.common.util.concurrent.MoreExecutors;
/**
* This class demonstrates the usage of Guava's exiting executor services that keep the VM from hanging.
* Without the exiting executor service, the task would hang indefinitely.
* This behaviour cannot be demonstrated in JUnit tests, as JUnit kills the VM after the tests.
*/
public class ExitingExecutorServiceExample {
public static void main(String... args) {
final ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(5);
final ExecutorService executorService = MoreExecutors.getExitingExecutorService(executor, 100, TimeUnit.MILLISECONDS);
executorService.submit((Runnable) () -> {
while (true) {
}
});
}
}

18
core-java-concurrency/src/main/java/com/baeldung/threadpool/TreeNode.java Normal file
View File

@@ -0,0 +1,18 @@
package com.baeldung.threadpool;
import java.util.Set;
import com.google.common.collect.Sets;
public class TreeNode {
int value;
Set<TreeNode> children;
public TreeNode(int value, TreeNode... children) {
this.value = value;
this.children = Sets.newHashSet(children);
}
}

41
core-java-concurrency/src/main/java/com/baeldung/transferqueue/Consumer.java Normal file
View File

@@ -0,0 +1,41 @@
package com.baeldung.transferqueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.TransferQueue;
import java.util.concurrent.atomic.AtomicInteger;
public class Consumer implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(Consumer.class);
private final TransferQueue<String> transferQueue;
private final String name;
private final int numberOfMessagesToConsume;
public final AtomicInteger numberOfConsumedMessages = new AtomicInteger();
public Consumer(TransferQueue<String> transferQueue, String name, int numberOfMessagesToConsume) {
this.transferQueue = transferQueue;
this.name = name;
this.numberOfMessagesToConsume = numberOfMessagesToConsume;
}
@Override
public void run() {
for (int i = 0; i < numberOfMessagesToConsume; i++) {
try {
LOG.debug("Consumer: " + name + " is waiting to take element...");
String element = transferQueue.take();
longProcessing(element);
LOG.debug("Consumer: " + name + " received element: " + element);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
private void longProcessing(String element) throws InterruptedException {
numberOfConsumedMessages.incrementAndGet();
Thread.sleep(500);
}
}

41
core-java-concurrency/src/main/java/com/baeldung/transferqueue/Producer.java Normal file
View File

@@ -0,0 +1,41 @@
package com.baeldung.transferqueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TransferQueue;
import java.util.concurrent.atomic.AtomicInteger;
public class Producer implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(Producer.class);
private final TransferQueue<String> transferQueue;
private final String name;
private final Integer numberOfMessagesToProduce;
public final AtomicInteger numberOfProducedMessages = new AtomicInteger();
public Producer(TransferQueue<String> transferQueue, String name, Integer numberOfMessagesToProduce) {
this.transferQueue = transferQueue;
this.name = name;
this.numberOfMessagesToProduce = numberOfMessagesToProduce;
}
@Override
public void run() {
for (int i = 0; i < numberOfMessagesToProduce; i++) {
try {
LOG.debug("Producer: " + name + " is waiting to transfer...");
boolean added = transferQueue.tryTransfer("A" + i, 4000, TimeUnit.MILLISECONDS);
if (added) {
numberOfProducedMessages.incrementAndGet();
LOG.debug("Producer: " + name + " transferred element: A" + i);
} else {
LOG.debug("can not add an element due to the timeout");
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

9
core-java-concurrency/src/main/java/log4j.properties Normal file
View File

@@ -0,0 +1,9 @@
# Set root logger level to DEBUG and its only appender to A1.
log4j.rootLogger=DEBUG, A1
# A1 is set to be a ConsoleAppender.
log4j.appender.A1=org.apache.log4j.ConsoleAppender
# A1 uses PatternLayout.
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n

19
core-java-concurrency/src/main/resources/logback.xml Normal file
View File

@@ -0,0 +1,19 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>web - %date [%thread] %-5level %logger{36} - %message%n
</pattern>
</encoder>
</appender>
<logger name="org.springframework" level="WARN" />
<logger name="org.springframework.transaction" level="WARN" />
<!-- in order to debug some marshalling issues, this needs to be TRACE -->
<logger name="org.springframework.web.servlet.mvc" level="WARN" />
<root level="INFO">
<appender-ref ref="STDOUT" />
</root>
</configuration>

184
core-java-concurrency/src/test/java/com/baeldung/completablefuture/CompletableFutureLongRunningUnitTest.java Normal file
View File

@@ -0,0 +1,184 @@
package com.baeldung.completablefuture;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class CompletableFutureLongRunningUnitTest {
private static final Logger LOG = LoggerFactory.getLogger(CompletableFutureLongRunningUnitTest.class);
@Test
public void whenRunningCompletableFutureAsynchronously_thenGetMethodWaitsForResult() throws InterruptedException, ExecutionException {
Future<String> completableFuture = calculateAsync();
String result = completableFuture.get();
assertEquals("Hello", result);
}
private Future<String> calculateAsync() throws InterruptedException {
CompletableFuture<String> completableFuture = new CompletableFuture<>();
Executors.newCachedThreadPool()
.submit(() -> {
Thread.sleep(500);
completableFuture.complete("Hello");
return null;
});
return completableFuture;
}
@Test
public void whenRunningCompletableFutureWithResult_thenGetMethodReturnsImmediately() throws InterruptedException, ExecutionException {
Future<String> completableFuture = CompletableFuture.completedFuture("Hello");
String result = completableFuture.get();
assertEquals("Hello", result);
}
private Future<String> calculateAsyncWithCancellation() throws InterruptedException {
CompletableFuture<String> completableFuture = new CompletableFuture<>();
Executors.newCachedThreadPool()
.submit(() -> {
Thread.sleep(500);
completableFuture.cancel(false);
return null;
});
return completableFuture;
}
@Test(expected = CancellationException.class)
public void whenCancelingTheFuture_thenThrowsCancellationException() throws ExecutionException, InterruptedException {
Future<String> future = calculateAsyncWithCancellation();
future.get();
}
@Test
public void whenCreatingCompletableFutureWithSupplyAsync_thenFutureReturnsValue() throws ExecutionException, InterruptedException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");
assertEquals("Hello", future.get());
}
@Test
public void whenAddingThenAcceptToFuture_thenFunctionExecutesAfterComputationIsFinished() throws ExecutionException, InterruptedException {
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<Void> future = completableFuture.thenAccept(s -> LOG.debug("Computation returned: " + s));
future.get();
}
@Test
public void whenAddingThenRunToFuture_thenFunctionExecutesAfterComputationIsFinished() throws ExecutionException, InterruptedException {
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<Void> future = completableFuture.thenRun(() -> LOG.debug("Computation finished."));
future.get();
}
@Test
public void whenAddingThenApplyToFuture_thenFunctionExecutesAfterComputationIsFinished() throws ExecutionException, InterruptedException {
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future = completableFuture.thenApply(s -> s + " World");
assertEquals("Hello World", future.get());
}
@Test
public void whenUsingThenCompose_thenFuturesExecuteSequentially() throws ExecutionException, InterruptedException {
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello")
.thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " World"));
assertEquals("Hello World", completableFuture.get());
}
@Test
public void whenUsingThenCombine_thenWaitForExecutionOfBothFutures() throws ExecutionException, InterruptedException {
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello")
.thenCombine(CompletableFuture.supplyAsync(() -> " World"), (s1, s2) -> s1 + s2);
assertEquals("Hello World", completableFuture.get());
}
@Test
public void whenUsingThenAcceptBoth_thenWaitForExecutionOfBothFutures() throws ExecutionException, InterruptedException {
CompletableFuture.supplyAsync(() -> "Hello")
.thenAcceptBoth(CompletableFuture.supplyAsync(() -> " World"), (s1, s2) -> LOG.debug(s1 + s2));
}
@Test
public void whenFutureCombinedWithAllOfCompletes_thenAllFuturesAreDone() throws ExecutionException, InterruptedException {
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "Beautiful");
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "World");
CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(future1, future2, future3);
// ...
combinedFuture.get();
assertTrue(future1.isDone());
assertTrue(future2.isDone());
assertTrue(future3.isDone());
String combined = Stream.of(future1, future2, future3)
.map(CompletableFuture::join)
.collect(Collectors.joining(" "));
assertEquals("Hello Beautiful World", combined);
}
@Test
public void whenFutureThrows_thenHandleMethodReceivesException() throws ExecutionException, InterruptedException {
String name = null;
// ...
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
if (name == null) {
throw new RuntimeException("Computation error!");
}
return "Hello, " + name;
})
.handle((s, t) -> s != null ? s : "Hello, Stranger!");
assertEquals("Hello, Stranger!", completableFuture.get());
}
@Test(expected = ExecutionException.class)
public void whenCompletingFutureExceptionally_thenGetMethodThrows() throws ExecutionException, InterruptedException {
CompletableFuture<String> completableFuture = new CompletableFuture<>();
// ...
completableFuture.completeExceptionally(new RuntimeException("Calculation failed!"));
// ...
completableFuture.get();
}
@Test
public void whenAddingThenApplyAsyncToFuture_thenFunctionExecutesAfterComputationIsFinished() throws ExecutionException, InterruptedException {
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future = completableFuture.thenApplyAsync(s -> s + " World");
assertEquals("Hello World", future.get());
}
}

38
core-java-concurrency/src/test/java/com/baeldung/concurrent/accumulator/LongAccumulatorUnitTest.java Normal file
View File

@@ -0,0 +1,38 @@
package com.baeldung.concurrent.accumulator;
import org.junit.Test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAccumulator;
import java.util.function.LongBinaryOperator;
import java.util.stream.IntStream;
import static junit.framework.TestCase.assertEquals;
public class LongAccumulatorUnitTest {
@Test
public void givenLongAccumulator_whenApplyActionOnItFromMultipleThrads_thenShouldProduceProperResult() throws InterruptedException {
// given
ExecutorService executorService = Executors.newFixedThreadPool(8);
LongBinaryOperator sum = Long::sum;
LongAccumulator accumulator = new LongAccumulator(sum, 0L);
int numberOfThreads = 4;
int numberOfIncrements = 100;
// when
Runnable accumulateAction = () -> IntStream.rangeClosed(0, numberOfIncrements).forEach(accumulator::accumulate);
for (int i = 0; i < numberOfThreads; i++) {
executorService.execute(accumulateAction);
}
// then
executorService.awaitTermination(500, TimeUnit.MILLISECONDS);
executorService.shutdown();
assertEquals(accumulator.get(), 20200);
}
}

67
core-java-concurrency/src/test/java/com/baeldung/concurrent/adder/LongAdderUnitTest.java Normal file
View File

@@ -0,0 +1,67 @@
package com.baeldung.concurrent.adder;
import org.junit.Test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.IntStream;
import static com.jayway.awaitility.Awaitility.await;
import static junit.framework.TestCase.assertEquals;
public class LongAdderUnitTest {
@Test
public void givenMultipleThread_whenTheyWriteToSharedLongAdder_thenShouldCalculateSumForThem() throws InterruptedException {
//given
LongAdder counter = new LongAdder();
ExecutorService executorService = Executors.newFixedThreadPool(8);
int numberOfThreads = 4;
int numberOfIncrements = 100;
//when
Runnable incrementAction = () -> IntStream
.range(0, numberOfIncrements)
.forEach((i) -> counter.increment());
for (int i = 0; i < numberOfThreads; i++) {
executorService.execute(incrementAction);
}
//then
executorService.awaitTermination(500, TimeUnit.MILLISECONDS);
executorService.shutdown();
assertEquals(counter.sum(), numberOfIncrements * numberOfThreads);
assertEquals(counter.sum(), numberOfIncrements * numberOfThreads);
}
@Test
public void givenMultipleThread_whenTheyWriteToSharedLongAdder_thenShouldCalculateSumForThemAndResetAdderAfterward() throws InterruptedException {
//given
LongAdder counter = new LongAdder();
ExecutorService executorService = Executors.newFixedThreadPool(8);
int numberOfThreads = 4;
int numberOfIncrements = 100;
//when
Runnable incrementAction = () -> IntStream
.range(0, numberOfIncrements)
.forEach((i) -> counter.increment());
for (int i = 0; i < numberOfThreads; i++) {
executorService.execute(incrementAction);
}
//then
executorService.awaitTermination(500, TimeUnit.MILLISECONDS);
executorService.shutdown();
assertEquals(counter.sumThenReset(), numberOfIncrements * numberOfThreads);
await().until(() -> assertEquals(counter.sum(), 0));
}
}

38
core-java-concurrency/src/test/java/com/baeldung/concurrent/atomic/ThreadSafeCounterTest.java Normal file
View File

@@ -0,0 +1,38 @@
package com.baeldung.concurrent.atomic;
import static org.junit.Assert.assertEquals;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
import org.junit.Test;
public class ThreadSafeCounterTest {
@Test
public void givenMultiThread_whenSafeCounterWithLockIncrement() throws InterruptedException {
ExecutorService service = Executors.newFixedThreadPool(3);
SafeCounterWithLock safeCounter = new SafeCounterWithLock();
IntStream.range(0, 1000)
.forEach(count -> service.submit(safeCounter::increment));
service.awaitTermination(100, TimeUnit.MILLISECONDS);
assertEquals(1000, safeCounter.getValue());
}
@Test
public void givenMultiThread_whenSafeCounterWithoutLockIncrement() throws InterruptedException {
ExecutorService service = Executors.newFixedThreadPool(3);
SafeCounterWithoutLock safeCounter = new SafeCounterWithoutLock();
IntStream.range(0, 1000)
.forEach(count -> service.submit(safeCounter::increment));
service.awaitTermination(100, TimeUnit.MILLISECONDS);
assertEquals(1000, safeCounter.getValue());
}
}

33
core-java-concurrency/src/test/java/com/baeldung/concurrent/atomic/ThreadUnsafeCounterManualTest.java Normal file
View File

@@ -0,0 +1,33 @@
package com.baeldung.concurrent.atomic;
import static org.junit.Assert.assertEquals;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
import org.junit.Test;
/**
* This test shows the behaviour of a thread-unsafe class in a multithreaded scenario. We are calling
* the increment methods 1000 times from a pool of 3 threads. In most of the cases, the counter will
* less than 1000, because of lost updates, however, occasionally it may reach 1000, when no threads
* called the method simultaneously. This may cause the build to fail occasionally. Hence excluding this
* test from build by adding this in manual test
*/
public class ThreadUnsafeCounterManualTest {
@Test
public void givenMultiThread_whenUnsafeCounterIncrement() throws InterruptedException {
ExecutorService service = Executors.newFixedThreadPool(3);
UnsafeCounter unsafeCounter = new UnsafeCounter();
IntStream.range(0, 1000)
.forEach(count -> service.submit(unsafeCounter::increment));
service.awaitTermination(100, TimeUnit.MILLISECONDS);
assertEquals(1000, unsafeCounter.getValue());
}
}

53
core-java-concurrency/src/test/java/com/baeldung/concurrent/copyonwrite/CopyOnWriteArrayListUnitTest.java Normal file
View File

@@ -0,0 +1,53 @@
package com.baeldung.concurrent.copyonwrite;
import org.junit.Test;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import static org.assertj.core.api.Assertions.assertThat;
public class CopyOnWriteArrayListUnitTest {
@Test
public void givenCopyOnWriteList_whenIterateAndAddElementToUnderneathList_thenShouldNotChangeIterator() {
//given
final CopyOnWriteArrayList<Integer> numbers =
new CopyOnWriteArrayList<>(new Integer[]{1, 3, 5, 8});
//when
Iterator<Integer> iterator = numbers.iterator();
numbers.add(10);
//then
List<Integer> result = new LinkedList<>();
iterator.forEachRemaining(result::add);
assertThat(result).containsOnly(1, 3, 5, 8);
//and
Iterator<Integer> iterator2 = numbers.iterator();
List<Integer> result2 = new LinkedList<>();
iterator2.forEachRemaining(result2::add);
//then
assertThat(result2).containsOnly(1, 3, 5, 8, 10);
}
@Test(expected = UnsupportedOperationException.class)
public void givenCopyOnWriteList_whenIterateOverItAndTryToRemoveElement_thenShouldThrowException() {
//given
final CopyOnWriteArrayList<Integer> numbers =
new CopyOnWriteArrayList<>(new Integer[]{1, 3, 5, 8});
//when
Iterator<Integer> iterator = numbers.iterator();
while (iterator.hasNext()) {
iterator.remove();
}
}
}

70
core-java-concurrency/src/test/java/com/baeldung/concurrent/countdownlatch/CountdownLatchExampleIntegrationTest.java Normal file
View File

@@ -0,0 +1,70 @@
package com.baeldung.concurrent.countdownlatch;
import org.junit.Test;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import static java.util.stream.Collectors.toList;
import static org.assertj.core.api.Assertions.assertThat;
public class CountdownLatchExampleIntegrationTest {
@Test
public void whenParallelProcessing_thenMainThreadWillBlockUntilCompletion() throws InterruptedException {
// Given
List<String> outputScraper = Collections.synchronizedList(new ArrayList<>());
CountDownLatch countDownLatch = new CountDownLatch(5);
List<Thread> workers = Stream.generate(() -> new Thread(new Worker(outputScraper, countDownLatch))).limit(5).collect(toList());
// When
workers.forEach(Thread::start);
countDownLatch.await(); // Block until workers finish
outputScraper.add("Latch released");
// Then
outputScraper.forEach(Object::toString);
assertThat(outputScraper).containsExactly("Counted down", "Counted down", "Counted down", "Counted down", "Counted down", "Latch released");
}
@Test
public void whenFailingToParallelProcess_thenMainThreadShouldTimeout() throws InterruptedException {
// Given
List<String> outputScraper = Collections.synchronizedList(new ArrayList<>());
CountDownLatch countDownLatch = new CountDownLatch(5);
List<Thread> workers = Stream.generate(() -> new Thread(new BrokenWorker(outputScraper, countDownLatch))).limit(5).collect(toList());
// When
workers.forEach(Thread::start);
final boolean result = countDownLatch.await(3L, TimeUnit.SECONDS);
// Then
assertThat(result).isFalse();
}
@Test
public void whenDoingLotsOfThreadsInParallel_thenStartThemAtTheSameTime() throws InterruptedException {
// Given
List<String> outputScraper = Collections.synchronizedList(new ArrayList<>());
CountDownLatch readyThreadCounter = new CountDownLatch(5);
CountDownLatch callingThreadBlocker = new CountDownLatch(1);
CountDownLatch completedThreadCounter = new CountDownLatch(5);
List<Thread> workers = Stream.generate(() -> new Thread(new WaitingWorker(outputScraper, readyThreadCounter, callingThreadBlocker, completedThreadCounter))).limit(5).collect(toList());
// When
workers.forEach(Thread::start);
readyThreadCounter.await(); // Block until workers start
outputScraper.add("Workers ready");
callingThreadBlocker.countDown(); // Start workers
completedThreadCounter.await(); // Block until workers finish
outputScraper.add("Workers complete");
// Then
outputScraper.forEach(Object::toString);
assertThat(outputScraper).containsExactly("Workers ready", "Counted down", "Counted down", "Counted down", "Counted down", "Counted down", "Workers complete");
}
}

82
core-java-concurrency/src/test/java/com/baeldung/concurrent/delayqueue/DelayQueueIntegrationTest.java Normal file
View File

@@ -0,0 +1,82 @@
package com.baeldung.concurrent.delayqueue;
import org.junit.FixMethodOrder;
import org.junit.Test;
import org.junit.runners.MethodSorters;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import static junit.framework.TestCase.assertEquals;
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
public class DelayQueueIntegrationTest {
@Test
public void givenDelayQueue_whenProduceElement_thenShouldConsumeAfterGivenDelay() throws InterruptedException {
//given
ExecutorService executor = Executors.newFixedThreadPool(2);
BlockingQueue<DelayObject> queue = new DelayQueue<>();
int numberOfElementsToProduce = 2;
int delayOfEachProducedMessageMilliseconds = 500;
DelayQueueConsumer consumer = new DelayQueueConsumer(queue, numberOfElementsToProduce);
DelayQueueProducer producer
= new DelayQueueProducer(queue, numberOfElementsToProduce, delayOfEachProducedMessageMilliseconds);
//when
executor.submit(producer);
executor.submit(consumer);
//then
executor.awaitTermination(5, TimeUnit.SECONDS);
executor.shutdown();
assertEquals(consumer.numberOfConsumedElements.get(), numberOfElementsToProduce);
}
@Test
public void givenDelayQueue_whenProduceElementWithHugeDelay_thenConsumerWasNotAbleToConsumeMessageInGivenTime() throws InterruptedException {
//given
ExecutorService executor = Executors.newFixedThreadPool(2);
BlockingQueue<DelayObject> queue = new DelayQueue<>();
int numberOfElementsToProduce = 1;
int delayOfEachProducedMessageMilliseconds = 10_000;
DelayQueueConsumer consumer = new DelayQueueConsumer(queue, numberOfElementsToProduce);
DelayQueueProducer producer
= new DelayQueueProducer(queue, numberOfElementsToProduce, delayOfEachProducedMessageMilliseconds);
//when
executor.submit(producer);
executor.submit(consumer);
//then
executor.awaitTermination(5, TimeUnit.SECONDS);
executor.shutdown();
assertEquals(consumer.numberOfConsumedElements.get(), 0);
}
@Test
public void givenDelayQueue_whenProduceElementWithNegativeDelay_thenConsumeMessageImmediately() throws InterruptedException {
//given
ExecutorService executor = Executors.newFixedThreadPool(2);
BlockingQueue<DelayObject> queue = new DelayQueue<>();
int numberOfElementsToProduce = 1;
int delayOfEachProducedMessageMilliseconds = -10_000;
DelayQueueConsumer consumer = new DelayQueueConsumer(queue, numberOfElementsToProduce);
DelayQueueProducer producer
= new DelayQueueProducer(queue, numberOfElementsToProduce, delayOfEachProducedMessageMilliseconds);
//when
executor.submit(producer);
executor.submit(consumer);
//then
executor.awaitTermination(1, TimeUnit.SECONDS);
executor.shutdown();
assertEquals(consumer.numberOfConsumedElements.get(), 1);
}
}

22
core-java-concurrency/src/test/java/com/baeldung/concurrent/future/FactorialSquareCalculatorUnitTest.java Normal file
View File

@@ -0,0 +1,22 @@
package com.baeldung.concurrent.future;
import org.junit.Test;
import java.util.concurrent.ForkJoinPool;
import static org.junit.Assert.assertEquals;
public class FactorialSquareCalculatorUnitTest {
@Test
public void whenCalculatesFactorialSquare_thenReturnCorrectValue() {
ForkJoinPool forkJoinPool = new ForkJoinPool();
FactorialSquareCalculator calculator = new FactorialSquareCalculator(10);
forkJoinPool.execute(calculator);
assertEquals("The sum of the squares from 1 to 10 is 385", 385, calculator.join().intValue());
}
}

99
core-java-concurrency/src/test/java/com/baeldung/concurrent/future/SquareCalculatorIntegrationTest.java Normal file
View File

@@ -0,0 +1,99 @@
package com.baeldung.concurrent.future;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class SquareCalculatorIntegrationTest {
private static final Logger LOG = LoggerFactory.getLogger(SquareCalculatorIntegrationTest.class);
@Rule
public TestName name = new TestName();
private long start;
private SquareCalculator squareCalculator;
@Test
public void givenExecutorIsSingleThreaded_whenTwoExecutionsAreTriggered_thenRunInSequence() throws InterruptedException, ExecutionException {
squareCalculator = new SquareCalculator(Executors.newSingleThreadExecutor());
Future<Integer> result1 = squareCalculator.calculate(4);
Future<Integer> result2 = squareCalculator.calculate(1000);
while (!result1.isDone() || !result2.isDone()) {
LOG.debug(String.format("Task 1 is %s and Task 2 is %s.", result1.isDone() ? "done" : "not done", result2.isDone() ? "done" : "not done"));
Thread.sleep(300);
}
assertEquals(16, result1.get().intValue());
assertEquals(1000000, result2.get().intValue());
}
@Test(expected = TimeoutException.class)
public void whenGetWithTimeoutLowerThanExecutionTime_thenThrowException() throws InterruptedException, ExecutionException, TimeoutException {
squareCalculator = new SquareCalculator(Executors.newSingleThreadExecutor());
Future<Integer> result = squareCalculator.calculate(4);
result.get(500, TimeUnit.MILLISECONDS);
}
@Test
public void givenExecutorIsMultiThreaded_whenTwoExecutionsAreTriggered_thenRunInParallel() throws InterruptedException, ExecutionException {
squareCalculator = new SquareCalculator(Executors.newFixedThreadPool(2));
Future<Integer> result1 = squareCalculator.calculate(4);
Future<Integer> result2 = squareCalculator.calculate(1000);
while (!result1.isDone() || !result2.isDone()) {
LOG.debug(String.format("Task 1 is %s and Task 2 is %s.", result1.isDone() ? "done" : "not done", result2.isDone() ? "done" : "not done"));
Thread.sleep(300);
}
assertEquals(16, result1.get().intValue());
assertEquals(1000000, result2.get().intValue());
}
@Test(expected = CancellationException.class)
public void whenCancelFutureAndCallGet_thenThrowException() throws InterruptedException, ExecutionException, TimeoutException {
squareCalculator = new SquareCalculator(Executors.newSingleThreadExecutor());
Future<Integer> result = squareCalculator.calculate(4);
boolean canceled = result.cancel(true);
assertTrue("Future was canceled", canceled);
assertTrue("Future was canceled", result.isCancelled());
result.get();
}
@Before
public void start() {
start = System.currentTimeMillis();
}
@After
public void end() {
LOG.debug(String.format("Test %s took %s ms \n", name.getMethodName(), System.currentTimeMillis() - start));
}
}

73
core-java-concurrency/src/test/java/com/baeldung/concurrent/locks/SharedObjectWithLockManualTest.java Normal file
View File

@@ -0,0 +1,73 @@
package com.baeldung.concurrent.locks;
import org.junit.Test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import static junit.framework.TestCase.assertEquals;
public class SharedObjectWithLockManualTest {
@Test
public void whenLockAcquired_ThenLockedIsTrue() {
final SharedObjectWithLock object = new SharedObjectWithLock();
final int threadCount = 2;
final ExecutorService service = Executors.newFixedThreadPool(threadCount);
executeThreads(object, threadCount, service);
assertEquals(true, object.isLocked());
service.shutdown();
}
@Test
public void whenLocked_ThenQueuedThread() {
final int threadCount = 4;
final ExecutorService service = Executors.newFixedThreadPool(threadCount);
final SharedObjectWithLock object = new SharedObjectWithLock();
executeThreads(object, threadCount, service);
assertEquals(object.hasQueuedThreads(), true);
service.shutdown();
}
public void whenTryLock_ThenQueuedThread() {
final SharedObjectWithLock object = new SharedObjectWithLock();
final int threadCount = 2;
final ExecutorService service = Executors.newFixedThreadPool(threadCount);
executeThreads(object, threadCount, service);
assertEquals(true, object.isLocked());
service.shutdown();
}
@Test
public void whenGetCount_ThenCorrectCount() throws InterruptedException {
final int threadCount = 4;
final ExecutorService service = Executors.newFixedThreadPool(threadCount);
final SharedObjectWithLock object = new SharedObjectWithLock();
executeThreads(object, threadCount, service);
Thread.sleep(1000);
assertEquals(object.getCounter(), 4);
service.shutdown();
}
private void executeThreads(SharedObjectWithLock object, int threadCount, ExecutorService service) {
for (int i = 0; i < threadCount; i++) {
service.execute(object::perform);
}
}
}

57
core-java-concurrency/src/test/java/com/baeldung/concurrent/locks/SynchronizedHashMapWithRWLockManualTest.java Normal file
View File

@@ -0,0 +1,57 @@
package com.baeldung.concurrent.locks;
import org.junit.Test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import static junit.framework.TestCase.assertEquals;
public class SynchronizedHashMapWithRWLockManualTest {
@Test
public void whenWriting_ThenNoReading() {
SynchronizedHashMapWithRWLock object = new SynchronizedHashMapWithRWLock();
final int threadCount = 3;
final ExecutorService service = Executors.newFixedThreadPool(threadCount);
executeWriterThreads(object, threadCount, service);
assertEquals(object.isReadLockAvailable(), false);
service.shutdown();
}
@Test
public void whenReading_ThenMultipleReadingAllowed() {
SynchronizedHashMapWithRWLock object = new SynchronizedHashMapWithRWLock();
final int threadCount = 5;
final ExecutorService service = Executors.newFixedThreadPool(threadCount);
executeReaderThreads(object, threadCount, service);
assertEquals(object.isReadLockAvailable(), true);
service.shutdown();
}
private void executeWriterThreads(SynchronizedHashMapWithRWLock object, int threadCount, ExecutorService service) {
for (int i = 0; i < threadCount; i++) {
service.execute(() -> {
try {
object.put("key" + threadCount, "value" + threadCount);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
private void executeReaderThreads(SynchronizedHashMapWithRWLock object, int threadCount, ExecutorService service) {
for (int i = 0; i < threadCount; i++)
service.execute(() -> {
object.get("key" + threadCount);
});
}
}

41
core-java-concurrency/src/test/java/com/baeldung/concurrent/phaser/PhaserUnitTest.java Normal file
View File

@@ -0,0 +1,41 @@
package com.baeldung.concurrent.phaser;
import org.junit.FixMethodOrder;
import org.junit.Test;
import org.junit.runners.MethodSorters;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Phaser;
import static junit.framework.TestCase.assertEquals;
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
public class PhaserUnitTest {
@Test
public void givenPhaser_whenCoordinateWorksBetweenThreads_thenShouldCoordinateBetweenMultiplePhases() {
//given
ExecutorService executorService = Executors.newCachedThreadPool();
Phaser ph = new Phaser(1);
assertEquals(0, ph.getPhase());
//when
executorService.submit(new LongRunningAction("thread-1", ph));
executorService.submit(new LongRunningAction("thread-2", ph));
executorService.submit(new LongRunningAction("thread-3", ph));
//then
ph.arriveAndAwaitAdvance();
assertEquals(1, ph.getPhase());
//and
executorService.submit(new LongRunningAction("thread-4", ph));
executorService.submit(new LongRunningAction("thread-5", ph));
ph.arriveAndAwaitAdvance();
assertEquals(2, ph.getPhase());
ph.arriveAndDeregister();
}
}

57
core-java-concurrency/src/test/java/com/baeldung/concurrent/priorityblockingqueue/PriorityBlockingQueueIntegrationTest.java Normal file
View File

@@ -0,0 +1,57 @@
package com.baeldung.concurrent.priorityblockingqueue;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.util.Lists.newArrayList;
public class PriorityBlockingQueueIntegrationTest {
private static final Logger LOG = LoggerFactory.getLogger(PriorityBlockingQueueIntegrationTest.class);
@Test
public void givenUnorderedValues_whenPolling_thenShouldOrderQueue() throws InterruptedException {
PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>();
ArrayList<Integer> polledElements = new ArrayList<>();
queue.add(1);
queue.add(5);
queue.add(2);
queue.add(3);
queue.add(4);
queue.drainTo(polledElements);
assertThat(polledElements).containsExactly(1, 2, 3, 4, 5);
}
@Test
public void whenPollingEmptyQueue_thenShouldBlockThread() throws InterruptedException {
PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>();
final Thread thread = new Thread(() -> {
LOG.debug("Polling...");
while (true) {
try {
Integer poll = queue.take();
LOG.debug("Polled: " + poll);
} catch (InterruptedException ignored) {
}
}
});
thread.start();
Thread.sleep(TimeUnit.SECONDS.toMillis(5));
LOG.debug("Adding to queue");
queue.addAll(newArrayList(1, 5, 6, 1, 2, 6, 7));
Thread.sleep(TimeUnit.SECONDS.toMillis(1));
}
}

114
core-java-concurrency/src/test/java/com/baeldung/concurrent/semaphores/SemaphoresManualTest.java Normal file
View File

@@ -0,0 +1,114 @@
package com.baeldung.concurrent.semaphores;
import org.junit.Test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class SemaphoresManualTest {
// ========= login queue ======
@Test
public void givenLoginQueue_whenReachLimit_thenBlocked() {
final int slots = 10;
final ExecutorService executorService = Executors.newFixedThreadPool(slots);
final LoginQueueUsingSemaphore loginQueue = new LoginQueueUsingSemaphore(slots);
IntStream.range(0, slots)
.forEach(user -> executorService.execute(loginQueue::tryLogin));
executorService.shutdown();
assertEquals(0, loginQueue.availableSlots());
assertFalse(loginQueue.tryLogin());
}
@Test
public void givenLoginQueue_whenLogout_thenSlotsAvailable() {
final int slots = 10;
final ExecutorService executorService = Executors.newFixedThreadPool(slots);
final LoginQueueUsingSemaphore loginQueue = new LoginQueueUsingSemaphore(slots);
IntStream.range(0, slots)
.forEach(user -> executorService.execute(loginQueue::tryLogin));
executorService.shutdown();
assertEquals(0, loginQueue.availableSlots());
loginQueue.logout();
assertTrue(loginQueue.availableSlots() > 0);
assertTrue(loginQueue.tryLogin());
}
// ========= delay queue =======
@Test
public void givenDelayQueue_whenReachLimit_thenBlocked() {
final int slots = 50;
final ExecutorService executorService = Executors.newFixedThreadPool(slots);
final DelayQueueUsingTimedSemaphore delayQueue = new DelayQueueUsingTimedSemaphore(1, slots);
IntStream.range(0, slots)
.forEach(user -> executorService.execute(delayQueue::tryAdd));
executorService.shutdown();
assertEquals(0, delayQueue.availableSlots());
assertFalse(delayQueue.tryAdd());
}
@Test
public void givenDelayQueue_whenTimePass_thenSlotsAvailable() throws InterruptedException {
final int slots = 50;
final ExecutorService executorService = Executors.newFixedThreadPool(slots);
final DelayQueueUsingTimedSemaphore delayQueue = new DelayQueueUsingTimedSemaphore(1, slots);
IntStream.range(0, slots)
.forEach(user -> executorService.execute(delayQueue::tryAdd));
executorService.shutdown();
assertEquals(0, delayQueue.availableSlots());
Thread.sleep(1000);
assertTrue(delayQueue.availableSlots() > 0);
assertTrue(delayQueue.tryAdd());
}
// ========== mutex ========
@Test
public void whenMutexAndMultipleThreads_thenBlocked() throws InterruptedException {
final int count = 5;
final ExecutorService executorService = Executors.newFixedThreadPool(count);
final CounterUsingMutex counter = new CounterUsingMutex();
IntStream.range(0, count)
.forEach(user -> executorService.execute(() -> {
try {
counter.increase();
} catch (final InterruptedException e) {
e.printStackTrace();
}
}));
executorService.shutdown();
assertTrue(counter.hasQueuedThreads());
}
@Test
public void givenMutexAndMultipleThreads_ThenDelay_thenCorrectCount() throws InterruptedException {
final int count = 5;
final ExecutorService executorService = Executors.newFixedThreadPool(count);
final CounterUsingMutex counter = new CounterUsingMutex();
IntStream.range(0, count)
.forEach(user -> executorService.execute(() -> {
try {
counter.increase();
} catch (final InterruptedException e) {
e.printStackTrace();
}
}));
executorService.shutdown();
assertTrue(counter.hasQueuedThreads());
Thread.sleep(5000);
assertFalse(counter.hasQueuedThreads());
assertEquals(count, counter.getCount());
}
}

120
core-java-concurrency/src/test/java/com/baeldung/concurrent/skiplist/ConcurrentSkipListSetIntegrationTest.java Normal file
View File

@@ -0,0 +1,120 @@
package com.baeldung.concurrent.skiplist;
import org.junit.Test;
import java.time.ZonedDateTime;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class ConcurrentSkipListSetIntegrationTest {
@Test
public void givenThreadsProducingEvents_whenGetForEventsFromLastMinute_thenReturnThoseEventsInTheLockFreeWay() throws InterruptedException {
//given
ExecutorService executorService = Executors.newFixedThreadPool(3);
EventWindowSort eventWindowSort = new EventWindowSort();
int numberOfThreads = 2;
//when
Runnable producer = () -> IntStream
.rangeClosed(0, 100)
.forEach(index -> eventWindowSort.acceptEvent(new Event(ZonedDateTime
.now()
.minusSeconds(index), UUID
.randomUUID()
.toString())));
for (int i = 0; i < numberOfThreads; i++) {
executorService.execute(producer);
}
Thread.sleep(500);
ConcurrentNavigableMap<ZonedDateTime, String> eventsFromLastMinute = eventWindowSort.getEventsFromLastMinute();
long eventsOlderThanOneMinute = eventsFromLastMinute
.entrySet()
.stream()
.filter(e -> e
.getKey()
.isBefore(ZonedDateTime
.now()
.minusMinutes(1)))
.count();
assertEquals(eventsOlderThanOneMinute, 0);
long eventYoungerThanOneMinute = eventsFromLastMinute
.entrySet()
.stream()
.filter(e -> e
.getKey()
.isAfter(ZonedDateTime
.now()
.minusMinutes(1)))
.count();
//then
assertTrue(eventYoungerThanOneMinute > 0);
executorService.awaitTermination(1, TimeUnit.SECONDS);
executorService.shutdown();
}
@Test
public void givenThreadsProducingEvents_whenGetForEventsOlderThanOneMinute_thenReturnThoseEventsInTheLockFreeWay() throws InterruptedException {
//given
ExecutorService executorService = Executors.newFixedThreadPool(3);
EventWindowSort eventWindowSort = new EventWindowSort();
int numberOfThreads = 2;
//when
Runnable producer = () -> IntStream
.rangeClosed(0, 100)
.forEach(index -> eventWindowSort.acceptEvent(new Event(ZonedDateTime
.now()
.minusSeconds(index), UUID
.randomUUID()
.toString())));
for (int i = 0; i < numberOfThreads; i++) {
executorService.execute(producer);
}
Thread.sleep(500);
ConcurrentNavigableMap<ZonedDateTime, String> eventsFromLastMinute = eventWindowSort.getEventsOlderThatOneMinute();
long eventsOlderThanOneMinute = eventsFromLastMinute
.entrySet()
.stream()
.filter(e -> e
.getKey()
.isBefore(ZonedDateTime
.now()
.minusMinutes(1)))
.count();
assertTrue(eventsOlderThanOneMinute > 0);
long eventYoungerThanOneMinute = eventsFromLastMinute
.entrySet()
.stream()
.filter(e -> e
.getKey()
.isAfter(ZonedDateTime
.now()
.minusMinutes(1)))
.count();
//then
assertEquals(eventYoungerThanOneMinute, 0);
executorService.awaitTermination(1, TimeUnit.SECONDS);
executorService.shutdown();
}
}

37
core-java-concurrency/src/test/java/com/baeldung/concurrent/synchronize/BaeldungSychronizedBlockTest.java Normal file
View File

@@ -0,0 +1,37 @@
package com.baeldung.concurrent.synchronize;
import org.junit.Test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
import static org.junit.Assert.assertEquals;
public class BaeldungSychronizedBlockTest {
@Test
public void givenMultiThread_whenBlockSync() throws InterruptedException {
ExecutorService service = Executors.newFixedThreadPool(3);
BaeldungSynchronizedBlocks synchronizedBlocks = new BaeldungSynchronizedBlocks();
IntStream.range(0, 1000)
.forEach(count -> service.submit(synchronizedBlocks::performSynchronisedTask));
service.awaitTermination(100, TimeUnit.MILLISECONDS);
assertEquals(1000, synchronizedBlocks.getCount());
}
@Test
public void givenMultiThread_whenStaticSyncBlock() throws InterruptedException {
ExecutorService service = Executors.newCachedThreadPool();
IntStream.range(0, 1000)
.forEach(count -> service.submit(BaeldungSynchronizedBlocks::performStaticSyncTask));
service.awaitTermination(100, TimeUnit.MILLISECONDS);
assertEquals(1000, BaeldungSynchronizedBlocks.getStaticCount());
}
}

51
core-java-concurrency/src/test/java/com/baeldung/concurrent/synchronize/BaeldungSynchronizeMethodsTest.java Normal file
View File

@@ -0,0 +1,51 @@
package com.baeldung.concurrent.synchronize;
import org.junit.Ignore;
import org.junit.Test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
import static org.junit.Assert.assertEquals;
public class BaeldungSynchronizeMethodsTest {
@Test
@Ignore
public void givenMultiThread_whenNonSyncMethod() throws InterruptedException {
ExecutorService service = Executors.newFixedThreadPool(3);
BaeldungSynchronizedMethods method = new BaeldungSynchronizedMethods();
IntStream.range(0, 1000)
.forEach(count -> service.submit(method::calculate));
service.awaitTermination(100, TimeUnit.MILLISECONDS);
assertEquals(1000, method.getSum());
}
@Test
public void givenMultiThread_whenMethodSync() throws InterruptedException {
ExecutorService service = Executors.newFixedThreadPool(3);
BaeldungSynchronizedMethods method = new BaeldungSynchronizedMethods();
IntStream.range(0, 1000)
.forEach(count -> service.submit(method::synchronisedCalculate));
service.awaitTermination(100, TimeUnit.MILLISECONDS);
assertEquals(1000, method.getSyncSum());
}
@Test
public void givenMultiThread_whenStaticSyncMethod() throws InterruptedException {
ExecutorService service = Executors.newCachedThreadPool();
IntStream.range(0, 1000)
.forEach(count -> service.submit(BaeldungSynchronizedMethods::syncStaticCalculate));
service.awaitTermination(100, TimeUnit.MILLISECONDS);
assertEquals(1000, BaeldungSynchronizedMethods.staticSum);
}
}

71
core-java-concurrency/src/test/java/com/baeldung/concurrent/volatilekeyword/SharedObjectManualTest.java Normal file
View File

@@ -0,0 +1,71 @@
package com.baeldung.concurrent.volatilekeyword;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import static junit.framework.Assert.assertEquals;
public class SharedObjectManualTest {
private SharedObject sharedObject;
private int valueReadByThread2;
private int valueReadByThread3;
@Before
public void setUp() {
sharedObject = new SharedObject();
}
@Test
public void whenOneThreadWrites_thenVolatileReadsFromMainMemory() throws InterruptedException {
Thread writer = new Thread(() -> sharedObject.increamentCount());
writer.start();
Thread readerOne = new Thread(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
valueReadByThread2 = sharedObject.getCount();
});
readerOne.start();
Thread readerTwo = new Thread(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
valueReadByThread3 = sharedObject.getCount();
});
readerTwo.start();
assertEquals(1, valueReadByThread2);
assertEquals(1, valueReadByThread3);
}
@Test
public void whenTwoThreadWrites_thenVolatileReadsFromMainMemory() throws InterruptedException {
Thread writerOne = new Thread(() -> sharedObject.increamentCount());
writerOne.start();
Thread.sleep(100);
Thread writerTwo = new Thread(() -> sharedObject.increamentCount());
writerTwo.start();
Thread.sleep(100);
Thread readerOne = new Thread(() -> valueReadByThread2 = sharedObject.getCount());
readerOne.start();
Thread readerTwo = new Thread(() -> valueReadByThread3 = sharedObject.getCount());
readerTwo.start();
assertEquals(2, valueReadByThread2);
assertEquals(2, valueReadByThread3);
}
}

78
core-java-concurrency/src/test/java/com/baeldung/java/concurrentmap/ConcurrentMapAggregateStatusManualTest.java Normal file
View File

@@ -0,0 +1,78 @@
package com.baeldung.java.concurrentmap;
import org.junit.Before;
import org.junit.Test;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
public class ConcurrentMapAggregateStatusManualTest {
private ExecutorService executorService;
private Map<String, Integer> concurrentMap;
private List<Integer> mapSizes;
private int MAX_SIZE = 100000;
@Before
public void init() {
executorService = Executors.newFixedThreadPool(2);
concurrentMap = new ConcurrentHashMap<>();
mapSizes = new ArrayList<>(MAX_SIZE);
}
@Test
public void givenConcurrentMap_whenSizeWithoutConcurrentUpdates_thenCorrect() throws InterruptedException {
Runnable collectMapSizes = () -> {
for (int i = 0; i < MAX_SIZE; i++) {
concurrentMap.put(String.valueOf(i), i);
mapSizes.add(concurrentMap.size());
}
};
Runnable retrieveMapData = () -> {
for (int i = 0; i < MAX_SIZE; i++) {
concurrentMap.get(String.valueOf(i));
}
};
executorService.execute(retrieveMapData);
executorService.execute(collectMapSizes);
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.MINUTES);
for (int i = 1; i <= MAX_SIZE; i++) {
assertEquals("map size should be consistently reliable", i, mapSizes.get(i - 1)
.intValue());
}
assertEquals(MAX_SIZE, concurrentMap.size());
}
@Test
public void givenConcurrentMap_whenUpdatingAndGetSize_thenError() throws InterruptedException {
Runnable collectMapSizes = () -> {
for (int i = 0; i < MAX_SIZE; i++) {
mapSizes.add(concurrentMap.size());
}
};
Runnable updateMapData = () -> {
for (int i = 0; i < MAX_SIZE; i++) {
concurrentMap.put(String.valueOf(i), i);
}
};
executorService.execute(updateMapData);
executorService.execute(collectMapSizes);
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.MINUTES);
assertNotEquals("map size collected with concurrent updates not reliable", MAX_SIZE, mapSizes.get(MAX_SIZE - 1)
.intValue());
assertEquals(MAX_SIZE, concurrentMap.size());
}
}

160
core-java-concurrency/src/test/java/com/baeldung/java/concurrentmap/ConcurrentMapNullKeyValueManualTest.java Normal file
View File

@@ -0,0 +1,160 @@
package com.baeldung.java.concurrentmap;
import org.junit.Before;
import org.junit.Test;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.junit.Assert.assertNull;
public class ConcurrentMapNullKeyValueManualTest {
ConcurrentMap<String, Object> concurrentMap;
@Before
public void setup() {
concurrentMap = new ConcurrentHashMap<>();
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenGetWithNullKey_thenThrowsNPE() {
concurrentMap.get(null);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenGetOrDefaultWithNullKey_thenThrowsNPE() {
concurrentMap.getOrDefault(null, new Object());
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenPutWithNullKey_thenThrowsNPE() {
concurrentMap.put(null, new Object());
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenPutNullValue_thenThrowsNPE() {
concurrentMap.put("test", null);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMapAndKeyAbsent_whenPutWithNullKey_thenThrowsNPE() {
concurrentMap.putIfAbsent(null, new Object());
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMapAndMapWithNullKey_whenPutNullKeyMap_thenThrowsNPE() {
Map<String, Object> nullKeyMap = new HashMap<>();
nullKeyMap.put(null, new Object());
concurrentMap.putAll(nullKeyMap);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMapAndMapWithNullValue_whenPutNullValueMap_thenThrowsNPE() {
Map<String, Object> nullValueMap = new HashMap<>();
nullValueMap.put("test", null);
concurrentMap.putAll(nullValueMap);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenReplaceNullKeyWithValues_thenThrowsNPE() {
concurrentMap.replace(null, new Object(), new Object());
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenReplaceWithNullNewValue_thenThrowsNPE() {
Object o = new Object();
concurrentMap.replace("test", o, null);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenReplaceOldNullValue_thenThrowsNPE() {
Object o = new Object();
concurrentMap.replace("test", null, o);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenReplaceWithNullValue_thenThrowsNPE() {
concurrentMap.replace("test", null);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenReplaceNullKey_thenThrowsNPE() {
concurrentMap.replace(null, "test");
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenReplaceAllMappingNull_thenThrowsNPE() {
concurrentMap.put("test", new Object());
concurrentMap.replaceAll((s, o) -> null);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenRemoveNullKey_thenThrowsNPE() {
concurrentMap.remove(null);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenRemoveNullKeyWithValue_thenThrowsNPE() {
concurrentMap.remove(null, new Object());
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenMergeNullKeyWithValue_thenThrowsNPE() {
concurrentMap.merge(null, new Object(), (o, o2) -> o2);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenMergeKeyWithNullValue_thenThrowsNPE() {
concurrentMap.put("test", new Object());
concurrentMap.merge("test", null, (o, o2) -> o2);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMapAndAssumeKeyAbsent_whenComputeWithNullKey_thenThrowsNPE() {
concurrentMap.computeIfAbsent(null, s -> s);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMapAndAssumeKeyPresent_whenComputeWithNullKey_thenThrowsNPE() {
concurrentMap.computeIfPresent(null, (s, o) -> o);
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenComputeWithNullKey_thenThrowsNPE() {
concurrentMap.compute(null, (s, o) -> o);
}
@Test
public void givenConcurrentHashMap_whenMergeKeyRemappingNull_thenRemovesMapping() {
Object oldValue = new Object();
concurrentMap.put("test", oldValue);
concurrentMap.merge("test", new Object(), (o, o2) -> null);
assertNull(concurrentMap.get("test"));
}
@Test
public void givenConcurrentHashMapAndKeyAbsent_whenComputeWithKeyRemappingNull_thenRemainsAbsent() {
concurrentMap.computeIfPresent("test", (s, o) -> null);
assertNull(concurrentMap.get("test"));
}
@Test
public void givenKeyPresent_whenComputeIfPresentRemappingNull_thenMappingRemoved() {
Object oldValue = new Object();
concurrentMap.put("test", oldValue);
concurrentMap.computeIfPresent("test", (s, o) -> null);
assertNull(concurrentMap.get("test"));
}
@Test
public void givenKeyPresent_whenComputeRemappingNull_thenMappingRemoved() {
Object oldValue = new Object();
concurrentMap.put("test", oldValue);
concurrentMap.compute("test", (s, o) -> null);
assertNull(concurrentMap.get("test"));
}
}

99
core-java-concurrency/src/test/java/com/baeldung/java/concurrentmap/ConcurrentMapPerformanceManualTest.java Normal file
View File

@@ -0,0 +1,99 @@
package com.baeldung.java.concurrentmap;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.Collections;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
public class ConcurrentMapPerformanceManualTest {
@Test
public void givenMaps_whenGetPut500KTimes_thenConcurrentMapFaster() throws Exception {
final Map<String, Object> hashtable = new Hashtable<>();
final Map<String, Object> synchronizedHashMap = Collections.synchronizedMap(new HashMap<>());
final Map<String, Object> concurrentHashMap = new ConcurrentHashMap<>();
final long hashtableAvgRuntime = timeElapseForGetPut(hashtable);
final long syncHashMapAvgRuntime = timeElapseForGetPut(synchronizedHashMap);
final long concurrentHashMapAvgRuntime = timeElapseForGetPut(concurrentHashMap);
System.out.println(String.format("Hashtable: %s, syncHashMap: %s, ConcurrentHashMap: %s", hashtableAvgRuntime, syncHashMapAvgRuntime, concurrentHashMapAvgRuntime));
assertTrue(hashtableAvgRuntime > concurrentHashMapAvgRuntime);
assertTrue(syncHashMapAvgRuntime > concurrentHashMapAvgRuntime);
}
private long timeElapseForGetPut(Map<String, Object> map) throws InterruptedException {
final ExecutorService executorService = Executors.newFixedThreadPool(4);
final long startTime = System.nanoTime();
for (int i = 0; i < 4; i++) {
executorService.execute(() -> {
for (int j = 0; j < 500_000; j++) {
final int value = ThreadLocalRandom.current().nextInt(10000);
final String key = String.valueOf(value);
map.put(key, value);
map.get(key);
}
});
}
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.MINUTES);
return (System.nanoTime() - startTime) / 500_000;
}
@Test
public void givenConcurrentMap_whenKeyWithSameHashCode_thenPerformanceDegrades() throws InterruptedException {
class SameHash {
@Override
public int hashCode() {
return 1;
}
}
final int executeTimes = 5000;
final Map<SameHash, Integer> mapOfSameHash = new ConcurrentHashMap<>();
ExecutorService executorService = Executors.newFixedThreadPool(2);
final long sameHashStartTime = System.currentTimeMillis();
for (int i = 0; i < 2; i++) {
executorService.execute(() -> {
for (int j = 0; j < executeTimes; j++) {
mapOfSameHash.put(new SameHash(), 1);
}
});
}
executorService.shutdown();
executorService.awaitTermination(5, TimeUnit.SECONDS);
final long mapOfSameHashDuration = System.currentTimeMillis() - sameHashStartTime;
final Map<Object, Integer> mapOfDefaultHash = new ConcurrentHashMap<>();
executorService = Executors.newFixedThreadPool(2);
final long defaultHashStartTime = System.currentTimeMillis();
for (int i = 0; i < 2; i++) {
executorService.execute(() -> {
for (int j = 0; j < executeTimes; j++) {
mapOfDefaultHash.put(new Object(), 1);
}
});
}
executorService.shutdown();
executorService.awaitTermination(5, TimeUnit.SECONDS);
final long mapOfDefaultHashDuration = System.currentTimeMillis() - defaultHashStartTime;
assertEquals(executeTimes * 2, mapOfDefaultHash.size());
assertEquals(executeTimes * 2, mapOfSameHash.size());
System.out.println(String.format("same-hash: %s, default-hash: %s", mapOfSameHashDuration, mapOfDefaultHashDuration));
assertTrue("same hashCode() should greatly degrade performance", mapOfSameHashDuration > (mapOfDefaultHashDuration * 10));
}
}

79
core-java-concurrency/src/test/java/com/baeldung/java/concurrentmap/ConcurrentNavigableMapManualTest.java Normal file
View File

@@ -0,0 +1,79 @@
package com.baeldung.java.concurrentmap;
import org.junit.Test;
import java.util.Iterator;
import java.util.NavigableMap;
import java.util.TreeMap;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
public class ConcurrentNavigableMapManualTest {
@Test
public void givenSkipListMap_whenAccessInMultiThreads_thenOrderingStable() throws InterruptedException {
NavigableMap<Integer, String> skipListMap = new ConcurrentSkipListMap<>();
updateMapConcurrently(skipListMap, 4);
Iterator<Integer> skipListIter = skipListMap.keySet().iterator();
int previous = skipListIter.next();
while (skipListIter.hasNext()) {
int current = skipListIter.next();
assertTrue(previous < current);
}
}
private void updateMapConcurrently(NavigableMap<Integer, String> navigableMap, int concurrencyLevel) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(concurrencyLevel);
for (int i = 0; i < concurrencyLevel; i++) {
executorService.execute(() -> {
ThreadLocalRandom random = ThreadLocalRandom.current();
for (int j = 0; j < 10000; j++) {
navigableMap.put(random.nextInt(), "test");
}
});
}
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.MINUTES);
}
@Test
public void givenSkipListMap_whenNavConcurrently_thenCountCorrect() throws InterruptedException {
NavigableMap<Integer, Integer> skipListMap = new ConcurrentSkipListMap<>();
int count = countMapElementByPollingFirstEntry(skipListMap, 10000, 4);
assertEquals(10000 * 4, count);
}
@Test
public void givenTreeMap_whenNavConcurrently_thenCountError() throws InterruptedException {
NavigableMap<Integer, Integer> treeMap = new TreeMap<>();
int count = countMapElementByPollingFirstEntry(treeMap, 10000, 4);
assertNotEquals(10000 * 4, count);
}
private int countMapElementByPollingFirstEntry(NavigableMap<Integer, Integer> navigableMap, int elementCount, int concurrencyLevel) throws InterruptedException {
for (int i = 0; i < elementCount * concurrencyLevel; i++) {
navigableMap.put(i, i);
}
AtomicInteger counter = new AtomicInteger(0);
ExecutorService executorService = Executors.newFixedThreadPool(concurrencyLevel);
for (int j = 0; j < concurrencyLevel; j++) {
executorService.execute(() -> {
for (int i = 0; i < elementCount; i++) {
if (navigableMap.pollFirstEntry() != null) {
counter.incrementAndGet();
}
}
});
}
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.MINUTES);
return counter.get();
}
}

72
core-java-concurrency/src/test/java/com/baeldung/java/concurrentmap/ConcurretMapMemoryConsistencyManualTest.java Normal file
View File

@@ -0,0 +1,72 @@
package com.baeldung.java.concurrentmap;
import org.junit.Test;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.*;
public class ConcurretMapMemoryConsistencyManualTest {
@Test
public void givenConcurrentMap_whenSumParallel_thenCorrect() throws Exception {
Map<String, Integer> map = new ConcurrentHashMap<>();
List<Integer> sumList = parallelSum100(map, 1000);
assertEquals(1, sumList.stream()
.distinct()
.count());
long wrongResultCount = sumList.stream()
.filter(num -> num != 100)
.count();
assertEquals(0, wrongResultCount);
}
@Test
public void givenHashtable_whenSumParallel_thenCorrect() throws Exception {
Map<String, Integer> map = new Hashtable<>();
List<Integer> sumList = parallelSum100(map, 1000);
assertEquals(1, sumList.stream()
.distinct()
.count());
long wrongResultCount = sumList.stream()
.filter(num -> num != 100)
.count();
assertEquals(0, wrongResultCount);
}
@Test
public void givenHashMap_whenSumParallel_thenError() throws Exception {
Map<String, Integer> map = new HashMap<>();
List<Integer> sumList = parallelSum100(map, 100);
assertNotEquals(1, sumList.stream()
.distinct()
.count());
long wrongResultCount = sumList.stream()
.filter(num -> num != 100)
.count();
assertTrue(wrongResultCount > 0);
}
private List<Integer> parallelSum100(Map<String, Integer> map, int executionTimes) throws InterruptedException {
List<Integer> sumList = new ArrayList<>(1000);
for (int i = 0; i < executionTimes; i++) {
map.put("test", 0);
ExecutorService executorService = Executors.newFixedThreadPool(4);
for (int j = 0; j < 10; j++) {
executorService.execute(() -> {
for (int k = 0; k < 10; k++)
map.computeIfPresent("test", (key, value) -> value + 1);
});
}
executorService.shutdown();
executorService.awaitTermination(5, TimeUnit.SECONDS);
sumList.add(map.get("test"));
}
return sumList;
}
}

80
core-java-concurrency/src/test/java/com/baeldung/java/concurrentmodification/ConcurrentModificationUnitTest.java Normal file
View File

@@ -0,0 +1,80 @@
package com.baeldung.java.concurrentmodification;
import org.junit.Test;
import java.util.ArrayList;
import java.util.Collection;
import java.util.ConcurrentModificationException;
import java.util.Iterator;
import java.util.List;
import static java.util.stream.Collectors.toList;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.util.Lists.newArrayList;
public class ConcurrentModificationUnitTest {
@Test(expected = ConcurrentModificationException.class)
public void givenIterating_whenRemoving_thenThrowException() throws InterruptedException {
List<Integer> integers = newArrayList(1, 2, 3);
for (Integer integer : integers) {
integers.remove(1);
}
}
@Test
public void givenIterating_whenUsingIteratorRemove_thenNoError() throws InterruptedException {
List<Integer> integers = newArrayList(1, 2, 3);
for (Iterator<Integer> iterator = integers.iterator(); iterator.hasNext();) {
Integer integer = iterator.next();
if(integer == 2) {
iterator.remove();
}
}
assertThat(integers).containsExactly(1, 3);
}
@Test
public void givenIterating_whenUsingRemovalList_thenNoError() throws InterruptedException {
List<Integer> integers = newArrayList(1, 2, 3);
List<Integer> toRemove = newArrayList();
for (Integer integer : integers) {
if(integer == 2) {
toRemove.add(integer);
}
}
integers.removeAll(toRemove);
assertThat(integers).containsExactly(1, 3);
}
@Test
public void whenUsingRemoveIf_thenRemoveElements() throws InterruptedException {
Collection<Integer> integers = newArrayList(1, 2, 3);
integers.removeIf(i -> i == 2);
assertThat(integers).containsExactly(1, 3);
}
@Test
public void whenUsingStream_thenRemoveElements() {
Collection<Integer> integers = newArrayList(1, 2, 3);
List<String> collected = integers
.stream()
.filter(i -> i != 2)
.map(Object::toString)
.collect(toList());
assertThat(collected).containsExactly("1", "3");
}
}

162
core-java-concurrency/src/test/java/com/baeldung/java8/Java8ExecutorServiceIntegrationTest.java Normal file
View File

@@ -0,0 +1,162 @@
package com.baeldung.java8;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.junit.Before;
import org.junit.Test;
public class Java8ExecutorServiceIntegrationTest {
private Runnable runnableTask;
private Callable<String> callableTask;
private List<Callable<String>> callableTasks;
@Before
public void init() {
runnableTask = () -> {
try {
TimeUnit.MILLISECONDS.sleep(300);
} catch (InterruptedException e) {
e.printStackTrace();
}
};
callableTask = () -> {
TimeUnit.MILLISECONDS.sleep(300);
return "Task's execution";
};
callableTasks = new ArrayList<>();
callableTasks.add(callableTask);
callableTasks.add(callableTask);
callableTasks.add(callableTask);
}
@Test
public void creationSubmittingTaskShuttingDown_whenShutDown_thenCorrect() {
ExecutorService executorService = Executors.newFixedThreadPool(10);
executorService.submit(runnableTask);
executorService.submit(callableTask);
executorService.shutdown();
assertTrue(executorService.isShutdown());
}
@Test
public void creationSubmittingTasksShuttingDownNow_whenShutDownAfterAwating_thenCorrect() {
ExecutorService threadPoolExecutor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
for (int i = 0; i < 100; i++) {
threadPoolExecutor.submit(callableTask);
}
List<Runnable> notExecutedTasks = smartShutdown(threadPoolExecutor);
assertTrue(threadPoolExecutor.isShutdown());
assertFalse(notExecutedTasks.isEmpty());
assertTrue(notExecutedTasks.size() > 0 && notExecutedTasks.size() < 98);
}
private List<Runnable> smartShutdown(ExecutorService executorService) {
List<Runnable> notExecutedTasks = new ArrayList<>();
executorService.shutdown();
try {
if (!executorService.awaitTermination(800, TimeUnit.MILLISECONDS)) {
notExecutedTasks = executorService.shutdownNow();
}
} catch (InterruptedException e) {
notExecutedTasks = executorService.shutdownNow();
}
return notExecutedTasks;
}
@Test
public void submittingTasks_whenExecutedOneAndAll_thenCorrect() {
ExecutorService executorService = Executors.newFixedThreadPool(10);
String result = null;
List<Future<String>> futures = new ArrayList<>();
try {
result = executorService.invokeAny(callableTasks);
futures = executorService.invokeAll(callableTasks);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
assertEquals("Task's execution", result);
assertTrue(futures.size() == 3);
}
@Test
public void submittingTaskShuttingDown_whenGetExpectedResult_thenCorrect() {
ExecutorService executorService = Executors.newFixedThreadPool(10);
Future<String> future = executorService.submit(callableTask);
String result = null;
try {
result = future.get();
result = future.get(200, TimeUnit.MILLISECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
e.printStackTrace();
}
executorService.shutdown();
assertEquals("Task's execution", result);
}
@Test
public void submittingTask_whenCanceled_thenCorrect() {
ExecutorService executorService = Executors.newFixedThreadPool(10);
Future<String> future = executorService.submit(callableTask);
boolean canceled = future.cancel(true);
boolean isCancelled = future.isCancelled();
executorService.shutdown();
assertTrue(canceled);
assertTrue(isCancelled);
}
@Test
public void submittingTaskScheduling_whenExecuted_thenCorrect() {
ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
Future<String> resultFuture = executorService.schedule(callableTask, 1, TimeUnit.SECONDS);
String result = null;
try {
result = resultFuture.get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
executorService.shutdown();
assertEquals("Task's execution", result);
}
}

88
core-java-concurrency/src/test/java/com/baeldung/synchronousqueue/SynchronousQueueIntegrationTest.java Normal file
View File

@@ -0,0 +1,88 @@
package com.baeldung.synchronousqueue;
import org.junit.FixMethodOrder;
import org.junit.Test;
import org.junit.runners.MethodSorters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import static junit.framework.TestCase.assertEquals;
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
public class SynchronousQueueIntegrationTest {
private static final Logger LOG = LoggerFactory.getLogger(SynchronousQueueIntegrationTest.class);
@Test
public void givenTwoThreads_whenWantToExchangeUsingLockGuardedVariable_thenItSucceed() throws InterruptedException {
//given
ExecutorService executor = Executors.newFixedThreadPool(2);
AtomicInteger sharedState = new AtomicInteger();
CountDownLatch countDownLatch = new CountDownLatch(1);
Runnable producer = () -> {
Integer producedElement = ThreadLocalRandom.current().nextInt();
LOG.debug("Saving an element: " + producedElement + " to the exchange point");
sharedState.set(producedElement);
countDownLatch.countDown();
};
Runnable consumer = () -> {
try {
countDownLatch.await();
Integer consumedElement = sharedState.get();
LOG.debug("consumed an element: " + consumedElement + " from the exchange point");
} catch (InterruptedException ex) {
ex.printStackTrace();
}
};
//when
executor.execute(producer);
executor.execute(consumer);
//then
executor.awaitTermination(500, TimeUnit.MILLISECONDS);
executor.shutdown();
assertEquals(countDownLatch.getCount(), 0);
}
@Test
public void givenTwoThreads_whenWantToExchangeUsingSynchronousQueue_thenItSucceed() throws InterruptedException {
//given
ExecutorService executor = Executors.newFixedThreadPool(2);
final SynchronousQueue<Integer> queue = new SynchronousQueue<>();
Runnable producer = () -> {
Integer producedElement = ThreadLocalRandom.current().nextInt();
try {
LOG.debug("Saving an element: " + producedElement + " to the exchange point");
queue.put(producedElement);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
};
Runnable consumer = () -> {
try {
Integer consumedElement = queue.take();
LOG.debug("consumed an element: " + consumedElement + " from the exchange point");
} catch (InterruptedException ex) {
ex.printStackTrace();
}
};
//when
executor.execute(producer);
executor.execute(consumer);
//then
executor.awaitTermination(500, TimeUnit.MILLISECONDS);
executor.shutdown();
assertEquals(queue.size(), 0);
}
}

35
core-java-concurrency/src/test/java/com/baeldung/threadlocal/ThreadLocalIntegrationTest.java Normal file
View File

@@ -0,0 +1,35 @@
package com.baeldung.threadlocal;
import org.junit.Test;
import java.util.concurrent.ExecutionException;
import static org.junit.Assert.assertEquals;
public class ThreadLocalIntegrationTest {
@Test
public void givenThreadThatStoresContextInAMap_whenStartThread_thenShouldSetContextForBothUsers() throws ExecutionException, InterruptedException {
//when
SharedMapWithUserContext firstUser = new SharedMapWithUserContext(1);
SharedMapWithUserContext secondUser = new SharedMapWithUserContext(2);
new Thread(firstUser).start();
new Thread(secondUser).start();
Thread.sleep(3000);
//then
assertEquals(SharedMapWithUserContext.userContextPerUserId.size(), 2);
}
@Test
public void givenThreadThatStoresContextInThreadLocal_whenStartThread_thenShouldStoreContextInThreadLocal() throws ExecutionException, InterruptedException {
//when
ThreadLocalWithUserContext firstUser = new ThreadLocalWithUserContext(1);
ThreadLocalWithUserContext secondUser = new ThreadLocalWithUserContext(2);
new Thread(firstUser).start();
new Thread(secondUser).start();
Thread.sleep(3000);
}
}

147
core-java-concurrency/src/test/java/com/baeldung/threadpool/CoreThreadPoolIntegrationTest.java Normal file
View File

@@ -0,0 +1,147 @@
package com.baeldung.threadpool;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.assertEquals;
public class CoreThreadPoolIntegrationTest {
private static final Logger LOG = LoggerFactory.getLogger(CoreThreadPoolIntegrationTest.class);
@Test(timeout = 1000)
public void whenCallingExecuteWithRunnable_thenRunnableIsExecuted() throws InterruptedException {
CountDownLatch lock = new CountDownLatch(1);
Executor executor = Executors.newSingleThreadExecutor();
executor.execute(() -> {
LOG.debug("Hello World");
lock.countDown();
});
lock.await(1000, TimeUnit.MILLISECONDS);
}
@Test
public void whenUsingExecutorServiceAndFuture_thenCanWaitOnFutureResult() throws InterruptedException, ExecutionException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
Future<String> future = executorService.submit(() -> "Hello World");
String result = future.get();
assertEquals("Hello World", result);
}
@Test
public void whenUsingFixedThreadPool_thenCoreAndMaximumThreadSizeAreTheSame() {
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);
executor.submit(() -> {
Thread.sleep(1000);
return null;
});
executor.submit(() -> {
Thread.sleep(1000);
return null;
});
executor.submit(() -> {
Thread.sleep(1000);
return null;
});
assertEquals(2, executor.getPoolSize());
assertEquals(1, executor.getQueue().size());
}
@Test
public void whenUsingCachedThreadPool_thenPoolSizeGrowsUnbounded() {
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newCachedThreadPool();
executor.submit(() -> {
Thread.sleep(1000);
return null;
});
executor.submit(() -> {
Thread.sleep(1000);
return null;
});
executor.submit(() -> {
Thread.sleep(1000);
return null;
});
assertEquals(3, executor.getPoolSize());
assertEquals(0, executor.getQueue().size());
}
@Test(timeout = 1000)
public void whenUsingSingleThreadPool_thenTasksExecuteSequentially() throws InterruptedException {
CountDownLatch lock = new CountDownLatch(2);
AtomicInteger counter = new AtomicInteger();
ExecutorService executor = Executors.newSingleThreadExecutor();
executor.submit(() -> {
counter.set(1);
lock.countDown();
});
executor.submit(() -> {
counter.compareAndSet(1, 2);
lock.countDown();
});
lock.await(1000, TimeUnit.MILLISECONDS);
assertEquals(2, counter.get());
}
@Test(timeout = 1000)
public void whenSchedulingTask_thenTaskExecutesWithinGivenPeriod() throws InterruptedException {
CountDownLatch lock = new CountDownLatch(1);
ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
executor.schedule(() -> {
LOG.debug("Hello World");
lock.countDown();
}, 500, TimeUnit.MILLISECONDS);
lock.await(1000, TimeUnit.MILLISECONDS);
}
@Test(timeout = 1000)
public void whenSchedulingTaskWithFixedPeriod_thenTaskExecutesMultipleTimes() throws InterruptedException {
CountDownLatch lock = new CountDownLatch(3);
ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
ScheduledFuture<?> future = executor.scheduleAtFixedRate(() -> {
LOG.debug("Hello World");
lock.countDown();
}, 500, 100, TimeUnit.MILLISECONDS);
lock.await();
future.cancel(true);
}
@Test
public void whenUsingForkJoinPool_thenSumOfTreeElementsIsCalculatedCorrectly() {
TreeNode tree = new TreeNode(5, new TreeNode(3), new TreeNode(2, new TreeNode(2), new TreeNode(8)));
ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
int sum = forkJoinPool.invoke(new CountingTask(tree));
assertEquals(20, sum);
}
}

54
core-java-concurrency/src/test/java/com/baeldung/threadpool/GuavaThreadPoolIntegrationTest.java Normal file
View File

@@ -0,0 +1,54 @@
package com.baeldung.threadpool;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class GuavaThreadPoolIntegrationTest {
@Test
public void whenExecutingTaskWithDirectExecutor_thenTheTaskIsExecutedInTheCurrentThread() {
Executor executor = MoreExecutors.directExecutor();
AtomicBoolean executed = new AtomicBoolean();
executor.execute(() -> {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
executed.set(true);
});
assertTrue(executed.get());
}
@Test
public void whenJoiningFuturesWithAllAsList_thenCombinedFutureCompletesAfterAllFuturesComplete() throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newCachedThreadPool();
ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(executorService);
ListenableFuture<String> future1 = listeningExecutorService.submit(() -> "Hello");
ListenableFuture<String> future2 = listeningExecutorService.submit(() -> "World");
String greeting = Futures.allAsList(future1, future2).get().stream().collect(Collectors.joining(" "));
assertEquals("Hello World", greeting);
}
}

74
core-java-concurrency/src/test/java/com/baeldung/transferqueue/TransferQueueIntegrationTest.java Normal file
View File

@@ -0,0 +1,74 @@
package com.baeldung.transferqueue;
import org.junit.FixMethodOrder;
import org.junit.Test;
import org.junit.runners.MethodSorters;
import java.util.concurrent.*;
import static junit.framework.TestCase.assertEquals;
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
public class TransferQueueIntegrationTest {
@Test
public void whenMultipleConsumersAndProducers_thenProcessAllMessages() throws InterruptedException {
//given
TransferQueue<String> transferQueue = new LinkedTransferQueue<>();
ExecutorService exService = Executors.newFixedThreadPool(3);
Producer producer1 = new Producer(transferQueue, "1", 3);
Producer producer2 = new Producer(transferQueue, "2", 3);
Consumer consumer1 = new Consumer(transferQueue, "1", 3);
Consumer consumer2 = new Consumer(transferQueue, "2", 3);
//when
exService.execute(producer1);
exService.execute(producer2);
exService.execute(consumer1);
exService.execute(consumer2);
//then
exService.awaitTermination(5000, TimeUnit.MILLISECONDS);
exService.shutdown();
assertEquals(producer1.numberOfProducedMessages.intValue(), 3);
assertEquals(producer2.numberOfProducedMessages.intValue(), 3);
}
@Test
public void whenUseOneConsumerAndOneProducer_thenShouldProcessAllMessages() throws InterruptedException {
//given
TransferQueue<String> transferQueue = new LinkedTransferQueue<>();
ExecutorService exService = Executors.newFixedThreadPool(2);
Producer producer = new Producer(transferQueue, "1", 3);
Consumer consumer = new Consumer(transferQueue, "1", 3);
//when
exService.execute(producer);
exService.execute(consumer);
//then
exService.awaitTermination(5000, TimeUnit.MILLISECONDS);
exService.shutdown();
assertEquals(producer.numberOfProducedMessages.intValue(), 3);
assertEquals(consumer.numberOfConsumedMessages.intValue(), 3);
}
@Test
public void whenUseOneProducerAndNoConsumers_thenShouldFailWithTimeout() throws InterruptedException {
//given
TransferQueue<String> transferQueue = new LinkedTransferQueue<>();
ExecutorService exService = Executors.newFixedThreadPool(2);
Producer producer = new Producer(transferQueue, "1", 3);
//when
exService.execute(producer);
//then
exService.awaitTermination(5000, TimeUnit.MILLISECONDS);
exService.shutdown();
assertEquals(producer.numberOfProducedMessages.intValue(), 0);
}
}

38
core-java-concurrency/src/test/java/org/baeldung/java/streams/ThreadPoolInParallelStreamIntegrationTest.java Normal file
View File

@@ -0,0 +1,38 @@
package org.baeldung.java.streams;
import org.junit.Test;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
import java.util.stream.Stream;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class ThreadPoolInParallelStreamIntegrationTest {
@Test
public void giveRangeOfLongs_whenSummedInParallel_shouldBeEqualToExpectedTotal() throws InterruptedException, ExecutionException {
long firstNum = 1;
long lastNum = 1_000_000;
List<Long> aList = LongStream.rangeClosed(firstNum, lastNum).boxed().collect(Collectors.toList());
ForkJoinPool customThreadPool = new ForkJoinPool(4);
long actualTotal = customThreadPool.submit(() -> aList.parallelStream().reduce(0L, Long::sum)).get();
assertEquals((lastNum + firstNum) * lastNum / 2, actualTotal);
}
@Test
public void givenList_whenCallingParallelStream_shouldBeParallelStream() {
List<Long> aList = new ArrayList<>();
Stream<Long> parallelStream = aList.parallelStream();
assertTrue(parallelStream.isParallel());
}
}

13
core-java-concurrency/src/test/resources/.gitignore vendored Normal file
View File

@@ -0,0 +1,13 @@
*.class
#folders#
/target
/neoDb*
/data
/src/main/webapp/WEB-INF/classes
*/META-INF/*
# Packaged files #
*.jar
*.war
*.ear