6. Concurrency

The many faces of concurrency

The problems that you solve with concurrency can be roughly classified as "speed" and "design manageability."

Faster execution

The speed issue sounds simple at first: If you want a program to run faster, break it into pieces and run each piece on a separate processor. Concurrency is a fundamental tool for multiprocessor programming.
However, concurrency can often improve the performance of programs running on a single processor. This sounds counter-intuitive with context-switch overhead, however blocking issues makes a difference.
In fact, from a performance standpoint, it makes no sense to use concurrency on a single-processor machine unless one of the tasks might block.
One very straightforward way to implement concurrency is at the operating system level, using processes.
Java took the more traditional approach of adding support for threading on top of a sequential language.
Instead of forking external processes in a multitasking operating system, threading creates tasks within the single process represented by the executing program.
One advantage that this provided was operating system transparency, which was an important design goal for Java.

Improving code design

Java’s threading is preemptive, which means that a scheduling mechanism provides time slices for each thread, periodically interrupting a thread and context switching to another thread so that each one is given a reasonable amount of time to drive its task.
Concurrency imposes costs, including complexity costs, but these are usually outweighed by improvements in program design, resource balancing, and user convenience.
In general, threads enable you to create a more loosely coupled design; otherwise, parts of your code would be forced to pay explicit attention to tasks that would normally be handled by threads.

Basic threading

Concurrent programming allows you to partition a program into separate, independently running tasks.
Using multithreading, each of these independent tasks (also called subtasks) is driven by a thread of execution.
A thread is a single sequential flow of control within a process.
A single process can thus have multiple concurrently executing tasks, but you program as if each task has the CPU to itself.
An underlying mechanism divides up the CPU time for you, but in general, you don’t need to think about it.

Defining tasks

A thread drives a task, so you need a way to describe that task. This is provided by the Runnable interface.
To define a task, simply implement Runnable and write a run( ) method to make the task do your bidding.
The identifier id distinguishes between multiple instances of the task.
A task’s run( ) method usually has some kind of loop that continues until the task is no longer necessary, so you must establish the condition on which to break out of this loop
The call to the static method Thread.yield( ) inside run( ) is a suggestion to the thread scheduler (the part of the Java threading mechanism that moves the CPU from one thread to the next). It’s completely optional.

//: concurrency/LiftOff.java

// Demonstration of the Runnable interface.

