3
\$\begingroup\$

I don't like the service class below, specifically, the way it exposes the internals of the producer. While it would be possible to combine all three classes into one producer-consumer, I would rather that they stay separated.

Service:

package net.bounceme.dur.client;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.logging.Logger;
import net.bounceme.dur.data.Title;

public class ProducerConsumerService implements Executor {

    private static final Logger log = Logger.getLogger(ProducerConsumerService.class.getName());
    private final BlockingQueue<Title> queue = new ArrayBlockingQueue<>(1);
    private Producer producer = null;
    private Consumer consumer = null;
    private Title title = null;

    public void ProducerConsumerService() {
    }

    public Title produce() throws InterruptedException, IOException, ClassNotFoundException {
        producer.produce();
        title = queue.take();
        consumer.consume(title);
        return title;
    }

    public void startService() throws IOException {
        Properties props = PropertiesReader.getProps();
        int portNumber = Integer.parseInt(props.getProperty("port"));
        String host = props.getProperty("server");
        Socket socket = new Socket(host, portNumber);
        ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
        ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
        producer = new Producer(queue, objectInputStream);
        consumer = new Consumer(queue, objectOutputStream);
        new Thread((Runnable) producer).start();
        new Thread((Runnable) consumer).start();
        log.info("started...");
    }

    @Override
    public void execute(Runnable command) {
        new Thread(command).start();
    }
}

Producer:

package net.bounceme.dur.client;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.concurrent.BlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;
import net.bounceme.dur.data.State;
import net.bounceme.dur.data.Title;

public class Producer implements Runnable {

    private static final Logger log = Logger.getLogger(Producer.class.getName());
    private final BlockingQueue<Title> queue;
    private ObjectInputStream objectInputStream = null;
    private Title title = null;

    public Producer(BlockingQueue<Title> q, ObjectInputStream objectInputStream) {
        this.queue = q;
        this.objectInputStream = objectInputStream;
    }

     void produce() throws IOException, ClassNotFoundException, InterruptedException {
        title = (Title) objectInputStream.readObject();
        queue.put(title);
    }

    @Override
    public void run() {
        try {
            produce();
        } catch (IOException | ClassNotFoundException | InterruptedException ex) {
            Logger.getLogger(Producer.class.getName()).log(Level.SEVERE, null, ex);
            //exit?
        }
    }
}

Consumer:

package net.bounceme.dur.client;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.concurrent.BlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;
import net.bounceme.dur.data.State;
import net.bounceme.dur.data.Title;

public class Consumer implements Runnable {

    private final BlockingQueue<Title> queue;
    private ObjectOutputStream objectOutputStream = null;

    public Consumer(BlockingQueue<Title> queue, ObjectOutputStream objectOutputStream) {
        this.queue = queue;
        this.objectOutputStream = objectOutputStream;
    }

    void consume(Title title) throws InterruptedException, IOException {
        title.setState(State.x);
        objectOutputStream.writeObject(title);
    }


    private void consume() throws InterruptedException, IOException {
        Title title = queue.take();
        title.setState(State.x);
        objectOutputStream.writeObject(title);
    }

    @Override
    public void run() {
        try {
            consume();
        } catch (InterruptedException | IOException ex) {
            Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE, null, ex);
            //exit?
        }
    }
}

I do like that the service class provides a sort of interface for dealing with the producer and consumer classes.

Code inspired from here.

\$\endgroup\$
2
\$\begingroup\$

Yeah, your dependency graph has gotten all tangled up.

In this case, the root problem seems to be that there are a number of different things going on, and you haven't really identified them. So you have your producer/consumer idiom, your queuing, and your threading conflated.

My recommendation is to start from a simple case, and then gradually add more elements.

In producer/consumer (publish/subscribe, event handling, etc), I find the the cleanest approach to take is to start by working a single threaded example, to identify the right abstractions, and then to add threading after it's clear that the basic abstractions are right.

In a single thread, the simplest produce consume logic would look something like

Title title = producer.produce();
consumer.consume(title);

