摘要:如果涉及返回值,就要用到本章提到的了。方法發送請求,并阻塞知道結果返回。當有消息時,進行計算并通過指定的發送給客戶端。當接收到,則檢查。如果和之前的匹配,則將消息返回給應用進行處理。
RPC模式
在第二章中我們學習了如何使用Work模式在多個worker之間派發時間敏感的任務。這種情況是不涉及到返回值的,worker執行任務就好。如果涉及返回值,就要用到本章提到的RPC(Remote Procedure Call)了。
本章我們使用RabbitMQ來構建一個RPC系統:一個客戶端和一個可擴展的RPC服務端。我們讓RPC服務返回一個斐波那契數組。
Client interface我們創建一個簡單的客戶端類來演示如何使用RPC服務。call方法發送RPC請求,并阻塞知道結果返回。
FibonacciRpcClient fibonacciRpc = new FibonacciRpcClient(); String result = fibonacciRpc.call("4"); System.out.println( "fib(4) is " + result);
Callback queueRPC貼士
雖然RPC的使用在計算機領域非常普遍,但是卻經常受到批評。主要問題是編碼人如果不注意使用的方法是本地還是遠程時,往往會造成問題。往往讓系統變得不可預知,增加不必要的復雜性和調試的難度。對此我們有如下幾點建議:是本地方法還是遠程方法要一目了然
把系統的依賴寫進文檔
系統要處理好超時的問題
如果可以盡量使用異步的pipeline來替代像RPC這種阻塞的操作。
在RabbitMQ上實現RPC是非常簡單的。客戶端發送一個request message,服務端回應一個response message。為了接受response message我們需要在發送request message的時候附帶上"callback" queue的地址。我們可以使用默認的queue。
callbackQueueName = channel.queueDeclare().getQueue(); BasicProperties props = new BasicProperties .Builder() .replyTo(callbackQueueName) .build(); channel.basicPublish("", "rpc_queue", props, message.getBytes()); // ... then code to read a response message from the callback_queue ...
Message的屬性
AMQP 0-9-1協議預定義了14個消息屬性,其中大部分很少使用,下面的屬性較為常用deliverMode: 標記message為持久(設置為2)或其他值。
contentType:message的編碼類型,我們經常使用JSON編碼,則設置為application/json
replyTo: 命名回調queue
correlationId:將RPC的請求和回應關聯起來
需要引入新的類
import com.rabbitmq.client.AMQP.BasicProperties;Correlaton Id
在上面的代碼中,每次RPC請求都會創建一個用于回調的臨時queue,我們有更好的方法,我們為每一個client創建一個回調queue。
但是這樣有新的問題,從回調queue中收到response無法和相應的request關聯起來。這時候就是correlationId屬性發揮作用的時候了。為每個request中設置唯一的值,在稍后的回調queue中收到的response里也有這個屬性,基于此,我們就可以關聯之前的request了。如果我們遇到一個匹配不到的correlationId,那么丟棄的行為是安全的。
你可能會問,為什么我們忽略這些無法匹配的message,而不是當做一個錯誤處理呢?主要是考慮服務端的競態條件,如果RPC服務器在發送response之后就宕機了,但是卻沒有發送ack消息。那么當RPC Server重啟之后,會繼續執行這個request。這就是為什么client需要冪等處理response。
Summary
我們的RPC向下面這樣進行工作:
對于一個RPC request,客戶端發送message時設置兩個屬性:replyTo設置成一個沒有名字的request獨有的queue;為每個request設置一個唯一的correlationId。
request發送到rpc_queue
RPC worker監聽rpc_queue。當有消息時,進行計算并通過replyTo指定的queue發送message給客戶端。
客戶端監聽回調queue。當接收到message,則檢查correlationId。如果和之前的request匹配,則將消息返回給應用進行處理。
開始執行斐波那契處理函數
private static int fib(int n) { if (n == 0) return 0; if (n == 1) return 1; return fib(n-1) + fib(n-2); }
這是一個簡易的實現,如果傳入一個較大的值,將會是個災難。
RPC服務器的代碼為RPCServer.java, 代碼是很簡單明確的
先是建立connection,channel和聲明queue.
設置prefetchCount,我們基于請求頻繁程度,會啟動多個RPC Server
使用basicConsume來接收,該方法提供回調參數設置(DeliverCallback).
RPC客戶端的代碼為RPCClient.java,代碼略微有點復雜
建立connection和channel。
call方法來發送RPC請求
生成correlationId
生成默認名字的queue用于reply,并訂閱它
發送request message,設置參數replyTo和correlationId.
然后返回并開始等待response到達
因為消費者發送response是在另一個線程中,我們需要讓main線程阻塞,在這里我們使用BlockingQueue。
消費者進行簡單的處理,為每一個response message檢查其correlationId,如果是,則將response添加進阻塞隊列
main函數阻塞在BlockingQueue返回
將response返回給用戶
RPCClient.java完整代碼
import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.UUID; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeoutException; public class RPCClient implements AutoCloseable { private Connection connection; private Channel channel; private String requestQueueName = "rpc_queue"; public RPCClient() throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); connection = factory.newConnection(); channel = connection.createChannel(); } public static void main(String[] argv) { try (RPCClient fibonacciRpc = new RPCClient()) { for (int i = 0; i < 32; i++) { String i_str = Integer.toString(i); System.out.println(" [x] Requesting fib(" + i_str + ")"); String response = fibonacciRpc.call(i_str); System.out.println(" [.] Got "" + response + """); } } catch (IOException | TimeoutException | InterruptedException e) { e.printStackTrace(); } } public String call(String message) throws IOException, InterruptedException { final String corrId = UUID.randomUUID().toString(); String replyQueueName = channel.queueDeclare().getQueue(); AMQP.BasicProperties props = new AMQP.BasicProperties .Builder() .correlationId(corrId) .replyTo(replyQueueName) .build(); channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8")); final BlockingQueueresponse = new ArrayBlockingQueue<>(1); String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> { if (delivery.getProperties().getCorrelationId().equals(corrId)) { response.offer(new String(delivery.getBody(), "UTF-8")); } }, consumerTag -> { }); String result = response.take(); channel.basicCancel(ctag); return result; } public void close() throws IOException { connection.close(); } }
RPCServer.java完整代碼
import com.rabbitmq.client.*; public class RPCServer { private static final String RPC_QUEUE_NAME = "rpc_queue"; private static int fib(int n) { if (n == 0) return 0; if (n == 1) return 1; return fib(n - 1) + fib(n - 2); } public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) { channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null); channel.queuePurge(RPC_QUEUE_NAME); channel.basicQos(1); System.out.println(" [x] Awaiting RPC requests"); Object monitor = new Object(); DeliverCallback deliverCallback = (consumerTag, delivery) -> { AMQP.BasicProperties replyProps = new AMQP.BasicProperties .Builder() .correlationId(delivery.getProperties().getCorrelationId()) .build(); String response = ""; try { String message = new String(delivery.getBody(), "UTF-8"); int n = Integer.parseInt(message); System.out.println(" [.] fib(" + message + ")"); response += fib(n); } catch (RuntimeException e) { System.out.println(" [.] " + e.toString()); } finally { channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8")); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // RabbitMq consumer worker thread notifies the RPC server owner thread synchronized (monitor) { monitor.notify(); } } }; channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, (consumerTag -> { })); // Wait and be prepared to consume the message from RPC client. while (true) { synchronized (monitor) { try { monitor.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } } } } }
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/73913.html
摘要:有助于將響應與請求關聯起來。如果發生這種情況,重新啟動的服務器將再次處理請求。又名服務器正在等待該隊列上的請求。當消息出現時,它檢查屬性。然后,我們進入循環,在其中等待請求消息,完成工作并發送響應。 (using php-amqplib) 前提必讀 本教程假設RabbitMQ是安裝在標準端口上運行(5672)。如果您使用不同的主機、端口或憑據,則連接設置需要調整。 如果您在本教程中遇到...
摘要:因為消費消息是在另外一個進程中,我們需要阻塞我們的進程直到結果返回,使用阻塞隊列是一種非常好的方式,這里我們使用了長度為的,的功能是檢查消息的的是不是我們之前所發送的,如果是,將返回值返回到。 推廣 RabbitMQ專題講座 https://segmentfault.com/l/15... CoolMQ開源項目 我們利用消息隊列實現了分布式事務的最終一致性解決方案,請大家圍觀。可以參考...
摘要:主題模式在上一章我們改進了我們的日志系統,如果使用我們只能簡單進行廣播,而使用則允許消費者可以進行一定程度的選擇。為的會同時發布到這兩個。當為時,會接收所有的。當中沒有使用通配符和時,的行為和一致。 主題模式 在上一章我們改進了我們的日志系統,如果使用fanout我們只能簡單進行廣播,而使用direct則允許消費者可以進行一定程度的選擇。但是direct還是有其局限性,其路由不支持多個...
摘要:路由模式在之前的文章中我們建立了一個簡單的日志系統。更形象的表示,如對中的感興趣。為了進行說明,像下圖這么來設置如圖,可以看到有兩個綁到了類型為的上。如圖的設置中,一個為的就會同時發送到和。接收程序可以選擇要接收日志的嚴重性級別。 路由模式 在之前的文章中我們建立了一個簡單的日志系統。我們可以通過這個系統將日志message廣播給很多接收者。 在本篇文章中,我們在這之上,添加一個新的功...
摘要:每個消費者會得到平均數量的。為了確保不會丟失,采用確認機制。如果中斷退出了關閉了,關閉了,或是連接丟失了而沒有發送,會認為該消息沒有完整的執行,會將該消息重新入隊。該消息會被發送給其他的。當消費者中斷退出,會重新分派。 Work模式 原文地址showImg(https://segmentfault.com/img/bVbqlXr?w=694&h=252); 在第一章中,我們寫了通過一個...
閱讀 1019·2022-07-19 10:19
閱讀 1794·2021-09-02 15:15
閱讀 1007·2019-08-30 15:53
閱讀 2653·2019-08-30 13:45
閱讀 2651·2019-08-26 13:57
閱讀 1983·2019-08-26 12:13
閱讀 1006·2019-08-26 10:55
閱讀 545·2019-08-26 10:46