昆山市住房和城乡建设网站,wordpress源码书籍,手工外包加工网,怎么自己做刷qq网站自定义注解 实现自定义消息上一次我们学习了java.util.concurrent.FutureT背后的原理 。 我们还发现#xff0c; FutureT通常由库或框架返回。 但是#xff0c;没有什么可以阻止我们在有意义的情况下自行实现所有功能。 它不是特别复杂#xff0c;可以显着改… 自定义注解 实现自定义消息 上一次我们学习了java.util.concurrent.FutureT背后的原理 。 我们还发现 FutureT通常由库或框架返回。 但是没有什么可以阻止我们在有意义的情况下自行实现所有功能。 它不是特别复杂可以显着改善您的设计。 我尽力为我们的示例选择有趣的用例。 JMSJava消息服务是用于发送异步消息的标准Java API。 当我们想到JMS时我们立即看到客户端以一发不可收拾的方式向服务器经纪人发送消息。 但是在JMS之上实现请求-答复消息传递模式同样普遍。 实现非常简单您将请求消息当然是异步地发送到另一侧的MDB。 MDB处理该请求然后将答复发送回硬编码的答复队列或客户机选择的任意队列并与JMSReplyTo属性中的消息一起发送。 第二种情况更有趣。 客户端可以创建一个临时队列并在发送请求时将其用作回复队列。 这样每个请求/答复对使用不同的答复队列因此不需要关联ID选择器等。 但是有一个问题。 向JMS代理发送消息是简单且异步的。 但是收到答复要麻烦得多。 您可以实现MessageListener以使用一条消息也可以使用阻塞MessageConsumer.receive() 。 第一种方法非常重很难在实践中使用。 第二个失败了异步消息传递的目的。 您还可以按一定间隔轮询回复队列这听起来更糟。 到现在为止了解Future抽象您应该有一些设计想法。 如果我们可以发送请求消息并取回FutureT 代表尚未发送的答复消息怎么办 Future抽象应该处理所有逻辑我们可以放心地将其用作未来结果的句柄。 这是用于创建临时队列和发送请求的管道代码 private T extends Serializable FutureT 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 JmsReplyFutureT(connection, session, tempReplyQueue);
} asynchRequest()方法仅将ConnectionFactory带到JMS代理和任意数据。 该对象将使用ObjectMessage发送到queue 。 最后一行至关重要–我们返回自定义的JmsReplyFutureT 它将表示尚未收到的回复。 注意我们如何将临时JMS队列传递给JMSReplyTo属性和Future 。 MDB方面的实现并不那么重要。 不用说是将回复发送回指定队列 final ObjectMessage reply session.createObjectMessage(...);
session.createProducer(request.getJMSReplyTo()).send(reply); 因此让我们深入研究JmsReplyFutureT 。 我假设请求和答复都是ObjectMessage 。 使用不同类型的消息不是很困难。 首先让我们看看如何设置从回复通道接收消息 public class JmsReplyFutureT extends Serializable implements FutureT, 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实现了FutureT 其中T是包装在ObjectMessage的对象的预期类型和JMS MessageListener 。 在构造函数中我们只是开始侦听replyQueue 。 根据我们的设计假设我们知道那里最多会有一条消息因为回复队列是临时丢弃队列。 在上一篇文章中我们了解到Future.get()应该在等待结果时阻塞。 另一方面 onMessage()是从某些内部JMS客户端线程/库调用的回调方法。 显然我们需要一些共享变量/锁以使等待中的get()知道答复已到达。 最好我们的解决方案应该是轻量级的并且不引入任何延迟因此忙于等待volatile变量是一个坏主意。 最初我虽然使用了Semaphore 但我将使用它来从onMessage()取消阻塞get() onMessage() 。 但是我仍然需要一些共享变量来保存实际的回复对象。 因此我想到了使用ArrayBlockingQueue的想法。 当我们知道不会再有一个项目时使用队列听起来可能很奇怪。 但是它利用旧的生产者-消费者模式很好地工作 Future.get()是一个消费者它阻塞了空队列的poll()方法。 另一方面 onMessage()是生产者将回复消息放入该队列中并立即取消阻塞消费者。 外观如下 public class JmsReplyFutureT extends Serializable implements FutureT, MessageListener {private final BlockingQueueT 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 JmsReplyFutureT extends Serializable implements FutureT, MessageListener {private static enum State {WAITING, DONE, CANCELLED}private final Connection connection;private final Session session;private final MessageConsumer replyConsumer;private final BlockingQueueT 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 FutureDouble replyFuture asynchRequest(connectionFactory, 7, square);
//do some more work
final double resp replyFuture.get(); //49 其中square是请求队列的名称。 如果我们重构它并使用依赖注入我们可以将其进一步简化为 final FutureDouble replyFuture calculator.square(7);
//do some more work
final double resp replyFuture.get(); //49 您知道该设计的最佳选择吗 即使我们正在利用相当先进的JMS功能此处也没有JMS代码。 此外我们稍后可以使用SOAP或GPU将calculator替换为其他实现。 就客户端代码而言我们仍然使用FutureDouble抽象。 尚未提供计算结果。 根本的机制无关紧要。 那就是抽象之美。 显然此实现尚未准备好生产到目前为止。 但更糟糕的是它缺少一些基本功能。 我们仍然在某个时候调用阻塞Future.get() 。 而且无法组成/链接期货例如 当响应到达时发送另一条消息 或等待最快的期货完成。 耐心一点 参考 NoBlogDefFound博客中的JCG合作伙伴 Tomasz Nurkiewicz 实现了自定义Future 。 翻译自: https://www.javacodegeeks.com/2013/02/implementing-custom-future.html自定义注解 实现自定义消息