public class LiftOff implements Runnable {

protected int countDown = 10; // Default

private static int taskCount = 0;

private final int id = taskCount++;

public LiftOff() {}

public LiftOff(int countDown) {

this.countDown = countDown;

}

public String status() {

return "#" + id + "(" +

(countDown > 0 ? countDown : "Liftoff!") + "), ";

}

public void run() {

while(countDown-- > 0) {

System.out.print(status());

Thread.yield();

}

} ///:~

In the following example, the task’s run( ) is not driven by a separate thread; it is simply called directly in main( ) (actually, this is using a thread: the one that is always allocated for main( )):
When a class is derived from Runnable, it must have a run( ) method, but that’s nothing special—it doesn’t produce any innate threading abilities. To achieve threading behavior, you must explicitly attach a task to a thread.

//: concurrency/MainThread.java

public class MainThread {

public static void main(String[] args) {

LiftOff launch = new LiftOff();

launch.run();

}

} /* Output:

#0(9), #0(8), #0(7), #0(6), #0(5), #0(4), #0(3), #0(2), #0(1), #0(Liftoff!),

*///:~

The Thread class

The traditional way to turn a Runnable object into a working task is to hand it to a Thread constructor. This example shows how to drive a Liftoff object using a Thread:
A Thread constructor only needs a Runnable object.
Calling a Thread object’s start( ) will perform the necessary initialization for the thread and then call that Runnable’s run( ) method to start the task in the new thread.
In effect, you have made a method call to LiftOff.run( ), and that method has not yet finished, but because LiftOff.run( ) is being executed by a different thread, you can still perform other operations in the main( ) or other thread.

//: concurrency/BasicThreads.java

//The most basic use of the Thread class.

public class BasicThreads {

public static void main(String[] args) {

Thread t = new Thread(new LiftOff());

t.start();

System.out.println("Waiting for LiftOff");

}

} /* Output: (90% match)

Waiting for LiftOff

#0(9), #0(8), #0(7), #0(6), #0(5), #0(4), #0(3), #0(2), #0(1), #0(Liftoff!),

*///:~

You can easily add more threads to drive more tasks.
The execution of the different tasks is mixed together as the threads are swapped in and out.
In this case, a single thread (main( )), is creating all the LiftOff threads. If you have multiple threads creating LiftOff threads, however, it is possible for more than one LiftOff to have the same id.
When main( ) creates the Thread objects, it isn’t capturing the references for any of them. With an ordinary object, this would make it fair game for garbage collection, but not with a Thread. Each Thread "registers" itself so there is actually a reference to it someplace, and the garbage collector can’t clean it up until the task exits its run( ) and dies.

//: concurrency/MoreBasicThreads.java

//Adding more threads.

public class MoreBasicThreads {

public static void main(String[] args) {

for(int i = 0; i < 5; i++)

new Thread(new LiftOff()).start();

System.out.println("Waiting for LiftOff");

}

} /* Output: (Sample)

Waiting for LiftOff

#0(9), #1(9), #2(9), #3(9), #4(9), #0(8), #1(8), #2(8), #3(8), #4(8), #0(7), #1(7), #2(7), #3(7), #4(7), #0(6), #1(6), #2(6), #3(6), #4(6), #0(5), #1(5), #2(5), #3(5), #4(5), #0(4), #1(4), #2(4), #3(4), #4(4), #0(3), #1(3), #2(3), #3(3), #4(3), #0(2), #1(2), #2(2), #3(2), #4(2), #0(1), #1(1), #2(1), #3(1), #4(1), #0(Liftoff!), #1(Liftoff!), #2(Liftoff!), #3(Liftoff!), #4(Liftoff!),

*///:~

Using Executors

Java SE5 java.util.concurrent Executors simplify concurrent programming by managing Thread objects lifecycle for you.
An ExecutorService (an Executor with a service lifecycle—e.g., shutdown) knows how to build the appropriate context to execute Runnable objects.
The CachedThreadPool creates one thread per task
Very often, a single Executor can be used to create and manage all the tasks in your system.
The call to shutdown( ) prevents new tasks from being submitted to that Executor.
The current thread (in this case, the one driving main( )) will continue to run all tasks submitted before shutdown( ) was called.
The program will exit as soon as all the tasks in the Executor finish.

//: concurrency/CachedThreadPool.java

import java.util.concurrent.*;

public class CachedThreadPool {

public static void main(String[] args) {

ExecutorService exec = Executors.newCachedThreadPool();

for(int i = 0; i < 5; i++)

exec.execute(new LiftOff());

exec.shutdown();

}

// ExecutorService exec = Executors.newFixedThreadPool(5);

You can easily replace to a FixedThreadPool, it uses a limited set of threads to execute the submitted tasks
With the FixedThreadPool, you do expensive thread allocation once, up front, and you thus limit the number of threads.
This saves time for thread creation with each task, service quickly, keep bound on resources, existing threads are reused
Consider using FixedThreadPools in production code based on system resources
A SingleThreadExecutor is like a FixedThreadPool with a size of one thread, handy for executing small tasks
If more than one task is submitted to a SingleThreadExecutor, the tasks will be queued and each task will run to completion before the next task is begun, all using the same thread, maintains its own (hidden) queue of pending tasks, synchronizing on the shared resource e.g. file system.

Producing return values from tasks

If you want the task to produce a value when it’s done, you can implement the Callable interface rather than the Runnable interface.
Callable, introduced in Java SE5, is a generic with a type parameter representing the return value from the method call( ) (instead of run( )), and must be invoked using an ExecutorService submit( ) method.
It also offers an important concurrency guarantee that the others do not—no two tasks will be called concurrently. This changes the locking requirements for the tasks.
The submit( ) method produces a Future object, parameterized for the particular type of result returned by the Callable.
You can query the Future with isDone( ) to see if it has completed. When the task is completed and has a result, you can call get( ) to fetch the result.
You can simply call get( ) without checking isDone( ), in which case get( ) will block until the result is ready.
You can also call get( ) with a timeout, or isDone( ) to see if the task has completed, before trying to call get( ) to fetch the result.

//: concurrency/CallableDemo.java

import java.util.concurrent.*;

import java.util.*;

class TaskWithResult implements Callable<String> {

private int id;

public TaskWithResult(int id) {

this.id = id;

}

public String call() {

return "result of TaskWithResult " + id;

}

public class CallableDemo {

public static void main(String[] args) {

ExecutorService exec = Executors.newCachedThreadPool();

ArrayList<Future<String>> results =

new ArrayList<Future<String>>();

for(int i = 0; i < 10; i++)

results.add(exec.submit(new TaskWithResult(i)));

for(Future<String> fs : results)

try {

// get() blocks until completion:

System.out.println(fs.get());

} catch(InterruptedException e) {

System.out.println(e);

return;

} catch(ExecutionException e) {

System.out.println(e);

} finally {

exec.shutdown();

}

/* Output:

result of TaskWithResult 0

result of TaskWithResult 1

result of TaskWithResult 2

result of TaskWithResult 3

result of TaskWithResult 4

result of TaskWithResult 5

result of TaskWithResult 6

result of TaskWithResult 7

result of TaskWithResult 8

result of TaskWithResult 9

*///:~

Sleeping

A simple way to affect the behavior of your tasks is by calling sleep( ) to cease (block) the execution of that task for a given time. In the LiftOff class, if you replace the call to yield( ) with a call to sleep( ).
The call to sleep( ) can throw an InterruptedException, and you can see that this is caught in run( ). Because exceptions won’t propagate across threads back to main( ), you must locally handle any exceptions that arise within a task.
TimeUnit provides better readability by allowing you to specify the units of the sleep( ) delay
You may notice that the tasks run in "perfectly distributed" order—zero through four, however this behavior is not guranteed
If you must control the order of execution of tasks, your best bet is to use synchronization controls (described later) or, in some cases, not to use threads at all, but instead to write your own cooperative routines that hand control to each other in a specified order.

//: concurrency/SleepingTask.java

// Calling sleep() to pause for a while.

import java.util.concurrent.*;

public class SleepingTask extends LiftOff {

public void run() {

try {

while(countDown-- > 0) {

System.out.print(status());

// Old-style:

// Thread.sleep(100);

// Java SE5/6-style:

TimeUnit.MILLISECONDS.sleep(100);

}

} catch(InterruptedException e) {

System.err.println("Interrupted");

}

public static void main(String[] args) {

ExecutorService exec = Executors.newCachedThreadPool();

for(int i = 0; i < 5; i++)

exec.execute(new SleepingTask());

exec.shutdown();

}

/* Output:

*///:~

Priority

Lower-priority threads just tend to run less often.
You can read the priority of an existing thread with getPriority( ) and change it at any time with setPriority( ).
Even with the calculation, you see that the thread with MAX_PRIORITY is given a higher preference by the thread scheduler.
However, to ensure that a context switch occurs, yield( ) statements are regularly called.
Although the JDK has 10 priority levels, this doesn’t map well to many operating systems.
The only portable approach is to stick to MAX_PRIORITY, NORM_PRIORITY, and MIN_PRIORITY when you’re adjusting priority levels.

//: concurrency/SimplePriorities.java

// Shows the use of thread priorities.

import java.util.concurrent.*;

public class SimplePriorities implements Runnable {

private int countDown = 5;

private volatile double d; // No optimization

private int priority;

public SimplePriorities(int priority) {

this.priority = priority;

}

public String toString() {

return Thread.currentThread() + ": " + countDown;

}

public void run() {

Thread.currentThread().setPriority(priority);

while(true) {

// An expensive, interruptable operation:

for(int i = 1; i < 100000; i++) {

d += (Math.PI + Math.E) / (double)i;

if(i % 1000 == 0)

Thread.yield();

}

System.out.println(this);

if(--countDown == 0) return;

}

public static void main(String[] args) {

ExecutorService exec = Executors.newCachedThreadPool();

for(int i = 0; i < 5; i++)

exec.execute(

new SimplePriorities(Thread.MIN_PRIORITY));

exec.execute(

new SimplePriorities(Thread.MAX_PRIORITY));

exec.shutdown();

}

/* Output: (70% match)

Thread[pool-1-thread-6,10,main]: 5

Thread[pool-1-thread-6,10,main]: 4

Thread[pool-1-thread-6,10,main]: 3

Thread[pool-1-thread-6,10,main]: 2

Thread[pool-1-thread-6,10,main]: 1

Thread[pool-1-thread-3,1,main]: 5

Thread[pool-1-thread-2,1,main]: 5

Thread[pool-1-thread-1,1,main]: 5

Thread[pool-1-thread-5,1,main]: 5

Thread[pool-1-thread-4,1,main]: 5

...

*///:~

Yielding

If you know that you’ve accomplished what you need to during one pass through a loop in your run( ) method, you can give a hint to the threadscheduling mechanism that you’ve done enough and that some other task might as well have the CPU.
This hint (and it is a hint—there’s no guarantee your implementation will listen to it) takes the form of the yield( ) method. When you call yield( ), you are suggesting that other threads of the same priority might be run.
In general, however, you can’t rely on yield( ) for any serious control or tuning of your application. Indeed, yield( ) is often used incorrectly.

Daemon Threads

A "daemon" thread is intended to provide a general service in the background as long as the program is running, but is not part of the essence of the program.
Thus, when all of the non-daemon threads complete, the program is terminated, killing all daemon threads in the process. Conversely, if there are any non-daemon threads still running, the program doesn’t terminate. There is, for instance, a non-daemon thread that runs main( ).
You must set the thread to be a daemon by calling setDaemon( ) before it is started.
SimpleDaemons.java creates explicit Thread objects in order to set their daemon flag. It is possible to customize the attributes (daemon, priority, name) of threads created by Executors by writing a custom ThreadFactory

//: concurrency/SimpleDaemons.java

// Daemon threads don’t prevent the program from ending.

import java.util.concurrent.*;

import static net.mindview.util.Print.*;

public class SimpleDaemons implements Runnable {

public void run() {

try {

while(true) {

TimeUnit.MILLISECONDS.sleep(100);

print(Thread.currentThread() + " " + this);

}

} catch(InterruptedException e) {

print("sleep() interrupted");

}

public static void main(String[] args) throws Exception {

for(int i = 0; i < 10; i++) {

Thread daemon = new Thread(new SimpleDaemons());

daemon.setDaemon(true); // Must call before start()

daemon.start();

}

print("All daemons started");

TimeUnit.MILLISECONDS.sleep(175);

}

/* Output: (Sample)

All daemons started

Thread[Thread-0,5,main] SimpleDaemons@530daa

Thread[Thread-1,5,main] SimpleDaemons@a62fc3

Thread[Thread-2,5,main] SimpleDaemons@89ae9e

Thread[Thread-3,5,main] SimpleDaemons@1270b73

Thread[Thread-4,5,main] SimpleDaemons@60aeb0

Thread[Thread-5,5,main] SimpleDaemons@16caf43

Thread[Thread-6,5,main] SimpleDaemons@66848c

Thread[Thread-7,5,main] SimpleDaemons@8813f2

Thread[Thread-8,5,main] SimpleDaemons@1d58aae

Thread[Thread-9,5,main] SimpleDaemons@83cc67

...

*///:~

The only difference from an ordinary ThreadFactory is that this one sets the daemon status to true. You can now pass a new DaemonThreadFactory as an argument to Executors.newCachedThreadPool( ).
Each of the static ExecutorService creation methods is overloaded to take a ThreadFactory object that it will use to create new threads.

//: net/mindview/util/DaemonThreadFactory.java

package net.mindview.util;

import java.util.concurrent.*;

public class DaemonThreadFactory implements ThreadFactory {

public Thread newThread(Runnable r) {

Thread t = new Thread(r);

t.setDaemon(true);

return t;

}

} ///:~

//: concurrency/DaemonFromFactory.java

// Using a Thread Factory to create daemons.

import java.util.concurrent.*;

import net.mindview.util.*;

import static net.mindview.util.Print.*;

public class DaemonFromFactory implements Runnable {

public void run() {

try {

while(true) {

TimeUnit.MILLISECONDS.sleep(100);

print(Thread.currentThread() + " " + this);

}

} catch(InterruptedException e) {

print("Interrupted");

}

public static void main(String[] args) throws Exception {

ExecutorService exec = Executors.newCachedThreadPool(

new DaemonThreadFactory());

for(int i = 0; i < 10; i++)

exec.execute(new DaemonFromFactory());

print("All daemons started");

TimeUnit.MILLISECONDS.sleep(500); // Run for a while

}

} /* (Execute to see output) *///:~

You can find out if a thread is a daemon by calling isDaemon( ).
If a thread is a daemon, then any threads it creates will automatically be daemons, as the following example demonstrates

//: concurrency/Daemons.java

// Daemon threads spawn other daemon threads.

import java.util.concurrent.*;

import static net.mindview.util.Print.*;

class Daemon implements Runnable {

private Thread[] t = new Thread[10];

public void run() {

for(int i = 0; i < t.length; i++) {

t[i] = new Thread(new DaemonSpawn());

t[i].start();

printnb("DaemonSpawn " + i + " started, ");

}

for(int i = 0; i < t.length; i++)

printnb("t[" + i + "].isDaemon() = " +

t[i].isDaemon() + ", ");

while(true)

Thread.yield();

}

class DaemonSpawn implements Runnable {

public void run() {

while(true)

Thread.yield();

}

public class Daemons {

public static void main(String[] args) throws Exception {

Thread d = new Thread(new Daemon());

d.setDaemon(true);

d.start();

printnb("d.isDaemon() = " + d.isDaemon() + ", ");

// Allow the daemon threads to

// finish their startup processes:

TimeUnit.SECONDS.sleep(1);

}

/* Output: (Sample)

d.isDaemon() = true, DaemonSpawn 0 started, DaemonSpawn 1 started, DaemonSpawn 2 started, DaemonSpawn 3 started, DaemonSpawn 4 started, DaemonSpawn 5 started, DaemonSpawn 6 started, DaemonSpawn 7 started, DaemonSpawn 8 started, DaemonSpawn 9 started, t[0].isDaemon() = true, t[1].isDaemon() = true, t[2].isDaemon() = true, t[3].isDaemon() = true, t[4].isDaemon() = true, t[5].isDaemon() = true, t[6].isDaemon() = true, t[7].isDaemon() = true, t[8].isDaemon() = true, t[9].isDaemon() = true,

*///:~

- You should be aware that daemon threads will terminate their run( ) methods without executing finally clauses.

Daemons are terminated "abruptly" not "gracefully" (so finally is not executed) when the last of the non-daemons terminates.
Non-daemon Executors are generally a better approach, since all the tasks controlled by an Executor can be shut down at once.

//: concurrency/DaemonsDontRunFinally.java

// Daemon threads don’t run the finally clause

import java.util.concurrent.*;

import static net.mindview.util.Print.*;

class ADaemon implements Runnable {

public void run() {

try {

print("Starting ADaemon");

TimeUnit.SECONDS.sleep(1);

} catch(InterruptedException e) {

print("Exiting via InterruptedException");

} finally {

print("This should always run?");

}

public class DaemonsDontRunFinally {

public static void main(String[] args) throws Exception {

Thread t = new Thread(new ADaemon());

t.setDaemon(true);

t.start();

}

/* Output:

Starting ADaemon

*///:~

Coding Variations

In very simple cases, you may want to use the alternative approach of inheriting directly from Thread

//: concurrency/SimpleThread.java

// Inheriting directly from the Thread class.

public class SimpleThread extends Thread {

private int countDown = 5;

private static int threadCount = 0;

public SimpleThread() {

// Store the thread name:

super(Integer.toString(++threadCount));

start();

}

public String toString() {

return "#" + getName() + "(" + countDown + "), ";

}

public void run() {

while(true) {

System.out.print(this);

if(--countDown == 0)

return;

}

public static void main(String[] args) {

for(int i = 0; i < 5; i++)

new SimpleThread();

}

/* Output:

#1(5), #1(4), #1(3), #1(2), #1(1), #2(5), #2(4), #2(3), #2(2), #2(1), #3(5), #3(4), #3(3), #3(2), #3(1), #4(5), #4(4), #4(3), #4(2), #4(1), #5(5), #5(4), #5(3), #5(2), #5(1),

*///:~

Terminology

Conceptually, we want to create a task that runs independently of other tasks, so we ought to be able to define a task, and then say "go," and not worry about details.
There’s a distinction between the task that’s being executed and the thread that drives it.
But physically, threads can be expensive to create, so you must conserve and manage them.
Thus it makes sense from an implementation standpoint to separate tasks from threads.
In addition, Java threading is based on the low-level pthreads approach which comes from C
Some of this low-level nature has trickled through into the Java implementation, so to stay at a higher level of abstraction, you must use discipline when writing code
Thus, if you are discussing a system at a conceptual level, you could just use the term "task" without mentioning the driving mechanism (a.k.a. Thread) at all.

Joining a thread

One thread may call join( ) on another thread to wait for the second thread to complete before proceeding. If a thread calls t.join( ) on another thread t, then the calling thread is suspended until the target thread t finishes (when t.isAlive( ) is false).
You may also call join( ) with a timeout argument (in either milliseconds or milliseconds and nanoseconds) so that if the target thread doesn’t finish in that period of time, the call to join( ) returns anyway.
The call to join( ) may be aborted by calling interrupt( ) on the calling thread, so a try-catch clause is required.
A Joiner is a task that waits for a Sleeper to wake up by calling join( ) on the Sleeper object. In main( ), each Sleeper has a Joiner, and you can see in the output that if the Sleeper either is interrupted or ends normally, the Joiner completes in conjunction with the Sleeper.

//: concurrency/Joining.java

// Understanding join().

import static net.mindview.util.Print.*;

class Sleeper extends Thread {

private int duration;

public Sleeper(String name, int sleepTime) {

super(name);

duration = sleepTime;

start();

}

public void run() {

try {

sleep(duration);

} catch(InterruptedException e) {

print(getName() + " was interrupted. " + "isInterrupted(): " + isInterrupted());

return;

}

print(getName() + " has awakened");

}

class Joiner extends Thread {

private Sleeper sleeper;

public Joiner(String name, Sleeper sleeper) {

super(name);

this.sleeper = sleeper;

start();

}

public void run() {

try {

sleeper.join();

} catch(InterruptedException e) {

print("Interrupted");

}

print(getName() + " join completed");

}

public class Joining {

public static void main(String[] args) {

Sleeper sleepy = new Sleeper("Sleepy", 1500),

grumpy = new Sleeper("Grumpy", 1500);

Joiner dopey = new Joiner("Dopey", sleepy),

doc = new Joiner("Doc", grumpy);

grumpy.interrupt();

}

/* Output:

Grumpy was interrupted. isInterrupted(): false

Doc join completed

Sleepy has awakened

Dopey join completed

*///:~

Thread groups

Thread groups are best viewed as an unsuccessful experiment, and you may simply ignore their existence.

Sharing resources

With concurrency, you now have the possibility of two or more tasks interfering with each other.

Resolving shared resource contention

To solve the problem of thread collision, virtually all concurrency schemes serialize access to shared resources.
This means that only one task at a time is allowed to access the shared resource. This is ordinarily accomplished by putting a clause around a piece of code that only allows one task at a time to pass through that piece of code.
Because this clause produces mutual exclusion, a common name for such a mechanism is mutex.
To prevent collisions over resources, Java has built-in support in the form of the synchronized keyword.
The shared resource is typically just a piece of memory in the form of an object, but may also be a file, an I/O port, or something like a printer.
To control access to a shared resource, you first put it inside an object. Then any method that uses the resource can be made synchronized.
In production code, you’ve already seen that you should make the data elements of a class private and access that memory only through methods. You can prevent collisions by declaring those methods synchronized.

synchronized void f() { /* ... */ }

synchronized void g() { /* ... */ }

All objects automatically contain a single lock (also referred to as a monitor).
When you call any synchronized method, that object is locked and no other synchronized method of that object can be called until the first one finishes and releases the lock.
For the preceding methods, if f( ) is called for an object by one task, a different task cannot call f( ) or g( ) for the same object until f( ) is completed and releases the lock.
Note that it’s especially important to make fields private when working with concurrency
As a task acquires the lock for the first time, the count goes to one. Each time the same task acquires another lock on the same object, the count is incremented.
Naturally, multiple lock acquisition is only allowed for the task that acquired the lock in the first place. Each time the task leaves a synchronized method, the count is decremented, until the count goes to zero, releasing the lock entirely for use by other tasks.
There’s also a single lock per class (as part of the Class object for the class), so that synchronized static methods can lock each other out from simultaneous access of static data on a class-wide basis.
Every method in the class object that accesses a critical shared resource must be synchronized or it won’t work right.

Condition Variable

Condition variables are a mechanism that allow you to test that a particular condition holds true before allowing your method to proceed.

...does a condition variable necessarily have to be within the 'mutex.acquire()' and 'mutex.release()' block?

Any calls to change the condition variables do need to be within a synchronized region - this can be through the built in synchronized keyword or one of the synchronizer classes provided by the java.util.concurrent package such as Lock. If you did not synchronize the condition variables there are two possible negative outcomes:
A missed signal - this is where one thread checks a condition and finds it does not hold, but before it blocks another thread comes in, performs some action to cause the condition to become true, and then signals all threads waiting on the condition. Unfortunately the first thread has already checked the condition and will block anyway even though it could actually proceed.
The second issue is the usual problem where you can have multiple threads attempting to modify the shared state simultaneously. In the case of your example multiple threads may call put() simultaneously, all of them then check the condition and see that the array is not full and attempt to insert into it, thereby overwriting elements in the array.

References

Reentrant Locks
What are Reentrant Locks
Synchronization and Locks
- Bounded Buffer Locks and Conditions
Producer Consumer