上一次我们学习了java.util.concurrent.Future<T>
背后的原理 。 我们还发现, Future<T>
通常由库或框架返回。 但是没有什么可以阻止我们在有意义的情况下自行实现所有功能。 它不是特别复杂,可以显着改善您的设计。 我尽力为我们的示例选择有趣的用例。
JMS(Java消息服务)是用于发送异步消息的标准Java API。 当我们想到JMS时,我们立即看到客户端以一发不可收拾的方式向服务器(代理)发送消息。 但是在JMS之上实现请求-答复消息传递模式同样普遍。 实现非常简单:您将请求消息(当然是异步地)发送到另一侧的MDB。
MDB处理该请求,然后将答复发送回硬编码的答复队列或客户机选择的任意队列,并与JMSReplyTo
属性中的消息一起发送。 第二种情况更有趣。 客户端可以创建一个临时队列,并在发送请求时将其用作回复队列。 这样,每个请求/答复对使用不同的答复队列,因此不需要关联ID,选择器等。
但是有一个问题。 向JMS代理发送消息是简单且异步的。 但是,收到答复要麻烦得多。 您可以实现MessageListener
以使用一条消息,也可以使用阻塞MessageConsumer.receive()
。 第一种方法非常笨重,很难在实践中使用。 第二个失败了异步消息传递的目的。 您还可以按一定的时间间隔轮询答复队列,这听起来更糟。
到现在为止,了解Future
抽象您应该有一些设计想法。 如果我们可以发送请求消息并取回Future<T>
(表示尚未收到的答复消息)怎么办? Future
抽象应该处理所有逻辑,我们可以放心地将其用作将来结果的句柄。 这是用于创建临时队列和发送请求的管道代码:
private <T extends Serializable> Future<T> asynchRequest(ConnectionFactory connectionFactory, Serializable request, String queue) throws JMSException {Connection connection = connectionFactory.createConnection();connection.start();final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);final Queue tempReplyQueue = session.createTemporaryQueue();final ObjectMessage requestMsg = session.createObjectMessage(request);requestMsg.setJMSReplyTo(tempReplyQueue);sendRequest(session.createQueue(queue), session, requestMsg);return new JmsReplyFuture<T>(connection, session, tempReplyQueue);
}
asynchRequest()
方法仅将ConnectionFactory
带到JMS代理和任意数据。 该对象将使用ObjectMessage
发送到queue
。 最后一行至关重要–我们返回自定义的JmsReplyFuture<T>
,它将表示尚未收到的回复。 注意,我们如何将临时JMS队列传递给JMSReplyTo
属性和Future
。 MDB方面的实现并不那么重要。 不用说是将回复发送回指定队列:
final ObjectMessage reply = session.createObjectMessage(...);
session.createProducer(request.getJMSReplyTo()).send(reply);
因此,让我们深入研究JmsReplyFuture<T>
。 我假设请求和答复都是ObjectMessage
。 使用不同类型的消息不是很困难。 首先让我们看看如何设置从回复通道接收消息:
public class JmsReplyFuture<T extends Serializable> implements Future<T>, MessageListener {//...public JmsReplyFuture(Connection connection, Session session, Queue replyQueue) throws JMSException {this.connection = connection;this.session = session;replyConsumer = session.createConsumer(replyQueue);replyConsumer.setMessageListener(this);}@Overridepublic void onMessage(Message message) {//...}}
如您所见, JmsReplyFuture
实现了Future<T>
(其中T
是包装在ObjectMessage
的对象的预期类型)和JMS MessageListener
。 在构造函数中,我们只是开始侦听replyQueue
。 根据我们的设计假设,我们知道那里最多会有一条消息,因为回复队列是临时丢弃队列。 在上一篇文章中,我们了解到Future.get()
应该在等待结果时阻塞。 另一方面, onMessage()
是从某些内部JMS客户端线程/库调用的回调方法。 显然,我们需要一些共享变量/锁,以使等待中的get()
知道答复已到达。 最好我们的解决方案应该是轻量级的,并且不引入任何延迟,因此忙于等待volatile
变量是一个坏主意。 最初,我虽然使用了Semaphore
,但我将使用它来取消阻塞onMessage()
get()
onMessage()
。 但是我仍然需要一些共享变量来保存实际的回复对象。 因此,我想到了使用ArrayBlockingQueue
的想法。 当我们知道不会再有一个项目时,使用队列听起来可能很奇怪。 但是它利用旧的生产者-消费者模式很好地工作: Future.get()
是一个消费者,它阻塞了空队列的poll()
方法。 另一方面, onMessage()
是生产者,将回复消息放置在该队列中并立即取消阻塞消费者。 外观如下:
public class JmsReplyFuture<T extends Serializable> implements Future<T>, MessageListener {private final BlockingQueue<T> reply = new ArrayBlockingQueue<>(1);//...@Overridepublic T get() throws InterruptedException, ExecutionException {return this.reply.take();}@Overridepublic T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {final T replyOrNull = reply.poll(timeout, unit);if (replyOrNull == null) {throw new TimeoutException();}return replyOrNull;}@Overridepublic void onMessage(Message message) {final ObjectMessage objectMessage = (ObjectMessage) message;final Serializable object = objectMessage.getObject();reply.put((T) object);//...}}
实施仍未完成,但涵盖了最重要的概念。 请注意, BlockingQueue.poll(long, TimeUnit)
方法非常适合Future.get(long, TimeUnit)
。 不幸的是,即使它们来自相同的程序包并且在相同的时间内或多或少地被开发,一种方法在超时时返回null
,而另一种方法应引发异常。 易于修复。
还要注意onMessage()
的实现变得多么容易。 我们只是将新收到的消息放在BlockingQueue reply
,而集合将为我们完成所有同步。 我们仍然缺少一些不太重要但仍然重要的细节–取消和清理。 无需赘述,下面是一个完整的实现:
public class JmsReplyFuture<T extends Serializable> implements Future<T>, MessageListener {private static enum State {WAITING, DONE, CANCELLED}private final Connection connection;private final Session session;private final MessageConsumer replyConsumer;private final BlockingQueue<T> reply = new ArrayBlockingQueue<>(1);private volatile State state = State.WAITING;public JmsReplyFuture(Connection connection, Session session, Queue replyQueue) throws JMSException {this.connection = connection;this.session = session;replyConsumer = session.createConsumer(replyQueue);replyConsumer.setMessageListener(this);}@Overridepublic boolean cancel(boolean mayInterruptIfRunning) {try {state = State.CANCELLED;cleanUp();return true;} catch (JMSException e) {throw Throwables.propagate(e);}}@Overridepublic boolean isCancelled() {return state == State.CANCELLED;}@Overridepublic boolean isDone() {return state == State.DONE;}@Overridepublic T get() throws InterruptedException, ExecutionException {return this.reply.take();}@Overridepublic T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {final T replyOrNull = reply.poll(timeout, unit);if (replyOrNull == null) {throw new TimeoutException();}return replyOrNull;}@Overridepublic void onMessage(Message message) {try {final ObjectMessage objectMessage = (ObjectMessage) message;final Serializable object = objectMessage.getObject();reply.put((T) object);state = State.DONE;cleanUp();} catch (Exception e) {throw Throwables.propagate(e);}}private void cleanUp() throws JMSException {replyConsumer.close();session.close();connection.close();}
}
我使用特殊的State
枚举来保存有关状态的信息。 与基于多个标志, null
检查等的复杂条件相比,我发现它更具可读性。要记住的第二件事是取消。 幸运的是,它非常简单。 我们基本上关闭了基础会话/连接。 在整个请求/答复消息交换的整个过程中,它必须保持打开状态,否则临时JMS答复队列将消失。 请注意,我们不能轻易通知经纪人/ MDB我们对答复不再感兴趣。 我们只是停止监听它,但是MDB仍将处理请求并尝试将答复发送到不再存在的临时队列。
那么这一切在实践中看起来如何? 假设我们有一个MDB接收一个数字并返回一个平方。 假设计算需要一点时间,所以我们提前开始计算,同时做一些工作,然后再取回结果。 这样的设计如下所示:
final Future<Double> replyFuture = asynchRequest(connectionFactory, 7, "square");
//do some more work
final double resp = replyFuture.get(); //49
其中"square"
是请求队列的名称。 如果我们重构它并使用依赖注入,我们可以将其进一步简化为:
final Future<Double> replyFuture = calculator.square(7);
//do some more work
final double resp = replyFuture.get(); //49
您知道该设计的最佳选择吗? 即使我们正在利用相当先进的JMS功能,此处也没有JMS代码。 此外,我们稍后可以使用SOAP或GPU将calculator
替换为其他实现。 就客户端代码而言,我们仍然使用Future<Double>
抽象。 尚未提供计算结果。 根本的机制是无关紧要的。 那就是抽象的美。
显然,此实现尚未准备好生产(到目前为止)。 但更糟糕的是,它缺少一些基本功能。 我们仍然在某个时候调用阻塞Future.get()
。 而且,无法组成/链接期货(例如, 当响应到达时,发送另一条消息 )或等待最快的期货完成。 耐心一点!
参考:在NoBlogDefFound博客上,从我们的JCG合作伙伴 Tomasz Nurkiewicz 实现自定义Future 。
翻译自: https://www.javacodegeeks.com/2013/02/implementing-custom-future.html