RabbitMQ学习总结（3）——入门实例教程详解

女爷i 2022-10-02 08:58 139阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

## 一、起航 ##

本章节，柯南君将从几个层面，用官网例子讲解一下RabbitMQ的实操经典程序案例，让大家重新回到经典“Hello world！”（The simplest thing that does something ）时代，RabbitMQ 支持N多种客户端（client），这里无法一一讲解，暂定java client，有时间的情况下，在弥补一下。

事先，先普及一下图标（我们会在下面的事例中，会大量用到，所以先普及一下，便于识别，最终更好理解事例的含义）

### 1、图标概念 ###

① producting（生产者）：在程序中 发送消息的一端，我们暂且称之为 生产者，在这里用“p”表示

![31125931_xUZa.jpg][]

② queue（队列）：队列是一个邮箱的名字。它住在RabbitMQ。尽管消息流经RabbitMQ和您的应用程序,他们只可以存储在一个队列中。队列是不受任何限制,它可以储存尽可能多的信息（按你兴趣来了）,它本质上是一个无限缓冲区。许多生产商可以发送消息到一个队列,许多消费者可以尝试接收数据从一个队列。

![31125931_vKWK.jpg][]

③ consuming(消费者)：消费者和生产者是对应的，较为相似的意思；在这里，我用“C”表示

![31125931_x4bU.jpg][]

### 2、The Java client library ###

RabbitMQ 中 AMQP这是一个开放的、通用的协议消息。有许多客户AMQP在许多不同的语言。我们将使用提供的Java客户机RabbitMQ。

我们需要下载（Download） client library package，并要核实每个jar包，解压到相应位置，如下图所示：

第一步：点击 [http://www.rabbitmq.com/java-client.html][http_www.rabbitmq.com_java-client.html]，然后找到相应的lib下载位置

![31125931_UNLO.jpg][]

第二步：选择合适的下载，比如我下载了zip包，如图所示：

![31125931_KbqO.jpg][]

第三步：Unzip it（解压它） 到你的working directory（工作目录）中 and grab （并且获得）你的jar包文件

`$ unzip rabbitmq-java-client-bin-*.zip $ cp rabbitmq-java-client-bin-*/*.jar ./`

### 二、程序案例 ###

#### 1） "Hello World"  ####

在这部分教程中我们将用Java写两个程序,发送一个消息的生产者,消费者接收信息并打印出来。我们会掩盖一些细节的Java API,集中在这个非常简单的东西开始。这是一个“Hello World”的消息。在下面的图中,“P”是我们的生产和“C”是我们的消费者。中间的框是一个队列,消息缓冲区RabbitMQ保持代表消费者。

![31125931_ol6d.jpg][]

##### ** ① sending （发送）** #####

![31125931_Wy1a.jpg][]

首先 让The sender（消息发送者）发送消息并且让我们的receiver （消息接收者）接收消息，The sender（消息发送者）将会connect to（连接）RabbitMQ,发送一个single message（单一的信息），然后exit（退出）。

*  在send.java 中，我们需要import一些class ，如下所示：

import com.rabbitmq.client.ConnectionFactory;
     import com.rabbitmq.client.Connection;
     import com.rabbitmq.client.Channel;

*  set up（设置）类和queue的name

public class Send { 
    
     		 private final static String QUEUE_NAME = "hello";
    
      		public static void main(String[] argv)
        	 	 throws java.io.IOException { 
         		 ...
     		 }
     }

*  then 我们create 一个connection （连接）到server（服务端）

onnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

**备注** **：**

*   这个connection 是抽象的socket connection链接；
 *   负责协议版本（protocol version negotigation）和身份认证（authentication ）；
 *  我们在本地机器上连接到一个代理即 localhost ，如果我们想要连接到代理不同机器上我们简单的指定其名称或者IP地址即可；

接下来，我们创建一个channel（通道），这个通道汇集了大多数的API服务！

为了发送，我们必须先声明一个为我们发送queue（队列），然后，往queue里发送一个message

channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        String message = "Hello World!";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

消息内容是一个字节数组,所以你可以编码任何你喜欢的。最后,我们关闭通道和连接;

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

问题 ：

如果 sending doesn‘t work！ 我们将怎么办？why？

如果这是你第一次使用RabbitMQ并且你看不到“发送的”消息,那么你可能抓耳挠 腮没有足够的空闲磁盘空间(默认情况下它需要至少1 gb免费),因此拒绝接受消息。检查代理日志文件确认,如果有必要减少限制。配置文件的文档将向您展示如何设置disk\_free\_limit。

接下来的是send.java所有源代码：

**\[java\]**  [view plain][] [copy][view plain] [print][view plain] [?][view plain]

1.  **import** com.rabbitmq.client.ConnectionFactory;  
2.  **import** com.rabbitmq.client.Connection;  
3.  **import** com.rabbitmq.client.Channel;  
4.    
5.  **public** **class** Send \{  
6.        
7.    **private** **final** **static** String QUEUE\_NAME = "hello";  
8.    
9.    **public** **static** **void** main(String\[\] argv) **throws** Exception \{  
10.                 
11.     ConnectionFactory factory = **new** ConnectionFactory();  
12.     factory.setHost("localhost");  
13.     Connection connection = factory.newConnection();  
14.     Channel channel = connection.createChannel();  
15.   
16.     channel.queueDeclare(QUEUE\_NAME, **false**, **false**, **false**, **null**);  
17.     String message = "Hello World!";  
18.     channel.basicPublish("", QUEUE\_NAME, **null**, message.getBytes());  
19.     System.out.println(" \[x\] Sent '" + message + "'");  
20.       
21.     channel.close();  
22.     connection.close();  
23.   \}  
24. \}

#####    ** ② Receiving （接收）** #####

这就是我们的发送者。我们的接收器是将消息从RabbitMQ,所以不像发送方发布一个消息,我们将保持运行监听消息并打印出来

![31125931_e2tS.jpg][]

*  The code (in [Recv.java][]) has almost the same imports as Send:

import com.rabbitmq.client.ConnectionFactory;
     import com.rabbitmq.client.Connection;
     import com.rabbitmq.client.Channel;
     import com.rabbitmq.client.QueueingConsumer;

额外的QueueingConsumer是一个类,我们将使用缓冲消息推到我们的服务器。设置发送者一样,我们打开一个连接和一个通道,并宣布我们将使用的队列。注意这与队列,发送发布。

public class Recv { 
    
      private final static String QUEUE_NAME = "hello";
    
      public static void main(String[] argv)
          throws java.io.IOException,
                 java.lang.InterruptedException { 
    
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
    
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        ...
        }
    }

注意,我们在这里声明队列。因为我们可能会在发送方之前 **开始启动接收方**,我们要确保队列存在之前我们尝试使用消息。我们要告诉服务器提供我们从队列的消息。因为它将异步消息,我们提供一个回调对象的形式,将缓冲的消息,直到我们准备使用它们。

QueueingConsumer要做什么呢？

QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME, true, consumer);
    
        while (true) { 
          QueueingConsumer.Delivery delivery = consumer.nextDelivery();
          String message = new String(delivery.getBody());
          System.out.println(" [x] Received '" + message + "'");
        }

