上一次我們學(xué)習(xí)了java.util.concurrent.Future<T>背后的原理 。 我們還發(fā)現(xiàn)， Future<T>通常由庫或框架返回。 但是沒有什么可以阻止我們在有意義的情況下自行實(shí)現(xiàn)所有功能。 它不是特別復(fù)雜，可以顯著改善您的設(shè)計(jì)。 我盡力為我們的示例選擇有趣的用例。

JMS（Java消息服務(wù)）是用于發(fā)送異步消息的標(biāo)準(zhǔn)Java API。 當(dāng)我們想到JMS時??，我們立即看到客戶端以一發(fā)不可收拾的方式向服務(wù)器（代理）發(fā)送消息。 但是在JMS之上實(shí)現(xiàn)請求-答復(fù)消息傳遞模式同樣普遍。 實(shí)現(xiàn)非常簡單：您將請求消息（當(dāng)然是異步地）發(fā)送到另一側(cè)的MDB。
MDB處理該請求，然后將答復(fù)發(fā)送回硬編碼的答復(fù)隊(duì)列或客戶機(jī)選擇的任意隊(duì)列，并與JMSReplyTo屬性中的消息一起發(fā)送。 第二種情況更有趣。 客戶端可以創(chuàng)建一個臨時隊(duì)列，并在發(fā)送請求時將其用作回復(fù)隊(duì)列。 這樣，每個請求/答復(fù)對使用不同的答復(fù)隊(duì)列，因此不需要關(guān)聯(lián)ID，選擇器等。

但是有一個問題。 向JMS代理發(fā)送消息是簡單且異步的。 但是，收到答復(fù)要麻煩得多。 您可以實(shí)現(xiàn)MessageListener以使用一條消息，也可以使用阻塞MessageConsumer.receive() 。 第一種方法非常笨重，很難在實(shí)踐中使用。 第二個失敗了異步消息傳遞的目的。 您還可以按一定的時間間隔輪詢答復(fù)隊(duì)列，這聽起來更糟。

到現(xiàn)在為止，了解Future抽象您應(yīng)該有一些設(shè)計(jì)想法。 如果我們可以發(fā)送請求消息并取回Future<T> （表示尚未收到的答復(fù)消息）怎么辦？ Future抽象應(yīng)該處理所有邏輯，我們可以放心地將其用作將來結(jié)果的句柄。 這是用于創(chuàng)建臨時隊(duì)列和發(fā)送請求的管道代碼：

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代理和任意數(shù)據(jù)。 該對象將使用ObjectMessage發(fā)送到queue 。 最后一行至關(guān)重要–我們返回自定義的JmsReplyFuture<T> ，它將表示尚未收到的回復(fù)。 注意，我們?nèi)绾螌⑴R時JMS隊(duì)列傳遞給JMSReplyTo屬性和Future 。 MDB方面的實(shí)現(xiàn)并不那么重要。 不用說是將回復(fù)發(fā)送回指定隊(duì)列：

final ObjectMessage reply = session.createObjectMessage(...); session.createProducer(request.getJMSReplyTo()).send(reply);

因此，讓我們深入研究JmsReplyFuture<T> 。 我假設(shè)請求和答復(fù)都是ObjectMessage 。 使用不同類型的消息不是很困難。 首先讓我們看看如何設(shè)置從回復(fù)通道接收消息：

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實(shí)現(xiàn)了Future<T> （其中T是包裝在ObjectMessage的對象的預(yù)期類型）和JMS MessageListener 。 在構(gòu)造函數(shù)中，我們只是開始偵聽replyQueue 。 根據(jù)我們的設(shè)計(jì)假設(shè)，我們知道那里最多會有一條消息，因?yàn)榛貜?fù)隊(duì)列是臨時丟棄隊(duì)列。 在上一篇文章中，我們了解到Future.get()應(yīng)該在等待結(jié)果時阻塞。 另一方面， onMessage()是從某些內(nèi)部JMS客戶端線程/庫調(diào)用的回調(diào)方法。 顯然，我們需要一些共享變量/鎖，以使等待中的get()知道答復(fù)已到達(dá)。 最好我們的解決方案應(yīng)該是輕量級的，并且不引入任何延遲，因此忙于等待volatile變量是一個壞主意。 最初，我雖然使用了Semaphore ，但我將使用它來取消阻塞onMessage() get() onMessage() 。 但是我仍然需要一些共享變量來保存實(shí)際的回復(fù)對象。 因此，我想到了使用ArrayBlockingQueue的想法。 當(dāng)我們知道不會再有一個項(xiàng)目時，使用隊(duì)列聽起來可能很奇怪。 但是它利用舊的生產(chǎn)者-消費(fèi)者模式很好地工作： Future.get()是一個消費(fèi)者，它阻塞了空隊(duì)列的poll()方法。 另一方面， onMessage()是生產(chǎn)者，將回復(fù)消息放置在該隊(duì)列中并立即取消阻塞消費(fèi)者。 外觀如下：

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);//...}}

