Thread data between threads with Java

Question

I am writing a multi-threaded application that simulates a movie theater. Each person involved is a stream of his own, and concurrency must be fully executed using semaphores. The only problem I encountered is how to basically bind the streams so that they can communicate (for example, through a channel).

For example:

The client [1], which is a stream, acquires a semaphore that allows it to approach Box Office. Client [1] should now inform the Box Office agent that they want to see the movie “X”. Then BoxOfficeAgent [1] also stream, must check that the film was not filled, either sell a ticket or inform the Client [1] to select another film.

How to transfer this data back and forth while maintaining concurrency semaphores?

Also, the only class that I can use from java.util.concurrent is the Semaphore class.

Answer 1

One simple way to transfer data between threads is to use the implementations of the BlockingQueue<E> interface located in the java.util.concurrent package.

These interfaces have methods for adding items to a collection with various forms of behavior:

• add(E) : adds, if possible, otherwise throws an exception
• boolean offer(E) : returns true if the item is added, otherwise
• boolean offer(E, long, TimeUnit) : trying to add an item, waiting for the specified amount of time
• put(E) : blocks the calling thread until the item is added.

It also defines methods for extracting elements with similar behavior:

• take() : blocks until an item is available
• poll(long, TimeUnit) : returns an element or returns null

The implementations that I use most often are: ArrayBlockingQueue , LinkedBlockingQueue and SynchronousQueue .

The first, ArrayBlockingQueue , has a fixed size, determined by the parameter passed to its constructor.

The second, LinkedBlockingQueue , has a limited size. It will always accept any elements, that is, offer will immediately return true, add will never throw an exception.

The third, and for me the most interesting, SynchronousQueue , is definitely a pipe. You can think of it as a queue with a size of 0. It will never save the element: this queue will only accept elements if there is any other thread trying to extract elements from it. Conversely, a search operation will only return an item if another thread tries to click it.

To fulfill the requirement of synchronization homework done exclusively with semaphores , you can get inspiration from the description I gave you about SynchronousQueue and write something very similar:

 class Pipe<E> { private E e; private final Semaphore read = new Semaphore(0); private final Semaphore write = new Semaphore(1); public final void put(final E e) { write.acquire(); this.e = e; read.release(); } public final E take() { read.acquire(); E e = this.e; write.release(); return e; } } 

Note that this class is similar to the behavior of what I described in SynchronousQueue.

After calling the put(E) methods, it gets a record semaphore that will remain empty, so another call to the same method will be blocked on the first line. Then this method saves a reference to the passed object and releases the read semaphore. This release will allow any thread that calls the take() method to continue.

The first step of the take() method, of course, is to get a readable semaphore to prevent any other thread from retrieving the item at the same time. After the item has been retrieved and stored in a local variable (exercise: what happens if this line, E e = this.e, was deleted?), The Method releases the write semaphore, so the put(E) method can be called again by any stream and returns what was stored in the local variable.

As an important note, note that the reference to the passed object is stored in a private field , and the take() and put(E) methods are equal as final >. This is of paramount importance and is often overlooked. If these methods were not final (or, even worse, the field is not private), the inheriting class could change the behavior of take() and put(E) , which violate the contract.

Finally, you could avoid having to declare a local variable in the take() method using try {} finally {} as follows:

 class Pipe<E> { // ... public final E take() { try { read.acquire(); return e; } finally { write.release(); } } } 

Here is the point of this example, if you just show the use of try/finally , which goes unnoticed among inexperienced developers. Obviously, in this case there is no real gain.

Damn, I basically finished your homework for you. In retaliation — and for you to test your knowledge of Semaphores — why don't you implement some of the other methods defined in the BlockingQueue contract? For example, you can implement the offer(E) and take(E, long, TimeUnit) !

Good luck.

Answer 2

Think about it in terms of shared memory with a read / write lock.

• Create a buffer to place the message.
• Access to the buffer should be controlled by locking / semaphore.
• Use this buffer to communicate between threads.

Hi

P