diff --git a/src/main/java/com/thealgorithms/datastructures/queues/ThreadSafeQueue.java b/src/main/java/com/thealgorithms/datastructures/queues/ThreadSafeQueue.java
new file mode 100644
index 000000000000..a943b0028974
--- /dev/null
+++ b/src/main/java/com/thealgorithms/datastructures/queues/ThreadSafeQueue.java
@@ -0,0 +1,186 @@
+package com.thealgorithms.datastructures.queues;
+
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * @brief Thread-safe bounded queue implementation using ReentrantLock and Condition variables
+ * @details A blocking queue that supports multiple producers and consumers.
+ * Uses a circular buffer internally with lock-based synchronization to ensure
+ * thread safety. Producers block when the queue is full, and consumers block
+ * when the queue is empty.
+ * @see Producer-Consumer Problem
+ */
+public class ThreadSafeQueue {
+
+ private final Object[] buffer;
+ private final int capacity;
+ private int head;
+ private int tail;
+ private int count;
+ private final ReentrantLock lock;
+ private final Condition notFull;
+ private final Condition notEmpty;
+
+ /**
+ * @brief Constructs a ThreadSafeQueue with the specified capacity
+ * @param capacity the maximum number of elements the queue can hold
+ * @throws IllegalArgumentException if capacity is less than or equal to zero
+ */
+ public ThreadSafeQueue(int capacity) {
+ if (capacity <= 0) {
+ throw new IllegalArgumentException("Capacity must be greater than zero.");
+ }
+ this.capacity = capacity;
+ this.buffer = new Object[capacity];
+ this.head = 0;
+ this.tail = 0;
+ this.count = 0;
+ this.lock = new ReentrantLock();
+ this.notFull = lock.newCondition();
+ this.notEmpty = lock.newCondition();
+ }
+
+ /**
+ * @brief Adds an element to the tail of the queue, blocking if full
+ * @param item the element to add
+ * @throws InterruptedException if the thread is interrupted while waiting
+ * @throws IllegalArgumentException if the item is null
+ */
+ public void enqueue(T item) throws InterruptedException {
+ if (item == null) {
+ throw new IllegalArgumentException("Cannot enqueue null item.");
+ }
+
+ lock.lock();
+ try {
+ while (count == capacity) {
+ notFull.await();
+ }
+ buffer[tail] = item;
+ tail = (tail + 1) % capacity;
+ count++;
+ notEmpty.signalAll();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * @brief Removes and returns the element at the head of the queue, blocking if empty
+ * @return the element at the head of the queue
+ * @throws InterruptedException if the thread is interrupted while waiting
+ */
+ @SuppressWarnings("unchecked")
+ public T dequeue() throws InterruptedException {
+ lock.lock();
+ try {
+ while (count == 0) {
+ notEmpty.await();
+ }
+ T item = (T) buffer[head];
+ buffer[head] = null;
+ head = (head + 1) % capacity;
+ count--;
+ notFull.signalAll();
+ return item;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * @brief Adds an element to the tail of the queue without blocking
+ * @param item the element to add
+ * @return true if the element was added, false if the queue was full
+ * @throws IllegalArgumentException if the item is null
+ */
+ public boolean offer(T item) {
+ if (item == null) {
+ throw new IllegalArgumentException("Cannot enqueue null item.");
+ }
+
+ lock.lock();
+ try {
+ if (count == capacity) {
+ return false;
+ }
+ buffer[tail] = item;
+ tail = (tail + 1) % capacity;
+ count++;
+ notEmpty.signalAll();
+ return true;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * @brief Removes and returns the element at the head without blocking
+ * @return the element at the head, or null if the queue is empty
+ */
+ @SuppressWarnings("unchecked")
+ public T poll() {
+ lock.lock();
+ try {
+ if (count == 0) {
+ return null;
+ }
+ T item = (T) buffer[head];
+ buffer[head] = null;
+ head = (head + 1) % capacity;
+ count--;
+ notFull.signalAll();
+ return item;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * @brief Returns the number of elements in the queue
+ * @return the current size of the queue
+ */
+ public int size() {
+ lock.lock();
+ try {
+ return count;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * @brief Checks if the queue is empty
+ * @return true if the queue contains no elements
+ */
+ public boolean isEmpty() {
+ lock.lock();
+ try {
+ return count == 0;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * @brief Checks if the queue is full
+ * @return true if the queue has reached its capacity
+ */
+ public boolean isFull() {
+ lock.lock();
+ try {
+ return count == capacity;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * @brief Returns the maximum capacity of the queue
+ * @return the capacity
+ */
+ public int capacity() {
+ return capacity;
+ }
+}
diff --git a/src/test/java/com/thealgorithms/datastructures/queues/ThreadSafeQueueTest.java b/src/test/java/com/thealgorithms/datastructures/queues/ThreadSafeQueueTest.java
new file mode 100644
index 000000000000..4c038c05b167
--- /dev/null
+++ b/src/test/java/com/thealgorithms/datastructures/queues/ThreadSafeQueueTest.java
@@ -0,0 +1,295 @@
+package com.thealgorithms.datastructures.queues;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ThreadSafeQueueTest {
+
+ @Test
+ public void testEnqueueDequeue() throws InterruptedException {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(5);
+ queue.enqueue(1);
+ queue.enqueue(2);
+ queue.enqueue(3);
+
+ Assertions.assertEquals(3, queue.size());
+ Assertions.assertEquals(1, queue.dequeue());
+ Assertions.assertEquals(2, queue.dequeue());
+ Assertions.assertEquals(3, queue.dequeue());
+ Assertions.assertTrue(queue.isEmpty());
+ }
+
+ @Test
+ public void testOfferPoll() {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(2);
+ Assertions.assertTrue(queue.offer("a"));
+ Assertions.assertTrue(queue.offer("b"));
+ Assertions.assertFalse(queue.offer("c"));
+
+ Assertions.assertEquals("a", queue.poll());
+ Assertions.assertEquals("b", queue.poll());
+ Assertions.assertNull(queue.poll());
+ }
+
+ @Test
+ public void testOfferRejectsWhenFull() {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(2);
+ Assertions.assertTrue(queue.offer(1));
+ Assertions.assertTrue(queue.offer(2));
+ Assertions.assertFalse(queue.offer(3));
+ Assertions.assertEquals(2, queue.size());
+ }
+
+ @Test
+ public void testPollReturnsNullWhenEmpty() {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(5);
+ Assertions.assertNull(queue.poll());
+ }
+
+ @Test
+ public void testEnqueueNullThrows() {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(5);
+ Assertions.assertThrows(IllegalArgumentException.class, () -> queue.enqueue(null));
+ }
+
+ @Test
+ public void testOfferNullThrows() {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(5);
+ Assertions.assertThrows(IllegalArgumentException.class, () -> queue.offer(null));
+ }
+
+ @Test
+ public void testInvalidCapacityThrows() {
+ Assertions.assertThrows(IllegalArgumentException.class, () -> new ThreadSafeQueue<>(0));
+ Assertions.assertThrows(IllegalArgumentException.class, () -> new ThreadSafeQueue<>(-1));
+ }
+
+ @Test
+ public void testIsEmptyAndIsFull() throws InterruptedException {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(2);
+ Assertions.assertTrue(queue.isEmpty());
+ Assertions.assertFalse(queue.isFull());
+
+ queue.enqueue(1);
+ Assertions.assertFalse(queue.isEmpty());
+ Assertions.assertFalse(queue.isFull());
+
+ queue.enqueue(2);
+ Assertions.assertFalse(queue.isEmpty());
+ Assertions.assertTrue(queue.isFull());
+
+ queue.dequeue();
+ Assertions.assertFalse(queue.isEmpty());
+ Assertions.assertFalse(queue.isFull());
+
+ queue.dequeue();
+ Assertions.assertTrue(queue.isEmpty());
+ Assertions.assertFalse(queue.isFull());
+ }
+
+ @Test
+ public void testCapacity() {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(10);
+ Assertions.assertEquals(10, queue.capacity());
+ }
+
+ @Test
+ public void testCircularBufferWrapAround() throws InterruptedException {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(3);
+ queue.enqueue(1);
+ queue.enqueue(2);
+ queue.enqueue(3);
+
+ Assertions.assertEquals(1, queue.dequeue());
+ Assertions.assertEquals(2, queue.dequeue());
+
+ queue.enqueue(4);
+ queue.enqueue(5);
+
+ Assertions.assertEquals(3, queue.dequeue());
+ Assertions.assertEquals(4, queue.dequeue());
+ Assertions.assertEquals(5, queue.dequeue());
+ }
+
+ @Test
+ public void testMultipleProducersSingleConsumer() throws InterruptedException {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(100);
+ int numProducers = 4;
+ int itemsPerProducer = 250;
+ int totalItems = numProducers * itemsPerProducer;
+ CountDownLatch doneLatch = new CountDownLatch(numProducers);
+ List results = new ArrayList<>();
+
+ ExecutorService executor = Executors.newFixedThreadPool(numProducers + 1);
+
+ for (int p = 0; p < numProducers; p++) {
+ final int producerId = p;
+ executor.submit(() -> {
+ try {
+ for (int i = 0; i < itemsPerProducer; i++) {
+ queue.enqueue(producerId * itemsPerProducer + i);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ doneLatch.countDown();
+ }
+ });
+ }
+
+ Thread consumerThread = new Thread(() -> {
+ try {
+ while (results.size() < totalItems) {
+ Integer item = queue.poll();
+ if (item != null) {
+ synchronized (results) {
+ results.add(item);
+ }
+ }
+ }
+ } catch (Exception e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ consumerThread.start();
+
+ Assertions.assertTrue(doneLatch.await(10, TimeUnit.SECONDS));
+ consumerThread.join(5000);
+
+ Assertions.assertEquals(totalItems, results.size());
+ executor.shutdown();
+ Assertions.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+
+ @Test
+ public void testSingleProducerMultipleConsumers() throws InterruptedException {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(50);
+ int numConsumers = 4;
+ int totalItems = 1000;
+ CountDownLatch doneLatch = new CountDownLatch(numConsumers);
+ AtomicInteger consumedCount = new AtomicInteger(0);
+
+ ExecutorService executor = Executors.newFixedThreadPool(numConsumers + 1);
+
+ executor.submit(() -> {
+ try {
+ for (int i = 0; i < totalItems; i++) {
+ queue.enqueue(i);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+
+ for (int c = 0; c < numConsumers; c++) {
+ executor.submit(() -> {
+ try {
+ while (consumedCount.get() < totalItems) {
+ Integer item = queue.poll();
+ if (item != null) {
+ consumedCount.incrementAndGet();
+ }
+ }
+ } finally {
+ doneLatch.countDown();
+ }
+ });
+ }
+
+ Assertions.assertTrue(doneLatch.await(10, TimeUnit.SECONDS));
+ Assertions.assertEquals(totalItems, consumedCount.get());
+ executor.shutdown();
+ Assertions.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+
+ @Test
+ public void testBlockingEnqueueWhenFull() throws InterruptedException {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(1);
+ queue.enqueue(1);
+
+ AtomicInteger blockedCount = new AtomicInteger(0);
+ Thread producer = new Thread(() -> {
+ try {
+ queue.enqueue(2);
+ blockedCount.incrementAndGet();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ producer.start();
+
+ Thread.sleep(100);
+ Assertions.assertEquals(1, queue.dequeue());
+
+ producer.join(2000);
+ Assertions.assertEquals(1, blockedCount.get());
+ Assertions.assertEquals(2, queue.dequeue());
+ }
+
+ @Test
+ public void testBlockingDequeueWhenEmpty() throws InterruptedException {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(5);
+
+ AtomicInteger result = new AtomicInteger(-1);
+ Thread consumer = new Thread(() -> {
+ try {
+ result.set(queue.dequeue());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ consumer.start();
+
+ Thread.sleep(100);
+ queue.enqueue(42);
+
+ consumer.join(2000);
+ Assertions.assertEquals(42, result.get());
+ }
+
+ @Test
+ public void testStressConcurrentAccess() throws InterruptedException {
+ ThreadSafeQueue queue = new ThreadSafeQueue<>(10);
+ int numThreads = 8;
+ int opsPerThread = 500;
+ CountDownLatch latch = new CountDownLatch(numThreads);
+ AtomicInteger enqueueCount = new AtomicInteger(0);
+ AtomicInteger dequeueCount = new AtomicInteger(0);
+
+ ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+
+ for (int t = 0; t < numThreads; t++) {
+ final boolean isProducer = t % 2 == 0;
+ executor.submit(() -> {
+ try {
+ for (int i = 0; i < opsPerThread; i++) {
+ if (isProducer) {
+ if (queue.offer(i)) {
+ enqueueCount.incrementAndGet();
+ }
+ } else {
+ if (queue.poll() != null) {
+ dequeueCount.incrementAndGet();
+ }
+ }
+ }
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ Assertions.assertTrue(latch.await(10, TimeUnit.SECONDS));
+ Assertions.assertTrue(enqueueCount.get() >= dequeueCount.get());
+ Assertions.assertEquals(enqueueCount.get() - dequeueCount.get(), queue.size());
+ executor.shutdown();
+ executor.awaitTermination(5, TimeUnit.SECONDS);
+ }
+}