RabbitMQ 進階教程（四）：RPC 遠程過程調用模式實戰

RPC 模式核心思想

客戶端發送請求 → 服務端處理 → 返回響應。RabbitMQ 通過 Callback Queue + Correlation ID 實現異步請求-響應匹配。

關鍵消息屬性

架構流程

1. 客戶端啟動 → 聲明獨占匿名回調隊列（隨機名、連接關閉自動刪除）。
2. 發送請求到 rpc_queue，附帶 replyTo（回調隊列）+ correlationId（UUID）。
3. 服務端消費 rpc_queue，處理業務，發送結果到 replyTo 隊列，攜帶相同 correlationId。
4. 客戶端監聽回調隊列，收到消息比對 correlationId，匹配則返回調用者。

服務端代碼

public class RPCServer {
    private static final String RPC_QUEUE = "rpc_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {
            ch.queueDeclare(RPC_QUEUE, false, false, false, null);
            ch.basicQos(1); // 公平分發，每次只處理一條
            QueueingConsumer consumer = new QueueingConsumer(ch);
            ch.basicConsume(RPC_QUEUE, false, consumer);
            System.out.println("RPC Server awaiting requests...");

            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                AMQP.BasicProperties props = delivery.getProperties();
                AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
                    .correlationId(props.getCorrelationId()).build();

                String msg = new String(delivery.getBody(), "UTF-8");
                int n = Integer.parseInt(msg);
                System.out.println("Computing fib(" + n + ")");
                String response = String.valueOf(fib(n));

                ch.basicPublish("", props.getReplyTo(), replyProps, response.getBytes());
                ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

    private static int fib(int n) {
        return (n == 0 || n == 1) ? n : fib(n-1) + fib(n-2);
    }
}

客戶端代碼

public class RPCClient {
    private final Connection conn;
    private final Channel ch;
    private final String replyQueue;
    private final QueueingConsumer consumer;

    public RPCClient() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        conn = factory.newConnection();
        ch = conn.createChannel();
        replyQueue = ch.queueDeclare().getQueue(); // 匿名獨占隊列
        consumer = new QueueingConsumer(ch);
        ch.basicConsume(replyQueue, true, consumer);
    }

    public String call(String msg) throws Exception {
        String corrId = UUID.randomUUID().toString();
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .correlationId(corrId).replyTo(replyQueue).build();
        ch.basicPublish("", "rpc_queue", props, msg.getBytes("UTF-8"));

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                return new String(delivery.getBody(), "UTF-8");
            }
        }
    }

    public void close() throws Exception { conn.close(); }

    public static void main(String[] args) throws Exception {
        try (RPCClient client = new RPCClient()) {
            System.out.println("Requesting fib(20)");
            String resp = client.call("20");
            System.out.println("Got: " + resp);
        }
    }
}

關鍵點解析

• Correlation ID 去重：服務端重啟可能導致重複響應，客戶端必須忽略未知/重複 ID。
• basicQos(1)：防止單個消費者積壓，實現負載均衡。
• 匿名回調隊列：每客戶端一個，隔離響應，自動清理。
• 擴展性：可啟動多個 RPCServer 實例，隊列自動分發。

1. 上一篇文章： JVM 方法區與永久代：HotSpot 實現細節
2. 下一篇文章： VisualVM 遠程監控實戰：基於 jstatd 的 JVM 性能分析

Java 綜合教學

Java AWT 動畫進階：下雪特效實戰（從亂飛到有序飄落的修正過程）

版本一：錯誤示範 —— 每幀重新隨機 @Override p

SSH 框架整合 POI 導出複雜 Excel：多 Sheet、合併儲存格、樣式與下載全流程

企業級報表導出實戰：從數據庫查詢到瀏覽器下載的完整鏈路 本文

Eclipse/MyEclipse「Java compiler level does not match」一鍵修復指南

錯誤現象 導入項目後，Problems 視圖紅報： Jav

JBoss / WildFly 版本演進全景圖與 EAP 5.1 生產級安裝部署實戰指南

JBoss 家族版本演進脈絡 系列關鍵版本里