本文共 22255 字，大约阅读时间需要 74 分钟。

最好的教程官方文档。

RabbitMQ 概述

RabbitMQ是一个消息代理，负责接收和转发消息。

你可以把它想象成一个邮局:当你把你想要邮寄的邮件放入一个邮箱时，你可以确定邮递员先生或女士最终会把邮件送到你的收件人那里。

RabbitMQ和邮局的主要区别在于它不处理纸张，而是接受、存储和转发二进制数据消息。

核心概念

1. server：又称broker，接受客户端连接，实现AMQP实体服务。
2. connection：连接和具体broker网络连接。
3. channel：网络信道，几乎所有操作都在channel中进行，channel是消息读写的通道。客户端可以建立多个channel，每个channel表示一个会话任务。
4. message：消息，服务器和应用程序之间传递的数据，由properties和body组成。properties可以对消息进行修饰，比如消息的路由键，交换区，队列，优先级，延迟，delivery-mode等高级特性；body是消息实体内容。
5. Virtual host：虚拟主机，用于逻辑隔离，最上层消息的路由。一个Virtual host可以若干个Exchange和Queue，同一个Virtual host不能有同名的Exchange或Queue。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 /
6. Exchange：交换机，接受消息，根据路由键转发消息到绑定的队列上。
7. banding：Exchange和Queue之间的虚拟连接，binding中可以包括routing key
8. routing key：一个路由规则，虚拟机根据他来确定如何路由 一条消息。
9. Queue：消息队列，用来存放消息的队列。

• producer

  

  生产者是一个发送消息的程序.

• queue

  消息通过RabbitMQ和你的应用程序传递，但它们只能存储在队列中.

  队列只受主机内存的约束。磁盘限制，它本质上是一个大的消息缓冲区。

• consumer

  

消费者是一个等待接收消息的程序.

生产者、消费者和代理不必驻留在同一个主机上;事实上，在大多数应用程序中，它们并不需要。应用程序也可以是生产者和消费者。

工作模式

“Hello World”

Producer 端

1. 创建连接工厂

   ConnectionFactory factory = new ConnectionFactory();

2. 设置参数

   factory.setHost("47.111.126.61");//ip  默认值 localhostfactory.setPort(5672); //端口  默认值 5672factory.setVirtualHost("/he");//虚拟机 默认值/factory.setUsername("he");//用户名 默认 guestfactory.setPassword("he");//密码 默认值 guest

3. 创建连接 Connection

   Connection connection = factory.newConnection();

4. 创建Channel

   Channel channel = connection.createChannel();

5. 创建队列Queue

   如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建

   queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map
        
          arguments)/*参数：    1. queue：队列名称    2. durable:是否持久化，当mq重启之后，是否还在    3. exclusive：是否独占。独占只能有一个消费者监听这队列    4. autoDelete:是否自动删除。服务器不再使用它时，自动删除掉    5. arguments：该队列的其他属性。*/
        

   channel.queueDeclare("hello_world", true, false, false, null);

6. 发送消息

   ​

   basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)        参数：            1. exchange：将消息发送到哪一个交换机。            	简单模式下交换机会使用默认的 ""            2. routingKey：路由名称            3. props：消息的其他属性-路由标头等            4. body：发送消息数据

   channel.basicPublish("", "hello_world", null, body.getBytes());

7. 释放资源

   channel.close();connection.close();

   Connection和Channel都实现了java.io.Closeable。这样我们就不需要在代码中显式地关闭它们。

Consumer端

我们的生产者就这样了。我们的消费者监听来自RabbitMQ的消息，所以不像发布者只发布一条消息，我们将保持消费者运行来监听消息并打印出来。

1. 创建连接工厂

   ConnectionFactory factory = new ConnectionFactory();

2. 设置参数

   factory.setHost("47.111.126.61");//ip  默认值 localhostfactory.setPort(5672); //端口  默认值 5672factory.setVirtualHost("/he");//虚拟机 默认值/factory.setUsername("he");//用户名 默认 guestfactory.setPassword("he");//密码 默认值 guest

3. 创建连接 Connection

   Connection connection = factory.newConnection();

4. 创建Channel

   Channel channel = connection.createChannel();

5. 创建队列

   ​ 如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建

   ​ 我们在这里也声明了队列。因为我们可能会在发布者之前启动使用者，所以我们希望在尝试使用来自该队列的消息之前确保队列存在。

   channel.queueDeclare("hello_world", true, false, false, null);

