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
- How do we exchange data between a producer on one thread and a consumer on another thread?
- 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.
