#029 J2SE 5.0 Concurrency

public class ServerUsingThreadPoolExecutor {

private final static int MAX_THREADS = 2;

private final ServerSocket serverSocket;
private final ThreadPoolExecutor pool;
private final ArrayBlockingQueue<Runnable> workQueue;

/**
*  Constructor
**/
public ServerUsingThreadPoolExecutor(int port, int poolSize) throws IOException {
/*  Create a new ServerSocket to listen for incoming connections  */
serverSocket = new ServerSocket(port);

/*  In order to create a pool of threads, we must first have a queue
*  that will be used to hold the work tasks before they are executed
*  For this example we use a ArrayBlockingQueue that can hold the
*  same number of tasks as we have maximum threads
*/
workQueue = new ArrayBlockingQueue<Runnable>(MAX_THREADS);

/*  Now create a ThreadPool.  The initial and maximum number of
*  threads are the same in this example.  Please note that the
*  MAX_THREADS is set to 2.
*/
pool = new ThreadPoolExecutor(MAX_THREADS, MAX_THREADS,
60, TimeUnit.SECONDS, workQueue);
}

/**
*  Service requests
**/
public void serviceRequests() {
int count = 1;
int qLength = 0;

try {
for (;;) {
if ((qLength = workQueue.size()) > 0)
System.out.println(“Queue length is ” + qLength);

pool.execute(new ConnectionHandler(serverSocket.accept(), count++));
}
} catch (IOException ioe) {
System.out.println(“IO Error in ConnectionHandler: ” + ioe.getMessage());
pool.shutdown();
}
}

/**
*  Main entry point
*
* @param args Command line arguments
**/
public static void main(String[] args) {
System.out.println(“Listening for connections…”);
ServerUsingThreadPoolExecutor ce = null;

try {
ce = new ServerUsingThreadPoolExecutor(8100, 4);

/*
* Serve incoming connection request until interrupted.
*/
ce.serviceRequests();
} catch (IOException ioe) {
System.out.println(“IO Error creating listener: ” + ioe.getMessage());
}
}
}

public class ConnectionHandler implements Runnable {
private final Socket socket;
private final int connectionID;

/**
*  Constructor
*
* @param socket The socket on which incoming data will arrive
**/
public ConnectionHandler(Socket socket, int connectionID) {
this.socket = socket;
this.connectionID = connectionID;
}

/**
*  run method to do the work of the handler
**/
public void run() {
System.out.println(“Connection ” + connectionID + “, started”);

try {
InputStream is = socket.getInputStream();

//  Loop to do something with the socket here
while (true) {
byte[] inData = new byte[100];

/*  If the number of bytes read is less than zero then the connection
*  has been terminated so we end the thread
*/
if (is.read(inData) < 0)
break;

System.out.println(“[” + connectionID + “]: ” + new String(inData));
}
} catch (IOException ioe) {
// Ignore
}

System.out.println(“Connection ” + connectionID + “, ended”);
}
}

/**
*  ConnectionClient pass characters from the keyboard to a socket.  Simplified version of
*  telnet
*/
public class ConnectionClient {

/**
*  Main entry point
*
* @param args Command line arguments
**/
public static void main(String[] args) {
String host = null;
int port = 0;

if (args.length < 2) {
host = “127.0.0.1”;
port = 8100;
}

OutputStream os = null;

try {
// The Socket() call returns only when the connection
// request from this client is accepted by the server.
Socket s = new Socket(host, port);
os = s.getOutputStream();
System.out.println(“Connection established to server. Type characters and press <ENTER> to send”);
System.out.println(“Type EXIT and press <RETURN> to exit”);

/*  Read data from the standard input and send it to the remote socket  */
while (true) {
byte[] inData = new byte[100];

System.in.read(inData);
String inString = new String(inData);

if (inString.substring(0, 4).compareTo(“EXIT”) == 0)
System.exit(1);

os.write(inData);
}
} catch (Exception e) {
System.out.println(“Failed to connect to remote host: ” +
e.getMessage());
}
}
}

/**
*  Demonstration of the use of a Future to return the results from a
*  Callable.  This shows the simplicity of synchronising two concurrent
*  threads
**/
public class CallableAndFuture {
/**
*  Constructor
**/
public CallableAndFuture() {
}

/**
*  Run the test
**/
public void runTest() {
/*  Use the Executors utility method to get an ExecutorService for a
*  separate thread of execution
*/
ExecutorService threadExecutor = Executors.newSingleThreadExecutor();
System.out.println(“Starting test in first thread”);

/*  Now attempt to run a task in the second thread  */
try {
Future<String> future = threadExecutor.submit(new CallableExample());
/*  Set this sleep to either 100 0r 1000 to see the synchronisation
*  effects
*/
Thread.sleep(1000);
System.out.println(“First thread work complete.  Asking future for result”);
String result = future.get();
System.out.println(“Result from Future is ” + result);
} catch (Exception e) {
System.out.println(“Got an exception executing the test”);
System.out.println(e.getMessage());
}

/*  Shutdown the second thread so the program terminates gracefully
*/
threadExecutor.shutdown();
}

/**
*  Main entry point
*
* @param args The command line arguments
**/
public static void main(String[] args) {
CallableAndFuture fe = new CallableAndFuture();
fe.runTest();
}
}