6. 接收消息

   String basicConsume(String queue, boolean autoAck, Consumer callback)        参数：            1. queue：队列名称            2. autoAck：是否自动确认            	如果服务器应该期望显式确认，则返回false            3. callback：回调对象的接口    	    4. 返回值 服务器生成的consumerTag        重载方法中常见的参数还有：    DeliverCallback deliverCallback 传递消息时回调    	    ConsumerShutdownSignalCallback shutdownSignalCallback 通道连接关闭时回调        CancelCallback cancelCallback 当消费者被取消时回调        Map
        
          arguments 消费者相关的配置参数
        

   Consumer consumer = new DefaultConsumer(channel) {         @Override    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {             System.out.println("consumerTag = " + consumerTag);        System.out.println("envelope = " + envelope);        System.out.println("properties = " + properties);        System.out.println("body = " + new String (body));    }};channel.basicConsume("hello_world", true, consumer);

后面代码由于和重复较多，就挑不一样的地方写出来了

任务队列

默认情况下，RabbitMQ会将每条消息依次发送给下一个消费者。平均而言，每个消费者将获得相同数量的消息

背后的主要思想是避免立即执行资源密集型任务，并不得不等待它完成。

生产者代码

public static void main(String[] args) throws IOException, TimeoutException {           //1.创建连接工厂        ConnectionFactory factory = new ConnectionFactory();        //2. 设置参数        factory.setHost("47.111.126.61");//ip  默认值 localhost        factory.setPort(5672); //端口  默认值 5672        factory.setVirtualHost("/he");//虚拟机 默认值/        factory.setUsername("he");//用户名 默认 guest        factory.setPassword("he");//密码 默认值 guest        //3. 创建连接 Connection        Connection connection = factory.newConnection();        //4. 创建Channel        Channel channel = connection.createChannel();        //5. 创建队列Queue        //如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建        channel.queueDeclare("work_queue", true, false, false, null);        for (int i = 0; i < 10; i++) {               String body = i + "hello rabbitmq!";            //6. 发送消息            channel.basicPublish("", "work_queue", null, body.getBytes());        }        //7.释放资源        channel.close();        connection.close();    }

消费者1代码

public static void main(String[] args) throws IOException, TimeoutException {           //1.创建连接工厂        ConnectionFactory factory = new ConnectionFactory();        //2. 设置参数        factory.setHost("47.111.126.61");//ip  默认值 localhost        factory.setPort(5672); //端口  默认值 5672        factory.setVirtualHost("/he");//虚拟机 默认值/        factory.setUsername("he");//用户名 默认 guest        factory.setPassword("he");//密码 默认值 guest        //3. 创建连接 Connection        Connection connection = factory.newConnection();        //4. 创建Channel        Channel channel = connection.createChannel();        //5. 创建队列Queue        /*        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map
   
     arguments)        参数：            1. queue：队列名称            2. durable:是否持久化，当mq重启之后，还在            3. exclusive：                * 是否独占。只能有一个消费者监听这队列                * 当Connection关闭时，是否删除队列                *            4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉            5. arguments：参数。         */        //如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建        channel.queueDeclare("work_queue", true, false, false, null);        channel.basicQos(2);        Consumer consumer = new DefaultConsumer(channel) {               @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                   System.out.println("body = " + new String (body));            }        };        channel.basicConsume("work_queue", true, consumer);    }
   

消费者2代码

public static void main(String[] args) throws IOException, TimeoutException {           //1.创建连接工厂        ConnectionFactory factory = new ConnectionFactory();        //2. 设置参数        factory.setHost("47.111.126.61");//ip  默认值 localhost        factory.setPort(5672); //端口  默认值 5672        factory.setVirtualHost("/he");//虚拟机 默认值/        factory.setUsername("he");//用户名 默认 guest        factory.setPassword("he");//密码 默认值 guest        //3. 创建连接 Connection        Connection connection = factory.newConnection();        //4. 创建Channel        Channel channel = connection.createChannel();        channel.basicQos(2);        //5. 创建队列Queue        /*        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map
   
     arguments)        参数：            1. queue：队列名称            2. durable:是否持久化，当mq重启之后，还在            3. exclusive：                * 是否独占。只能有一个消费者监听这队列                * 当Connection关闭时，是否删除队列                *            4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉            5. arguments：参数。         */        //如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建        channel.queueDeclare("work_queue", true, false, false, null);        Consumer consumer = new DefaultConsumer(channel) {               @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                   System.out.println("body = " + new String (body));            }        };        channel.basicConsume("work_queue", true, consumer);    }
   

发布/订阅

回顾

生产者是发送消息的用户应用程序。

队列是存储消息的缓冲区。

消费者是接收消息的用户应用程序。

RabbitMQ消息模型的核心思想是，生产者从不直接向队列发送任何消息。实际上，通常情况下，生产者甚至根本不知道消息是否会被传递到任何队列。

交换机(exchange)

这时就有了交换机这个中间产物，生产者只能将消息发送到交换器。它一方面它接收来自生产者的消息，另一方面它将消息推送到队列。

它应该被附加到一个特定的队列吗?它应该被附加到许多队列中吗?或者它应该被丢弃。这些规则由exchange类型定义。

有几种可用的交换类型:

• direct 直连类型，即在绑定时设定一个 routing_key, 消息的 routing_key 匹配时, 才会被交换器投送到绑定的队列中去，原则是先匹配、后投送；
• topic 按照规则转发类型，支持通配符匹配，和 Direct 功能一样，但是在匹配 routing_key的时候，更加灵活，支持通配符匹配，原则也是先匹配、后投送；
• headers 头部信息匹配转发类型，根据消息头部中的 header attribute 参数类型，将消息转发到对应的队列，原则也是先匹配、后投送；
• fanout 广播 ，正如您可能从名称中猜到的那样，它只是将它接收到的所有消息广播到它知道的所有队列。

Headers

先学一下Header，剩下的几个将会在下面的几个模式中用到，但是因为 header 交换机性能差很多，目前几乎用不到了

headers 也是根据规则匹配, 相比 direct 和 topic 固定地使用 routing_key , headers 则是通过一个自定义匹配规则的消息头部类进行匹配。

在队列与交换器绑定时，会设定一组键值对规则，消息中也包括一组键值对( headers 属性)，当这些键值对有一对, 或全部匹配时，消息被投送到对应队列。

声明交换机

与之前学习的模式不同，我们需要去声明一个 交换机。

exchangeDeclare(String exchange,//交换机名称    BuiltinExchangeType type,//交换机类型    boolean durable,//是否持久化    boolean autoDelete,//是否自动删除    boolean internal,//内部使用。一般false    Map
   
     arguments)//交换机的其他参数配置
   

绑定

queueBind(String queue,//队列名称          String exchange,//交换机名称          String routingKey)//用于绑定的路由密钥

代码示例

• 生成者

//4. 创建Channel        Channel channel = connection.createChannel();        //5. 创建交换机        String exchangeName = "test_fanout";        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT, true, false, false, null);        //6. 创建队列        String queueName1 = "test_fanout_queue1";        String queueName2 = "test_fanout_queue2";        channel.queueDeclare(queueName1, true, false, false, null);        channel.queueDeclare(queueName2, true, false, false, null);        //7. 绑定队列交换机        channel.queueBind(queueName1,exchangeName, "");        channel.queueBind(queueName2,exchangeName, "");        //8. 发送消息        String body = "日志消息张三调用了findAll方法，日志级别info";        channel.basicPublish(exchangeName, "", null, body.getBytes());        //9. 释放资源        channel.close();        connection.close();

• 消费者1

Consumer consumer = new DefaultConsumer(channel) {               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                   System.out.println("打印到控制台......");                System.out.println("body = " + new String (body));            }        };        channel.basicConsume(queueName1 , true, consumer);

消费者2

Consumer consumer = new DefaultConsumer(channel) {               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                   System.out.println("保存到数据库......");                System.out.println("body = " + new String (body));            }        };        channel.basicConsume(queueName2 , true, consumer);

