自定義注解 實(shí)現(xiàn)自定義消息

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

JMS（Java消息服務(wù)）是用于發(fā)送異步消息的標(biāo)準(zhǔn)Java API。 當(dāng)我們想到JMS時(shí)??，我們立即看到客戶端以一發(fā)不可收拾的方式向服務(wù)器（經(jīng)紀(jì)人）發(fā)送消息。 但是在JMS之上實(shí)現(xiàn)請(qǐng)求-答復(fù)消息傳遞模式同樣普遍。 實(shí)現(xiàn)非常簡(jiǎn)單：您將請(qǐng)求消息（當(dāng)然是異步地）發(fā)送到另一側(cè)的MDB。

MDB處理該請(qǐng)求，然后將答復(fù)發(fā)送回硬編碼的答復(fù)隊(duì)列或客戶機(jī)選擇的任意隊(duì)列，并與JMSReplyTo屬性中的消息一起發(fā)送。 第二種情況更有趣。 客戶端可以創(chuàng)建一個(gè)臨時(shí)隊(duì)列，并在發(fā)送請(qǐng)求時(shí)將其用作回復(fù)隊(duì)列。 這樣，每個(gè)請(qǐng)求/答復(fù)對(duì)使用不同的答復(fù)隊(duì)列，因此不需要關(guān)聯(lián)ID，選擇器等。

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

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

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ù)。 該對(duì)象將使用ObjectMessage發(fā)送到queue 。 最后一行至關(guān)重要–我們返回自定義的JmsReplyFuture<T> ，它將表示尚未收到的回復(fù)。 注意我們?nèi)绾螌⑴R時(shí)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è)請(qǐng)求和答復(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的對(duì)象的預(yù)期類型）和JMS MessageListener 。 在構(gòu)造函數(shù)中，我們只是開始偵聽replyQueue 。 根據(jù)我們的設(shè)計(jì)假設(shè)，我們知道那里最多會(huì)有一條消息，因?yàn)榛貜?fù)隊(duì)列是臨時(shí)丟棄隊(duì)列。 在上一篇文章中，我們了解到Future.get()應(yīng)該在等待結(jié)果時(shí)阻塞。 另一方面， onMessage()是從某些內(nèi)部JMS客戶端線程/庫調(diào)用的回調(diào)方法。 顯然，我們需要一些共享變量/鎖，以使等待中的get()知道答復(fù)已到達(dá)。 最好我們的解決方案應(yīng)該是輕量級(jí)的，并且不引入任何延遲，因此忙于等待volatile變量是一個(gè)壞主意。 最初，我雖然使用了Semaphore ，但我將使用它來從onMessage()取消阻塞get() onMessage() 。 但是我仍然需要一些共享變量來保存實(shí)際的回復(fù)對(duì)象。 因此，我想到了使用ArrayBlockingQueue的想法。 當(dāng)我們知道不會(huì)再有一個(gè)項(xiàng)目時(shí)，使用隊(duì)列聽起來可能很奇怪。 但是，它利用舊的生產(chǎn)者-消費(fèi)者模式很好地工作： Future.get()是一個(gè)消費(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í)施仍未完成，但涵蓋了最重要的概念。 請(qǐng)注意， BlockingQueue.poll(long, TimeUnit)方法非常適合Future.get(long, TimeUnit) 。 不幸的是，即使它們來自相同的程序包并且在相同的時(shí)間內(nèi)或多或少地被開發(fā)，一種方法在超時(shí)時(shí)返回null ，而另一種方法應(yīng)引發(fā)異常。 易于修復(fù)。

還要注意onMessage()的實(shí)現(xiàn)變得多么容易。 我們只是將新收到的消息放在BlockingQueue reply ，而集合將為我們完成所有同步。 我們?nèi)匀蝗鄙僖恍┎惶匾匀恢匾募?xì)節(jié)–取消和清理。 無需贅述，下面是一個(gè)完整的實(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)的信息。 與基于多個(gè)標(biāo)志， null檢查等的復(fù)雜條件相比，我發(fā)現(xiàn)它更具可讀性。要記住的第二件事是取消。 幸運(yùn)的是，它非常簡(jiǎn)單。 我們基本上關(guān)閉了基礎(chǔ)會(huì)話/連接。 在整個(gè)請(qǐng)求/答復(fù)消息交換的整個(gè)過程中，它必須保持打開狀態(tài)，否則臨時(shí)JMS答復(fù)隊(duì)列將消失。 請(qǐng)注意，我們不能輕易通知經(jīng)紀(jì)人/ MDB我們對(duì)答復(fù)不再感興趣。 我們只是停止監(jiān)聽它，但是MDB仍將處理請(qǐng)求并嘗試將答復(fù)發(fā)送到不再存在的臨時(shí)隊(duì)列。

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

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

其中"square"是請(qǐng)求隊(duì)列的名稱。 如果我們重構(gòu)它并使用依賴注入，我們可以將其進(jìn)一步簡(jiǎ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)匀辉谀硞€(gè)時(shí)候調(diào)用阻塞Future.get() 。 而且，無法組成/鏈接期貨（例如， 當(dāng)響應(yīng)到達(dá)時(shí)，發(fā)送另一條消息 ）或等待最快的期貨完成。 耐心一點(diǎn)！

參考： NoBlogDefFound博客中的JCG合作伙伴 Tomasz Nurkiewicz 實(shí)現(xiàn)了自定義Future 。

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

自定義注解 實(shí)現(xiàn)自定義消息

創(chuàng)作挑戰(zhàn)賽新人創(chuàng)作獎(jiǎng)勵(lì)來咯，堅(jiān)持創(chuàng)作打卡瓜分現(xiàn)金大獎(jiǎng)

總結(jié)

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

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