国产xxxx99真实实拍_久久不雅视频_高清韩国a级特黄毛片_嗯老师别我我受不了了小说

資訊專欄INFORMATION COLUMN

【譯】RabbitMQ系列(六)-RPC模式

894974231 / 964人閱讀

摘要:如果涉及返回值,就要用到本章提到的了。方法發送請求,并阻塞知道結果返回。當有消息時,進行計算并通過指定的發送給客戶端。當接收到,則檢查。如果和之前的匹配,則將消息返回給應用進行處理。

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);

RPC貼士
雖然RPC的使用在計算機領域非常普遍,但是卻經常受到批評。主要問題是編碼人如果不注意使用的方法是本地還是遠程時,往往會造成問題。往往讓系統變得不可預知,增加不必要的復雜性和調試的難度。對此我們有如下幾點建議:

是本地方法還是遠程方法要一目了然

把系統的依賴寫進文檔

系統要處理好超時的問題

如果可以盡量使用異步的pipeline來替代像RPC這種阻塞的操作。

Callback queue

在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 BlockingQueue response = 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

相關文章

  • RabbitMQ+PHP 教程RPC

    摘要:有助于將響應與請求關聯起來。如果發生這種情況,重新啟動的服務器將再次處理請求。又名服務器正在等待該隊列上的請求。當消息出現時,它檢查屬性。然后,我們進入循環,在其中等待請求消息,完成工作并發送響應。 (using php-amqplib) 前提必讀 本教程假設RabbitMQ是安裝在標準端口上運行(5672)。如果您使用不同的主機、端口或憑據,則連接設置需要調整。 如果您在本教程中遇到...

    anquan 評論0 收藏0
  • 白話RabbitMQ(): RPC

    摘要:因為消費消息是在另外一個進程中,我們需要阻塞我們的進程直到結果返回,使用阻塞隊列是一種非常好的方式,這里我們使用了長度為的,的功能是檢查消息的的是不是我們之前所發送的,如果是,將返回值返回到。 推廣 RabbitMQ專題講座 https://segmentfault.com/l/15... CoolMQ開源項目 我們利用消息隊列實現了分布式事務的最終一致性解決方案,請大家圍觀。可以參考...

    KevinYan 評論0 收藏0
  • RabbitMQ系列(五) - 主題模式

    摘要:主題模式在上一章我們改進了我們的日志系統,如果使用我們只能簡單進行廣播,而使用則允許消費者可以進行一定程度的選擇。為的會同時發布到這兩個。當為時,會接收所有的。當中沒有使用通配符和時,的行為和一致。 主題模式 在上一章我們改進了我們的日志系統,如果使用fanout我們只能簡單進行廣播,而使用direct則允許消費者可以進行一定程度的選擇。但是direct還是有其局限性,其路由不支持多個...

    pingan8787 評論0 收藏0
  • RabbitMQ系列(四) - 路由模式

    摘要:路由模式在之前的文章中我們建立了一個簡單的日志系統。更形象的表示,如對中的感興趣。為了進行說明,像下圖這么來設置如圖,可以看到有兩個綁到了類型為的上。如圖的設置中,一個為的就會同時發送到和。接收程序可以選擇要接收日志的嚴重性級別。 路由模式 在之前的文章中我們建立了一個簡單的日志系統。我們可以通過這個系統將日志message廣播給很多接收者。 在本篇文章中,我們在這之上,添加一個新的功...

    liuchengxu 評論0 收藏0
  • RabbitMQ系列(二)-Work模式

    摘要:每個消費者會得到平均數量的。為了確保不會丟失,采用確認機制。如果中斷退出了關閉了,關閉了,或是連接丟失了而沒有發送,會認為該消息沒有完整的執行,會將該消息重新入隊。該消息會被發送給其他的。當消費者中斷退出,會重新分派。 Work模式 原文地址showImg(https://segmentfault.com/img/bVbqlXr?w=694&h=252); 在第一章中,我們寫了通過一個...

    lcodecorex 評論0 收藏0

發表評論

0條評論

894974231

|高級講師

TA的文章

閱讀更多
最新活動
閱讀需要支付1元查看
<