路由(Routing)

绑定是交换机和队列之间的一种关系。

队列对来自此交换机的消息感兴趣。绑定可以使用额外的routing_key参数。

消息被发送到其绑定键与消息的路由键完全匹配的队列中。

Direct exchange

• 第一个队列用绑定键橙色绑定，第二个队列有两个绑定，一个绑定键黑色绑定，另一个绑定键绿色绑定。

• 发布到交换机的带有路由键橙色的消息将被路由到队列Q1。带有黑色或绿色路由键的消息将被发送到Q2。所有其他消息将被丢弃。

Multiple bindings

使用相同的绑定键绑定多个队列是完全合法的。

一个带有routing key 为 black的消息将被发送到Q1和Q2。

代码示例

• 生产者

channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null);        //6. 创建队列        String queue1Name = "test_direct_queue1";        String queue2Name = "test_direct_queue2";        channel.queueDeclare(queue1Name,true,false,false,null);        channel.queueDeclare(queue2Name,true,false,false,null);        //7. 绑定队列和交换机        //队列1绑定 error        channel.queueBind(queue1Name,exchangeName,"error");        //队列2绑定 info  error  warning        channel.queueBind(queue2Name,exchangeName,"info");        channel.queueBind(queue2Name,exchangeName,"error");        channel.queueBind(queue2Name,exchangeName,"warning");        String body = "日志信息：张三调用了delete方法...出错误了。。。日志级别：error...";        //8. 发送消息        channel.basicPublish(exchangeName,"warning",null,body.getBytes());