> QueueingConsumer.nextDelivery()块,直到另一个来自服务器的消息交付。  
> 
>   
> 
> 
> **下面是Recv.java 源代码：**

**\[java\]**  [view plain][] [copy][view plain] [print][view plain] [?][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][31125932_Zm0F.svg]][31125932_Zm0F.svg 1]

1.  **import** com.rabbitmq.client.ConnectionFactory;  
2.  **import** com.rabbitmq.client.Connection;  
3.  **import** com.rabbitmq.client.Channel;  
4.  **import** com.rabbitmq.client.QueueingConsumer;  
5.    
6.  **public** **class** Recv \{  
7.        
8.      **private** **final** **static** String QUEUE\_NAME = "hello";  
9.    
10.     **public** **static** **void** main(String\[\] argv) **throws** Exception \{  
11.   
12.     ConnectionFactory factory = **new** ConnectionFactory();  
13.     factory.setHost("localhost");  
14.     Connection connection = factory.newConnection();  
15.     Channel channel = connection.createChannel();  
16.   
17.     channel.queueDeclare(QUEUE\_NAME, **false**, **false**, **false**, **null**);  
18.     System.out.println(" \[\*\] Waiting for messages. To exit press CTRL+C");  
19.       
20.     QueueingConsumer consumer = **new** QueueingConsumer(channel);  
21.     channel.basicConsume(QUEUE\_NAME, **true**, consumer);  
22.       
23.     **while** (**true**) \{  
24.       QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
25.       String message = **new** String(delivery.getBody());  
26.       System.out.println(" \[x\] Received '" + message + "'");  
27.     \}  
28.   \}  
29. \}

转载于:https://my.oschina.net/zhanghaiyang/blog/594585

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220113/0adf6d6cf20b4bd9b5f0827d766bcba8.png
[31125931_xUZa.jpg]: /images/20220113/cb91a8ad278242f68fde3d441c2237b4.png
[31125931_vKWK.jpg]: /images/20220113/f992f68e5f7b4d2790f54989ade0cf21.png
[31125931_x4bU.jpg]: /images/20220113/0d0c7a09e1e64a93a9951febcfdb94e7.png
[http_www.rabbitmq.com_java-client.html]: http://www.rabbitmq.com/java-client.html
[31125931_UNLO.jpg]: /images/20220113/c7b27266ada74a558f22da6e81c5d77c.png
[31125931_KbqO.jpg]: /images/20220113/a6d9a3f407ad45dfa5a7386b432245d7.png
[31125931_ol6d.jpg]: /images/20220113/d354fd0f27f8421886a6d69b1163b316.png
[31125931_Wy1a.jpg]: /images/20220113/2f372a25e44f40838287a07f4829f24c.png
[view plain]: http://blog.csdn.net/sun305355024sun/article/details/41925469#
[31125931_e2tS.jpg]: /images/20220113/208ba89a0ca84402a70608ac62cb12ed.png
[Recv.java]: http://github.com/rabbitmq/rabbitmq-tutorials/blob/master/java/Recv.java
[CODE]: http://static.oschina.net/uploads/img/201512/31125931_n0i9.png
[CODE_CODE]: https://code.csdn.net/snippets/551841
[31125932_Zm0F.svg]: http://static.oschina.net/uploads/img/201512/31125932_Zm0F.svg
[31125932_Zm0F.svg 1]: https://code.csdn.net/snippets/551841/fork