/**
*  A simple implementation of the Callable interface that can return a value
*  to the thread that started it
**/
public class CallableExample implements Callable<String> {
/**
*  The entry point called when this object is invoked as a new thread
*  of execution
*
* @returns A String as a simple result
**/
public String call() {
System.out.println(“Starting call() method in second thread”);

try {
Thread.sleep(500);
} catch (InterruptedException ie) {
// Ignore
}

System.out.println(“Completed call() method in second thread”);
return “Finished”;
}
}

import java.net.*;
import java.io.*;
import java.util.concurrent.*;

/**
*  Use resources from a pool and return them at random times
**/
public class ResourceUser implements Runnable {
private ResourcePool pool;
private int threadId;
private int runCount;
private long hold;
private long pause;

/**
*  Constructor
*
* @param pool The pool to get the resource from
* @param threadId The id number of this thread
* @param runCount The number of times to run the test loop
* @param hold How long to hold the resource before releasing it
* @param pause How long to wait before trying to re-aquire the resource
**/
public ResourceUser(ResourcePool pool, int threadId, int runCount,
long hold, long pause) {
this.pool = pool;
this.threadId = threadId;
this.runCount = runCount;
this.hold = hold;
this.pause = pause;
}

/**
*  Request a resource from the pool and
*  return them after a random time period
**/
public void run() {
for (int i = 0; i < runCount; i++) {
try {
/*  Get the resource from the pool  */
System.out.println(“[” + threadId + “] trying to get resource”);
Integer resource = pool.getResource();
System.out.println(“[” + threadId + “] aquired resource”);

Thread.sleep(hold);
System.out.println(“[” + threadId + “] releasing resource”);
pool.putResource(resource);
Thread.sleep(pause);
} catch (InterruptedException ie) {
//  Silently ignore
}
}
}

/**
*  Main entry point
*
* @param args The command line arguments
**/
public static void main(String[] args) {
/*  Create a resource pool with two resources in it  */
ResourcePool pool = new ResourcePool(2);

/*  Use the Executors utility class to get a new FixedThreadPool.
*/
ExecutorService threadExecutor = Executors.newFixedThreadPool(3);

/*  Create three new resource users and start them using the previously
*  initialised thread pool
*/
ResourceUser r1 = new ResourceUser(pool, 1, 5, 400, 500);
ResourceUser r2 = new ResourceUser(pool, 2, 5, 800, 500);
ResourceUser r3 = new ResourceUser(pool, 3, 5, 1200, 500);
threadExecutor.execute(r1);
threadExecutor.execute(r2);
threadExecutor.execute(r3);

/*  Clean up the thread pool so the program terminates cleanly  */
threadExecutor.shutdown();
}
}

/**
*  Use a Semaphore to control the usage of a pool of resources.  If a
*  consumer requests a resource when all are being used the consumer
*  will block until a resource is returned and made available.
**/
public class ResourcePool {
private int poolSize;
private Semaphore available;
private Integer[] resources;
private boolean[] used;

/**
*  Constructor
*
* @param poolSize Size of fixed pool of resources
**/
public ResourcePool(int poolSize) {
this.poolSize = poolSize;

/*  Create a pool of resources (for this example just a set of Integer
*  objects.  Create a new semaphore to control access to the resources
*/
available = new Semaphore(poolSize);

used = new boolean[poolSize];
resources = new Integer[poolSize];

for (int i = 0; i < poolSize; i++)
resources[i] = new Integer(i);
}

/**
*  Get a resource.  If all are currently being used this will block
*  until one is returned to the pool.  This is a synchronised method
*  to make the code fully thread safe.
*
* @return The resource to use
**/
public Integer getResource() {
try {
available.acquire();
} catch (InterruptedException ie) {
// Ignore
}

for (int i = 0; i < poolSize; i++) {
if (used[i] == false) {
used[i] = true;
return resources[i];
}
}

return null;
}

/**
*  Return a resource to the pool
*
* @param resource The resource being returned to the pool
**/
public void putResource(Integer resource) {
/*  Note use of auto-unboxing  */
used[resource] = false;
available.release();
}
}