• 消费者1

// 接收消息        Consumer consumer = new DefaultConsumer(channel){               /*                回调方法，当收到消息后，会自动执行该方法                1. consumerTag：标识                2. envelope：获取一些信息，交换机，路由key...                3. properties:配置信息                4. body：数据             */            @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                   System.out.println("body："+new String(body));                System.out.println("将日志信息打印到控制台.....");            }        };        channel.basicConsume(queue1Name,true,consumer);

• 消费者2

// 接收消息        Consumer consumer = new DefaultConsumer(channel){               @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                   System.out.println("body："+new String(body));                System.out.println("将日志信息存储到数据库.....");            }        };        channel.basicConsume(queue2Name,true,consumer);

通配符模式(Topics)

通配符模式背后的逻辑类似于直接交换——使用特定路由键发送的消息将被发送到所有与匹配绑定键绑定的队列。

在我们的日志系统中，我们可能不仅希望根据严重程度订阅日志，还希望根据发出日志的源订阅日志。

Routingkey 可以有任意多的单词，最多255个字节，多个单词之间以”.”分割。

• 匹配模式规则

  *(星号)可以完全替代一个词。

  #(哈希)可以替换0个或多个单词。

• 举例

  hw.#：能够匹配 hw.insert.abc 或者 hw.insert

  hw.*：只能匹配 hw.insert

图示

该路由键由三个单词(两个点)组成，路由密钥中的第一个单词将描述速度，第二个单词描述颜色，第三个单词描述物种:“速度、颜色、物种”。

​ Q1对所有的橙色动物感兴趣。

​ Q2想听关于兔子和懒惰动物的一切。

• quick.orange.rabbit Q1 Q2
• lazy.orange.elephant Q1 Q2
• 如果我们违反合同，用一到四个字来传递信息，比如orange或quick.orange.male.rabbit，会怎么样?

​ 这些消息将不匹配任何绑定，并将丢失。

• lazy.orange.male.rabbit 即使它有四个单词，将被发送到第二个队列(匹配第二个)

代码示例

• 生产者

//5. 创建交换机        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC, true, false, false, null);        //6. 创建队列        String queue1Name = "test_topic_queue1";        String queue2Name = "test_topic_queue2";        channel.queueDeclare(queue1Name, true, false, false, null);        channel.queueDeclare(queue2Name, true, false, false, null);        //7. 绑定队列和交换机           // routing key  系统的名称.日志的级别。        //=需求： 所有error级别的日志存入数据库，所有order系统的日志存入数据库        channel.queueBind(queue1Name, exchangeName, "#.error");        channel.queueBind(queue1Name, exchangeName, "order.*");        channel.queueBind(queue2Name, exchangeName, "*.*");        String body = "日志信息：张三调用了findAll方法...日志级别：info...";        //8. 发送消息        channel.basicPublish(exchangeName, "goods.error", null, body.getBytes());        //9. 释放资源        channel.close();        connection.close();

• 消费者1

// 接收消息        Consumer consumer = new DefaultConsumer(channel){               @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                   System.out.println("body："+new String(body));                System.out.println("将日志信息存入数据库.......");            }        };        channel.basicConsume(queue1Name,true,consumer);

• 消费者2

// 接收消息        Consumer consumer = new DefaultConsumer(channel){               @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                   System.out.println("body："+new String(body));                System.out.println("将日志信息打印控制台.......");            }        };        channel.basicConsume(queue2Name,true,consumer);

RPC

如果我们需要在远程计算机上运行一个函数并等待结果呢?那就不一样了。这种模式通常被称为远程过程调用或RPC。

我们将使用RabbitMQ构建一个RPC系统:一个客户端和一个可扩展的RPC服务器。因为我们没有任何值得分发的耗时任务，所以我们将创建一个返回斐波那契数列的虚拟RPC服务。

RPC注意事项

​ 尽管RPC是计算中非常常见的模式，但它经常受到批评。当程序员不知道函数调用是本地的还是缓慢的RPC时，就会出现问题。这样的混乱会导致不可预测的系统，并为调试增加不必要的复杂性。

• 确保明确哪个函数调用是本地的，哪个是远程的。
• 明确组件之间的依赖关系。
• 处理错误情况。当RPC服务器长时间关闭时，客户端应该如何反应?

