码上敲享录 > RabbitMQ教程(java) > Rabbitmq远程过程调用(RPC)模式

Rabbitmq远程过程调用(RPC)模式

上一章章节目录下一章 2019-12-20已有1551人阅读 评论(0)

Rabbitmq远程过程调用(RPC)模式

目录

idea创建java项目整合rabbitmq

Rabbitmq简单模式

Rabbitmq工作队列模式(任务队列模式)

Rabbitmq发布/订阅模式

Rabbitmq直接(direct)路由模式

Rabbitmq主题(topic)路由模式

Rabbitmq发布者确认模式

Rabbitmq远程过程调用(RPC)模式


前面的几种模式都是生产者发布了消息到达队列之后,这些消息如何被处理就不管了,现在RPC模式就是客户端发送RPC请求并进入阻塞状态,直到收到服务端处理这些消息之后并返回答案为止,在远程计算机上运行功能并等待结果这种模式通常称为“ 远程过程调用”或“ RPC”。


回调队列:客户端发送请求消息到服务器使用了一个队列,然而服务器要向客户端发送响应消息也需要一个队列。所以我们需要客户端要使用replyTo方法设置一个临时队列给服务端使用。


relatedId(关联ID):将为每个请求将其设置为唯一值,当我们在客户端收到响应的消息时,响应的relatedId与请求的relatedId进行比较。如果不相等值,我们可以放心地丢弃该消息-它不属于我们的请求。

例如下面代码中的if (delivery.getProperties().getCorrelationId().equals(corrId))


开始演示:

第一步、执行服务端RPCServer.java,监听客户端RPC请求。

第二步、执行客户端RPCClient.java,发送5个RPC请求,并阻塞等待服务端响应返回数据。


客户端打印的结果:

客户端开始发送消息:0

远程服务端响应的内容:本服务已处理的消息【0】

---------------------------------------------------------------------------

客户端开始发送消息:1

远程服务端响应的内容:本服务已处理的消息【1】

---------------------------------------------------------------------------

客户端开始发送消息:2

远程服务端响应的内容:本服务已处理的消息【2】

---------------------------------------------------------------------------

客户端开始发送消息:3

远程服务端响应的内容:本服务已处理的消息【3】

---------------------------------------------------------------------------

客户端开始发送消息:4

远程服务端响应的内容:本服务已处理的消息【4】


服务端打印的结果:

服务器启动成功,正等待客户端RPC请求......

服务端收到的RPC请求:0

服务端收到的RPC请求:1

服务端收到的RPC请求:2

服务端收到的RPC请求:3

服务端收到的RPC请求:4


演示结论:

客户端发送RPC请求并进入阻塞状态,直到收到服务端处理这些消息之后并返回答案为止


客户端代码:

package com.demo;

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("127.0.0.1");

       factory.setPort(5672);

       factory.setUsername("guest");

       factory.setPassword("guest");

       connection = factory.newConnection();

       channel = connection.createChannel();

   }

   public static void main(String[] argv) {

       try (RPCClient fibonacciRpc = new RPCClient()) {

           for (int i = 0; i < 5; i++) {

               String i_str = Integer.toString(i);

               System.out.println("客户端开始发送消息:" + i_str );

               String response = fibonacciRpc.call(i_str);

               System.out.println("远程服务端响应的内容:" + response);

               System.out.println("---------------------------------------------------------------------------");

           }

       } 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<String> 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();

   }

}


服务端代码:

package com.demo;

import com.rabbitmq.client.*;

public class RPCServer {

    private static final String RPC_QUEUE_NAME = "rpc_queue";

   public static void main(String[] argv) throws Exception {

       ConnectionFactory factory = new ConnectionFactory();

       factory.setHost("127.0.0.1");

       factory.setPort(5672);

       factory.setUsername("guest");

       factory.setPassword("guest");

       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("服务器启动成功,正等待客户端RPC请求......");

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

                   System.out.println("服务端收到的RPC请求:" + message);

                   response += "本服务已处理的消息【"+message+"】";

               } catch (RuntimeException e) {

                  e.printStackTrace();

               } finally {

                   channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));

                   channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

                   synchronized (monitor) {

                       monitor.notify();

                   }

               }

           };

           channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, (consumerTag -> { }));

           while (true) {

               synchronized (monitor) {

                   try {

                       monitor.wait();

                   } catch (InterruptedException e) {

                       e.printStackTrace();

                   }

               }

           }

       }

   }

}


向大家推荐《Activiti工作流实战教程》:https://xiaozhuanlan.com/activiti
0

有建议,请留言!

  • *您的姓名:

  • *所在城市:

  • *您的联系电话:

    *您的QQ:

  • 咨询问题:

  • 提 交