A key idea is that, in this trivial example, it should be clear that the glue code doesn't care where producer is getting the thing, or what the consumer is doing with it. The producer and consumer are completely unaware of each other as well. This helps us to learn that we have two simple interfaces to consider

interface TitleProducer {
    Title produce();
}

interface TitleConsumer {
    void consume(Title title);
}

Where does the Title come from? That's an implementation detail - the code that uses the interface doesn't need to know anything about it. But the code that implements the interface does, and the code that creates the instance of the implementation will have some awareness of it.

For example, we can put together a really simple implementation.

class TrivialTitleProducer implements TitleProducer {
    private final Title title;

    TrivialTitleProducer(Title title) {
        this.title = title;
    }

    public Title produce () {
        return title;
    }
}

class TrivialTitleConsumer implements TitleConsumer {
    public void consume(Title title) {
        System.out.println(title.toString());
    }
}

Granted, those aren't particularly exciting implementations, but they are enough to see that the hand off is working.

Producer producer = new TrivialTitleProducer(new Title());
Consumer consumer = new TrivialTitleConsumer();

Title title = producer.produce();
consumer.consume(title);

Now, in your case, you had specific streams in mind to produce and consume the objects. The interfaces we've discovered so far don't allow you to specify the streams to be used -- that's deliberate. Because the streams are really part of the implementation only, we don't want them in the signature of the methods, but instead in the signature of the constructors. This is usually described as dependency injection (a useful idea to google).

StreamTitleProducer implements TitleProducer {
    private final ObjectInputStream in;

    StreamTitleProducer(ObjectInputStream in) {
        this.in = in;
    }

    public Title produce () {
        return (Title) in.readObject();
    }
}

StreamTitleConsumer implements TitleConsumer {
    private final ObjectOutputStream out;

    StreamTitleConsumer(ObjectOutputStream out) {
        this.out = out;
    }

    public void consume(Title title) {
        out.writeObject(title);
    }
}

You can easily see that our earlier test is readily adapted to the useful implementation

ObjectInputStream in = ...
ObjectOutputStream out = ...

Producer producer = new StreamTitleProducer(in);
Consumer consumer = new StreamTitleConsumer(out);

Title title = producer.produce();
consumer.consume(title);

Now, we want to produce and consume on separate threads. This is actually two different problems to solve

  1. How do we exchange data between a producer on one thread and a consumer on another thread?
  2. How do we execute that code on separate threads.

Important trick -- implement the thread-safe data exchange in a single thread. Let's walk through it slowly. BlockingQueue<T> is a reasonable way to communicate between threads, so let's start there. What would a trivial implementation look like?

BlockingQueue<Title> queue = ...

Title title = producer.produce();
queue.put(title);
Title queuedTitle = queue.take();
consumer.consume(title);

Yeah, it's a complicated way to do a single thread hand off. But it affords a key insight - queue.put() looks just like consume() it, and queue.take() looks just like produce(). So let's codify that.

QueueTitleProducer implements TitleProducer {
    private final BlockingQueue<Title> queue;

    QueueTitleProducer(BlockingQueue<Title> queue) {
        this.queue = queue;
    }

    public Title produce () {
        return this.queue.take();
    }
}

QueueTitleConsumer implements TitleConsumer {
    private final BlockingQueue<Title> queue;

    QueueTitleConsumer(BlockingQueue<Title> queue) {
        this.queue = queue;
    }

    public void consume(Title title) {
        this.queue.put(title);
    }
}

BlockingQueue<Title> queue = ...
Consumer queueConsumer = new QueueTitleConsumer(queue);
Producer queueProducer = new QueueTitleProducer(queue);

// Eventually, we want this bit in the first thread...
Title title = producer.produce();
queueConsumer.consume(title);
// ...and this bit in the second thread.
Title queuedTitle = queueProducer.produce();
consumer.consume(title);

It will be useful to notice at this point we have two separate tasks that look very similar here. Each of these tasks will pull from a Producer and push to a Consumer, and the tasks doesn't really need to understand the specifics. This is a good place to introduce a Runnable.

