消息队列——RabbitMQ
25r9n4qy8
8年前
<h2><strong>1. 写在前面</strong></h2> <p>昨天简单学习了一个 <strong>消息队列</strong> 项目—— <strong>RabbitMQ</strong> ，今天趁热打铁，将学到的东西记录下来。</p> <p>学习的资料主要是官网给出的6个基本的消息发送/接收 模型 ，或者称为6种不同的使用场景，本文便是对这6种模型加以叙述。</p> <h2><strong>2. Tutorials</strong></h2> <p>在学习6种模型之前，我们首先需要安装RabbitMQ。RabbitMQ支持多种系统平台，各平台的安装方法。安装好之后，我们使用如下命令启用Web端的管理插件： rabbitmq-plugins enable rabbitmq_management ，然后启动RabbitMQ。接着用浏览器访问 http://localhost:15672/ ，若能看到RabbitMQ相关Web页面，说明启动成功。</p> <h3><strong>2.1 Hello World</strong></h3> <p>正所谓万事开头难，我们先从最简单的 <strong>Hello World</strong> 开始。首先当然是新建一个项目，导入RabiitMQ相关jar。我采用Maven来构建项目，因此只需要在pom文件中添加如下依赖：</p> <pre> <code class="language-dust"><dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>3.6.5</version> </dependency></code></pre> <p>接下来学习最简单的消息队列模型，如下图：</p> <p style="text-align:center"><img src="https://simg.open-open.com/show/ec1c1ed3525c8971c26f12620578e287.png"></p> <p>在图中，P代表 producer，它是消息的 <strong>生产者</strong>；C代表 consumer ，它是消息的 <strong>消费者</strong> ；而红色的矩形正是我们所谓的 <strong>消息队列</strong> ，它位于 RabbitMQ 中（ RabbitMQ 中可以有很多这样的队列，并且每个队列都有一个唯一的名字）。生产者（们）可以将消息发送到消息队列中，消费者（们）可以从消息队列中取出消息。</p> <p>这种模型是不是很简单呢？下面我们使用 <strong>Java</strong> ，借助于RabbitMQ来实现这种模型的消息通信。</p> <p>首先我们介绍如何 send 消息到消息队列。 send 之前，当然是和RabbitMQ服务器建立连接：</p> <pre> <code class="language-dust">ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection();</code></pre> <p>接下来我们创建一个 channel ，大多数API都是通过这个对象来调用的：</p> <pre> <code class="language-dust">Channel channel = connection.createChannel();</code></pre> <p>之后，我们便可以调用 channel 的如下方法去声明一个队列：</p> <pre> <code class="language-dust">channel.queueDeclare("hello", false, false, false, null);</code></pre> <p>该方法的第一个参数是队列的名称，其余的参数先不管，之后会介绍。我们可以尝试着去执行以上的5行代码，然后打开Web端，可以看到新建了一个叫作 hello 的队列：</p> <p style="text-align: center;"><img src="https://simg.open-open.com/show/665fba8e97cce8c23d613ee73c687ce8.png"></p> <p>有了队列，我们便可以向其中发送消息了，同样还是调用 channel 对象的API：</p> <pre> <code class="language-dust">channel.basicPublish("", "hello", null, "Hello World".getBytes());</code></pre> <p>以上代码所做的事情就是发送了一条字符串消息“Hello World”（第4个参数）到消息队列。你可能注意到我们调用了String对象的 getBytes 方法，没错，我们发送的实际上二进制数据。因此，理论上你能够发送任何数据到消息队列中，而不仅仅是文本信息。</p> <p>第2个参数叫做 <strong>路由键（routingKey）</strong> ，在该模型下必须与队列名相同，至于为什么，和其他参数一样，之后会了解到。</p> <p>我们可以修改发送的文本，再次执行上述代码，然后打开Web端查看，便可以查看到我们发送的消息：</p> <p style="text-align: center;"><img src="https://simg.open-open.com/show/5179cbeaa84d39c7b98d0d717b59f1f9.png"></p> <p>点击上图的 <strong>name</strong> 字段下的 <strong>hello</strong> ，可以查看 hello 队列中的具体信息：</p> <p style="text-align:center"><img src="https://simg.open-open.com/show/909a48393b1a4b18f5e672dc05a4421d.png"></p> <p>接下来，我们去尝试着去获取 <strong>生产者</strong> 发送的消息，和 send 方法一样，我们同样需要连接服务器，创建 channel ，声明队列：</p> <pre> <code class="language-dust">ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare("hello", false, false, false, null);</code></pre> <p>之后我们可以调用 channel 的相关方法去监听队列，接收消息：</p> <pre> <code class="language-dust">channel.basicConsume("hello", true, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, "UTF-8")); } });</code></pre> <p>以上 basicConsume 方法中，第一个参数是队列的名字；第二个参数表示是否自动确认消息的接收情况，我们使用true，自动确认；第三个参数需要传入一个实现了 Consumer 接口的对象，我们简单的 new 一个默认的 Consumer 的实现类 DefaultConsumer ，然后在 handleDelivery 方法中去处理接收到的消息（ handleDelivery 方法会在接收到消息时被回调）。</p> <p>运行以上代码，我们可以打印出之前向队列中 send 的数据：</p> <pre> <code class="language-dust">Hello World Hello World2</code></pre> <p>下面是 <strong>Hello World</strong> 的完整代码：</p> <pre> <code class="language-dust">public class App { @Test public void send() throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare("hello", false, false, false, null); channel.basicPublish("", "hello", null, "Hello World2".getBytes()); channel.close(); connection.close(); } @Test public void receive() throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare("hello", false, false, false, null); channel.basicConsume("hello", true, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, "UTF-8")); } }); synchronized (this){ // 因为以上接收消息的方法是异步的（非阻塞），当采用单元测试方式执行该方法时，程序会在打印消息前结束，因此使用wait来防止程序提前终止。若使用main方法执行，则不需要担心该问题。 wait(); } } }</code></pre> <h3><strong>2.2 Work queues</strong></h3> <p>接下来我们学习第二种模型—— <strong>Work Queues</strong> 。顾名思义，这种模型描述的是一个生产者（Boss）向队列发消息（任务），多个消费者（worker）从队列接受消息（任务），如下图所示：</p> <p style="text-align:center"><img src="https://simg.open-open.com/show/0fde8ffebb772c7d486921464ae55a29.png"></p> <p>下面我们用代码去实现。先是生产者 send 消息到队列，这次我们多发送些数据：</p> <pre> <code class="language-dust">@Test public void send() throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare("hello", false, false, false, null); for (int i = 0; i < 9; i++) { channel.basicPublish("", "hello", null, String.valueOf(i).getBytes()); } channel.close(); connection.close(); }</code></pre> <p>然后是消费者接收数据：</p> <pre> <code class="language-dust">@Test public void receive() throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare("hello", false, false, false, null); channel.basicConsume("hello", true, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, "UTF-8")); try { // Thread.sleep(1000); Thread.sleep(10000); } catch (InterruptedException e) { e.printStackTrace(); } } }); synchronized (this) { wait(); } }</code></pre> <p>代码基本上和 Hello World 的代码一样，只是加上句 sleep 来模拟消费者（worker）处理消息所花的时间。</p> <p>我们可以先执行三次 receive 方法（修改 sleep 的时间，其中消费者1 sleep 10s，消费者2,3 sleep 1s），让三个消费者（worker）一起等待消息的到来，然后执行 send 方法发送9条消息，观察三个消费者收到的消息情况。</p> <p>若不出意外，你会看到如下的打印结果：</p> <pre> <code class="language-dust">// 消费者1 0 // 10s 后 3 // 10s 后 6 // 消费者2 1 // 1s 后 4 // 1s 后 7 // 消费者3 2 // 1s 后 5 // 1s 后 8</code></pre> <p>通过打印结果，我们可以总结出 <strong>Work queues</strong> 的几个特点：</p> <ol> <li>一条消息只会被一个消费者接收；</li> <li>消息是平均分配给消费者的；</li> <li>消费者只有在处理完某条消息后，才会收到下一条消息。</li> </ol> <p>Work queues（Task Queuess）的概念在一些Web场景的应用中是很有用的，比如我们能够用它来构建一个master-slave结构的分布式爬虫系统：系统中有一个master节点和多个slave节点，master节点负责向各个slave节点分配爬取任务。</p> <h3><strong>2.3 Publish/Subscribe</strong></h3> <p>但有些时候，我们可能希望一条消息能够被多个消费者接受到，比如一些公告信息等，这时候用 <strong>Work Queue</strong> 模型显然不合适，而 <strong>Publish/Subscribe</strong> 模型正是对应这种使用场景的。</p> <p>在介绍Publish/Subscribe之前，我们快速回顾之前的两个模型，它们好像都是生产者将消息直接发送到消息队列，但其实不是这样的，甚至有可能生产者根本就不知道消息发送到了哪一个消息队列。</p> <p>先别着急，下面我们完整地介绍RabbitMQ消息发送/接受的方式。</p> <p>事实上，生产者是把消息发送到了 <strong>交换机（exchange）</strong> 中，然后交换机负责（决定）将消息发送到（哪一个）消息队列中。其模型如下图：</p> <p style="text-align:center"><img src="https://simg.open-open.com/show/78834739c797fabd64b5dbec0863752b.jpg"></p> <p>这时候你可能会疑惑：既然消息是被发送到了交换机中，那我们之前发送的消息是被发送到了哪一个交换机中了？它有没有机制能够让特定的消息发送到指定的队列？</p> <p>先回答第一个问题。还记得我们在 <strong>Hello World</strong> 中写的发送消息的代码吗？</p> <pre> <code class="language-dust">channel.basicPublish("", "hello", null, message.getBytes());</code></pre> <p>事实上第一个参数便是指定交换机的名字，即指定消息被发送到哪一个交换机。空字符串表示 <strong>默认交换机（Default Exchange）</strong> ，即我们之前发送的消息都是先发送到 <strong>默认交换机</strong> ，然后它再路由到相应的队列中。其实我们可以通过Web页面去查看所有存在的交换机：</p> <p style="text-align:center"><img src="https://simg.open-open.com/show/e559a4c7b1fe908d341347bb43a1b657.png"></p> <p>接着回答第二个问题。路由的依据便是通过第二个参数—— <strong>路由键（routing key）</strong> 指定的，之前已经提到过。在之前代码中，我们指定第二个参数为"hello"，便是指定消息应该被交换机路由到路由键为hello的队列中。而 <strong>默认交换机（Default Exchange）</strong> 有一个非常有用的性质：</p> <p>每一个被创建的队列都会被自动的绑定到默认交换机上，并且路由键就是队列的名字。</p> <p>交换机还有4种不同的类型，分别是 direct ， fanout ， topic ， headers ，每种类型路由的策略不同。</p> <p>direct 类型的交换机要求和它绑定的队列带有一个路由键K，若有一个带有路由键R的消息到达了交换机，交换机会将此消息路由到路由键K = R的队列。默认交换机便是该类型。因此，在下图中，消息会沿着绿色箭头路由：</p> <p style="text-align:center"><img src="https://simg.open-open.com/show/a84fa50921a68eb0a2fb6217d234340a.jpg"></p> <p>fanout 类型的交换机会路由每一条消息到所有和它绑定的队列，忽略路由键。</p> <p>剩下的两种类型之后再做介绍。</p> <p>在以上概念基础上，我们来看第3种消息模型： <strong>Publish/Subscribe</strong> 。如下图：</p> <p style="text-align:center"><img src="https://simg.open-open.com/show/59a7bcc69b4bf5388ff28f0790ca5b90.png"></p> <p>该模型是要让所有的消费者都能够接收到每一条消息。显然， fanout 类型的交换机更符合我们当前的需求。为此，先创建一个 fanout 类型的交换机。</p> <pre> <code class="language-dust">channel.exchangeDeclare("notice", "fanout");</code></pre> <p>其中，第一个参数是交换机的名称；第二个参数是交换机的类型。</p> <p>然后我们可以 send 消息了：</p> <pre> <code class="language-dust">channel.basicPublish( "notice", "", null, message.getBytes());</code></pre> <p>对于消费者，我们需要为每一个消费者创建一个独立的队列，然后将队列绑定到刚才指定的交换机上即可：</p> <pre> <code class="language-dust">// 该方法会创建一个名称随机的临时队列 String queueName = channel.queueDeclare().getQueue(); // 将队列绑定到指定的交换机（"notice"）上 channel.queueBind(queueName, "notice", "");</code></pre> <p>以下完整的代码：</p> <pre> <code class="language-dust">@Test public void send() throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare("notice", "fanout"); channel.basicPublish( "notice", "", null, "Hello China".getBytes()); channel.close(); connection.close(); } @Test public void receive() throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare("notice", "fanout"); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, "notice", ""); channel.basicConsume(queueName, true, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, "UTF-8")); } }); synchronized (this) { wait(); } }</code></pre> <p>首先运行两次 receive 方法，让两个消费者等待接收消息，然后可以在Web端查看此时的队列情况，如下图所示：</p> <p style="text-align: center;"><img src="https://simg.open-open.com/show/b4f653f0ac8d6bd90784652ecd3f906a.png"></p> <p>可以看到图中有两个名称随机的队列。接着运行 send 方法发送一条消息，最终我们会看到两个消费者都打印出了 Hello China 。然后停止虚拟机让消费者断开连接，再次在Web端查看队列情况，会发现刚才的两个队列被自动删除了。</p> <h3><strong>2.4 Routing</strong></h3> <p>以上是Publish/Subscribe模式，它已经能让我们的通知（notice）系统正常运转了。现在再考虑这样一个新需求：对于一些机密通知，我们只想让部分人看到。这就要求交换机对绑定在其上的队列进行筛选，于是引出了又一个新的模型： <strong>Routing</strong> 。</p> <p>之前我们说过，对于 direct 类型的交换机，它会根据 <strong>routing key</strong> 进行路由，因此我们可以借助它来实现我们的需求，模型结构如下图：</p> <p style="text-align: center;"><img src="https://simg.open-open.com/show/05d48de4439d8e1a8d1778295a308652.jpg"></p> <p>下面用代码来实现。先看生产者。</p> <p>首先要声明一个 direct 类型的交换机：</p> <pre> <code class="language-dust">// 这里名称改为notice2 channel.exchangeDeclare("notice2", "direct");</code></pre> <p>需要注意的是， 因为我们之前声明了一个 fanout 类型的名叫 notice 的交换机，因此不能再声明一个同名的类型却不一样的交换机。</p> <p>然后可以发送消息了，我们发送10条消息，其中偶数条消息是秘密消息，只能被routing key 为s的队列接受，其余的消息所有的队列都能接受。</p> <pre> <code class="language-dust">for (int i = 0; i < 10; i++) { String routingKey = "n"; // normal if (i % 2 == 0) { routingKey = "s"; // secret } channel.basicPublish("notice2", routingKey, null, String.valueOf(i).getBytes()); }</code></pre> <p>接下来是消费者：</p> <pre> <code class="language-dust">// 声明一个名称随机的临时的队列 String queueName = channel.queueDeclare().getQueue(); // 绑定交换机，同时带上routing key channel.queueBind(queueName, "notice2", "n"); // 消费者2号运行时，打开以下注释 // channel.queueBind(queueName, "notice2", "s");</code></pre> <p>注意，我们可以多次调用队列绑定方法，调用时，队列名和交换机名都相同，而routing key不同，这样可以使一个队列带有多个routing key。</p> <p>以下是完整代码：</p> <pre> <code class="language-dust">@Test public void send() throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare("notice2", "direct"); for (int i = 0; i < 10; i++) { String routingKey = "n"; // normal if (i % 2 == 0) { routingKey = "s"; // secret } channel.basicPublish("notice2", routingKey, null, String.valueOf(i).getBytes()); } channel.close(); connection.close(); } @Test public void receive() throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare("notice2", "direct"); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, "notice2", "n"); // channel.queueBind(queueName, "notice2", "s"); channel.basicConsume(queueName, true, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, "UTF-8")); } }); synchronized (this) { wait(); } }</code></pre> <p>测试时，我们可以先运行一个 receive ，然后打开 channel.queueBind(queueName, "notice2", "s") 注释，再运行一次 receive ，这样就有两个消费者绑定到notice2交换机上，其中消费者1只能收到normal类型的消息，而消费者2既能收到normal类型的消息，又能收到secret类型的消息。接着可以运行send方法。如不出意外，可以看到如下打印结果：</p> <pre> <code class="language-dust">// 消费者1 1 3 5 7 9 // 消费者2 0 1 2 3 4 5 6 7 8 9</code></pre> <h3>2.5 Topics</h3> <p>有了以上的改进，我们的 notice 系统基本ok了。但有些时候，我们还需要更加灵活的消息刷选方式。比如我们对于电影信息，我们可能需要对它的地区，类型，限制级进行筛选。这时候就要借助 <strong>Topics</strong> 模型了。</p> <p>在 <strong>Topics</strong> 模型中，我们“升级”了 <strong>routing key</strong> ，它可以由多个关键词组成，词与词之间由点号( . )隔开。特别地，规定 <strong> * 表示任意的一个词； # 号表示任意的0个或多个词 </strong> 。</p> <p>假设我们现在需要接收电影信息，每条电影消息附带的 <strong>routingKey</strong> 有地区、类型、限制级3个关键字，即： district.type.age 。现在想实现的功能如下图：</p> <p style="text-align:center"><img src="https://simg.open-open.com/show/1d388579b2fc933cd5fa7583e45e2ad6.jpg"></p> <p>如上图所示，队列Q1只关心美国适合13岁以上的电影信息，队列Q2对动作片感兴趣，而队列Q3喜欢中国电影。</p> <p>下面用 <strong>Java</strong> 代码去实现上述功能，相较于之前基本上没有什么改变，下面直接给出代码：</p> <pre> <code class="language-dust">@Test public void send() throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare("movie", "topic"); channel.basicPublish("movie", "American.action.13", null, "The Bourne Ultimatum".getBytes()); channel.basicPublish("movie", "American.comedy.R", null, "The Big Lebowski".getBytes()); channel.basicPublish("movie", "American.romantic.13", null, "Titanic".getBytes()); channel.basicPublish("movie", "Chinese.action.13", null, "卧虎藏龙".getBytes()); channel.basicPublish("movie", "Chinese.comedy.13", null, "大话西游".getBytes()); channel.basicPublish("movie", "Chinese.romantic.13", null, "梁山伯与祝英台".getBytes()); channel.close(); connection.close(); } @Test public void receive() throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare("movie", "topic"); // 队列1 String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, "movie", "American.*.13"); // 队列2 // String queueName = channel.queueDeclare().getQueue(); // channel.queueBind(queueName, "movie", "*.action.*"); // 队列3 // String queueName = channel.queueDeclare().getQueue(); // channel.queueBind(queueName, "movie", "Chinese.#"); channel.basicConsume(queueName, true, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, "UTF-8")); } }); synchronized (this) { wait(); } }</code></pre> <p>运行3次 receive 方法，注意打开或关闭相应的注释；再运行 send 方法，可以看到控制台输出如下内容：</p> <pre> <code class="language-dust">// 消费者1 The Bourne Ultimatum Titanic // 消费者2 The Bourne Ultimatum 卧虎藏龙 // 消费者3 卧虎藏龙 大话西游 梁山伯与祝英台</code></pre> <h3>2.6 RPC</h3> <p>第6种模型是用来做RPC（Remote procedure call， 远程程序调用）的。这里直接贴上代码，就不做解释了 。代码演示的是，客户端调用服务端的 fib 方法，得到返回结果。</p> <p>RPCServer.java</p> <pre> <code class="language-dust">import com.rabbitmq.client.*; import com.rabbitmq.client.AMQP.BasicProperties; /** * Description: * * @author derker * @Time 2016-10-26 18:24 */ 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) { Connection connection = null; Channel channel = null; try { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); connection = factory.newConnection(); channel = connection.createChannel(); channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null); channel.basicQos(1); QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(RPC_QUEUE_NAME, false, consumer); System.out.println(" [x] Awaiting RPC requests"); while (true) { String response = null; QueueingConsumer.Delivery delivery = consumer.nextDelivery(); AMQP.BasicProperties props = delivery.getProperties(); BasicProperties replyProps = new BasicProperties .Builder() .correlationId(props.getCorrelationId()) .build(); try { String message = new String(delivery.getBody(), "UTF-8"); int n = Integer.parseInt(message); System.out.println(" [.] fib(" + message + ")"); response = "" + fib(n); } catch (Exception e) { System.out.println(" [.] " + e.toString()); response = ""; } finally { channel.basicPublish("", props.getReplyTo(), replyProps, response.getBytes("UTF-8")); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } } catch (Exception e) { e.printStackTrace(); } finally { if (connection != null) { try { connection.close(); } catch (Exception ignore) { } } } } }</code></pre> <p>RPCClient.java</p> <pre> <code class="language-dust">import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.Connection; import com.rabbitmq.client.Channel; import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.client.AMQP.BasicProperties; import java.util.UUID; /** * Description: * * @author derker * @Time 2016-10-26 18:36 */ public class RPCClient { private Connection connection; private Channel channel; private String requestQueueName = "rpc_queue"; private String replyQueueName; private QueueingConsumer consumer; public RPCClient() throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); connection = factory.newConnection(); channel = connection.createChannel(); replyQueueName = channel.queueDeclare().getQueue(); consumer = new QueueingConsumer(channel); channel.basicConsume(replyQueueName, true, consumer); } public String call(String message) throws Exception { String response = null; String corrId = UUID.randomUUID().toString(); BasicProperties props = new BasicProperties .Builder() .correlationId(corrId) .replyTo(replyQueueName) .build(); channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8")); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); if (delivery.getProperties().getCorrelationId().equals(corrId)) { response = new String(delivery.getBody(), "UTF-8"); break; } } return response; } public void close() throws Exception { connection.close(); } public static void main(String[] argv) { RPCClient fibonacciRpc = null; String response = null; try { fibonacciRpc = new RPCClient(); System.out.println(" [x] Requesting fib(10)"); response = fibonacciRpc.call("10"); System.out.println(" [.] Got '" + response + "'"); } catch (Exception e) { e.printStackTrace(); } finally { if (fibonacciRpc != null) { try { fibonacciRpc.close(); } catch (Exception ignore) { } } } } }</code></pre> <p> </p> <p>来自：http://www.cnblogs.com/dongkuo/p/6001791.html</p> <p> </p>