實(shí)施仍未完成，但涵蓋了最重要的概念。 請注意， BlockingQueue.poll(long, TimeUnit)方法非常適合Future.get(long, TimeUnit) 。 不幸的是，即使它們來自相同的程序包并且在相同的時間內(nèi)或多或少地被開發(fā)，一種方法在超時時返回null ，而另一種方法應(yīng)引發(fā)異常。 易于修復(fù)。

還要注意onMessage()的實(shí)現(xiàn)變得多么容易。 我們只是將新收到的消息放在BlockingQueue reply ，而集合將為我們完成所有同步。 我們?nèi)匀蝗鄙僖恍┎惶匾匀恢匾募?xì)節(jié)–取消和清理。 無需贅述，下面是一個完整的實(shí)現(xiàn)：

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枚舉來保存有關(guān)狀態(tài)的信息。 與基于多個標(biāo)志， null檢查等的復(fù)雜條件相比，我發(fā)現(xiàn)它更具可讀性。要記住的第二件事是取消。 幸運(yùn)的是，它非常簡單。 我們基本上關(guān)閉了基礎(chǔ)會話/連接。 在整個請求/答復(fù)消息交換的整個過程中，它必須保持打開狀態(tài)，否則臨時JMS答復(fù)隊(duì)列將消失。 請注意，我們不能輕易通知經(jīng)紀(jì)人/ MDB我們對答復(fù)不再感興趣。 我們只是停止監(jiān)聽它，但是MDB仍將處理請求并嘗試將答復(fù)發(fā)送到不再存在的臨時隊(duì)列。

那么這一切在實(shí)踐中看起來如何？ 假設(shè)我們有一個MDB接收一個數(shù)字并返回一個平方。 假設(shè)計(jì)算需要一點(diǎn)時間，所以我們提前開始計(jì)算，同時做一些工作，然后再取回結(jié)果。 這樣的設(shè)計(jì)如下所示：

final Future<Double> replyFuture = asynchRequest(connectionFactory, 7, "square"); //do some more work final double resp = replyFuture.get(); //49

其中"square"是請求隊(duì)列的名稱。 如果我們重構(gòu)它并使用依賴注入，我們可以將其進(jìn)一步簡化為：

final Future<Double> replyFuture = calculator.square(7); //do some more work final double resp = replyFuture.get(); //49

您知道該設(shè)計(jì)的最佳選擇嗎？ 即使我們正在利用相當(dāng)先進(jìn)的JMS功能，此處也沒有JMS代碼。 此外，我們稍后可以使用SOAP或GPU將calculator替換為其他實(shí)現(xiàn)。 就客戶端代碼而言，我們?nèi)匀皇褂肍uture<Double>抽象。 尚未提供計(jì)算結(jié)果。 根本的機(jī)制是無關(guān)緊要的。 那就是抽象的美。

顯然，此實(shí)現(xiàn)尚未準(zhǔn)備好生產(chǎn)（到目前為止）。 但更糟糕的是，它缺少一些基本功能。 我們?nèi)匀辉谀硞€時候調(diào)用阻塞Future.get() 。 而且，無法組成/鏈接期貨（例如， 當(dāng)響應(yīng)到達(dá)時，發(fā)送另一條消息 ）或等待最快的期貨完成。 耐心一點(diǎn)！

參考：在NoBlogDefFound博客上，從我們的JCG合作伙伴 Tomasz Nurkiewicz 實(shí)現(xiàn)自定義Future 。

翻譯自: https://www.javacodegeeks.com/2013/02/implementing-custom-future.html

總結(jié)

以上是生活随笔為你收集整理的实现自定义的未来的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。