当有疑问时，避免RPC。如果可以，您应该使用异步管道——而不是类似rpc的阻塞，结果将异步地推到下一个计算阶段。

客户端接口

为了说明如何使用RPC服务，我们将创建一个简单的客户机类。它将公开一个名为call的方法，它发送一个RPC请求并阻塞，直到收到答案:

FibonacciRpcClient fibonacciRpc = new FibonacciRpcClient();String result = fibonacciRpc.call("4");System.out.println( "fib(4) is " + result);

回调队列

一般来说，在RabbitMQ上执行RPC很容易。客户端发送请求消息，服务器用响应消息进行响应。为了接收响应，我们需要发送一个回调队列地址与请求。我们可以使用默认队列(在Java客户机中是独占的)。

callbackQueueName = channel.queueDeclare().getQueue();BasicProperties props = new BasicProperties                            .Builder()                            .replyTo(callbackQueueName)                            .build();channel.basicPublish("", "rpc_queue", props, message.getBytes());// ... 从callback_queue读取响应消息后的逻辑

消息属性

AMQP 0-9-1协议预先定义了一组带有消息的14个属性。

• deliveryMode:将消息标记为持久(值为2)或瞬态(任何其他值)。

• contentType:用于描述编码的mime类型。例如，对于经常使用的JSON编码，最好将此属性设置为:application/ JSON。

• replyTo:通常用于命名回调队列。

• correlationId:用于关联RPC响应和请求。

Correlation Id

在上述方法中，我们建议为每个RPC请求创建一个回调队列。这非常低效，但幸运的是，有一种更好的方法——让我们为每个客户机创建一个回调队列。

这就产生了一个新问题，在队列中接收到响应后，并不清楚该响应属于哪个请求。这时就要使用correlationId属性。我们将为每个请求设置一个唯一的值。稍后，当我们在回调队列中接收到一条消息时，我们将查看此属性，并基于此，我们将能够将响应与请求匹配起来。如果我们看到一个未知的correlationId值，我们可以安全地丢弃消息——它不属于我们的请求。

您可能会问，为什么我们要忽略回调队列中的未知消息，而不是因为错误而失败呢?这是由于服务器端可能存在竞争条件。尽管不太可能，但RPC服务器有可能在向我们发送应答之后，但在为请求发送确认消息之前死亡。如果发生这种情况，重新启动的RPC服务器将再次处理请求。这就是为什么在客户机上我们必须优雅地处理重复响应，并且RPC理想情况下应该是幂等的。

工作原理图示

RPC工作流程:

1. 对于RPC请求，客户机发送带有两个属性的消息:replyTo，它被设置为一个仅为请求创建的匿名独占队列，correlationId，它被设置为每个请求的唯一值。

2. 请求被发送到rpc_queue队列。RPC工作者(又名:服务器)正在等待该队列上的请求。

3. 当请求出现时，它执行任务并使用replyTo字段中的队列将带有结果的消息发送回客户机。

4. 客户端在应答队列中等待数据。当消息出现时，它检查correlationId属性。如果它与请求中的值匹配，则将响应返回给应用程序。

我们的RPC服务器代码可以在这里找到: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();                   }                }            }        }    }}

1. 像往常一样，我们首先建立连接、通道和声明队列。
2. 我们可能想要运行多个服务器进程。为了将负载均匀地分散到多个服务器上，我们需要在channel.basicQos中设置prefetchCount设置。
3. 我们使用basicConsume访问队列，其中我们以对象(DeliverCallback)的形式提供一个回调，该回调将完成工作并将响应发送回去。

rpccclient .java。

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

1. 我们建立connection 和channel。
2. 我们的call方法发出实际的RPC请求。
3. 我们首先生成一个唯一的correlationId编号并保存它---------我们的消费者回调将使用这个值来匹配适当的响应。
4. 然后，我们为应答创建一个专用的独占队列并订阅它。
5. 接下来，我们发布带有两个属性的请求消息:replyTo和correlationId。
6. 因为我们的使用者交付处理是在一个单独的线程中进行的，所以我们需要一些东西在响应到达之前挂起主线程。使用BlockingQueue是一种可能的解决方案。这里我们创建的ArrayBlockingQueue的容量设置为1，因为我们只需要等待一个响应。

消费者所做的工作非常简单，对于每个被消费的响应消息，它都会检查correlationId是否是我们正在寻找的。如果是，它将响应放置到BlockingQueue。与此同时，主线程正在等待来自BlockingQueue的响应。

最后，我们将响应返回给用户。

转载地址：http://sqtg.baihongyu.com/