PipelineTask implements Runnable {
    private final Producer producer;
    private final Consumer consumer;

    PipelineTask(Producer producer, Consumer consumer) {
        this.producer = producer;
        this.consumer = consumer;
    }

    public void run() {
        Title title = producer.produce();
        consumer.consume(title);
    }
}

We can then implement our trivial single threaded approach with a single pipeline

    PipelineTask task = new PipelineTask(producer, consumer);
    task.run();

Or our more complicated single threaded approach with two pipelines run in series.

    PipelineTask readTask = new PipelineTask(producer, queueConsumer);
    toQueue.run();

    PipelineTask writeTask = new PipelineTask(queueProducer, consumer);
    fromQueue.run();

Now that we have Runnables, putting them on separate threads is straight forward. Better still to delegate the thread management elsewhere -- using an ExecutorService hides the details while providing useful methods for interrupting the tasks that are running and checking to see when everything has finished.

startService(), in your example, had the responsibility for assembling the object graph -- with these new ideas, the code would look something like....

...
ObjectInputStream in = new ObjectInputStream(socket.getInputStream());
Producer fromClient = new StreamTitleProducer(in);

ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
Consumer toClient = new StreamTitleConsumer(out);

BlockingQueue<Title> queue = ...
Consumer toQueue = new QueueTitleConsumer(queue);
Producer fromQueue = new QueueTitleProducer(queue);

PipelineTask readTask = new PipelineTask(fromClient, toQueue);
PipelineTask writeTask = new PipelineTask(fromQueue, toClient);

ExecutorService executor ...
executor.submit(readTask);
executor.submit(writeTask);

In your implementation, the consumer logic was responsible for making an edit to the Title. A straightforward way to do this is with the Decorator pattern, using one Consumer to make the edit, and then pass the data to the next.

EditingTitleConsumer implements TitleConsumer {
    private final TitleConsumer next;

    EditingTitleConsumer(TitleConsumer next) {
        this.next = next;
    }

    public void consumer(Title title) {
        title.setState(State.x);
        next.consumer(title);
    }
}

I often find that decorators make it hard to come up with good names for instance variables; it's common to re-use the variable name. ... Consumer toClient = new StreamTitleConsumer(out); toClient = new EditingTitleConsumer(toClient);

It's also common that people will nest the calls to constructors

Consumer toClient = new EditingTitleConsumer(new StreamTitleConsumer(out));

Further notes: There's really nothing in this design specific to Title; the interface TitleProducer should really be Producer<Title>, likewise Consumer<Title> and PipelineTask<Title>.

It's usually the case that you'll want to produce(), or consume(), more than one object at a time. So the PipelineTask.run() is more likely to look like....

    public void run() {
        while(notDone()) {
            Title title = producer.produce();
            consumer.consume(title);
        }
    }

It might be, for instance, that the producer can detect when it has finished, in which case your producer starts to look a bit like an Iterator<T>

    public void run() {
        while(producer.hasNext()) {
            Title title = producer.next();
            consumer.consume(title);
        }
    }

Similarly, your consumer may need to know when it is finished (so that it can flush data, close streams, and so on).

    public void run() {
        while(producer.hasNext()) {
            Title title = producer.next();
            consumer.consume(title);
        }
        consumer.close();
    }

Shutting down the task that publishes to the queue is generally straightforward - there's no more data, so stop. Easy. Shutting down the task pulling from the queue is harder -- if the queue is empty, you want the consumers to idle if more data may be coming, but exit if no more data is coming... furthermore, the consumer may be blocking on the empty queue when you want to tell it that the job is finished. Find and read Java Concurrency in Practice; it has an important chapter on cancellation and shutdown.

Finally: for advanced readings on this topic, check out the LMAX Disruptor library.

\$\endgroup\$
  • \$\begingroup\$ I just wanted to mention that I keep coming back to this answer to re-read the solution(s) and techniques. \$\endgroup\$ – Thufir Jul 27 '14 at 9:01

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.