RabbitMQ 支持多种语言访问，以 Java 为例看下一般使用 RabbitMQ 的步骤。

1. maven工程的pom文件中添加依赖
   
   Java

   <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>4.1.0</version> </dependency>

   1 2 3 4 5

   <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>4.1.0</version> </dependency>

2. 消息生产者
   
   Java

   package org.study.rabbitmq; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Producer { public static void main(String[] args) throws IOException, TimeoutException { //创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); factory.setUsername("guest"); factory.setPassword("guest"); //设置 RabbitMQ 地址 factory.setHost("localhost"); //建立到代理服务器到连接 Connection conn = factory.newConnection(); //获得信道 Channel channel = conn.createChannel(); //声明交换器 String exchangeName = "hello-exchange"; channel.exchangeDeclare(exchangeName, "direct", true); String routingKey = "hola"; //发布消息 byte[] messageBodyBytes = "quit".getBytes(); channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes); channel.close(); conn.close(); } }

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32

   package org.study.rabbitmq; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Producer {       public static void main(String[] args) throws IOException, TimeoutException {         //创建连接工厂         ConnectionFactory factory = new ConnectionFactory();         factory.setUsername("guest");         factory.setPassword("guest");         //设置 RabbitMQ 地址         factory.setHost("localhost");         //建立到代理服务器到连接         Connection conn = factory.newConnection();         //获得信道         Channel channel = conn.createChannel();         //声明交换器         String exchangeName = "hello-exchange";         channel.exchangeDeclare(exchangeName, "direct", true);           String routingKey = "hola";         //发布消息         byte[] messageBodyBytes = "quit".getBytes();         channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);           channel.close();         conn.close();     } }

3. 消息消费者
   
   Java

   package org.study.rabbitmq; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setUsername("guest"); factory.setPassword("guest"); factory.setHost("localhost"); //建立到代理服务器到连接 Connection conn = factory.newConnection(); //获得信道 final Channel channel = conn.createChannel(); //声明交换器 String exchangeName = "hello-exchange"; channel.exchangeDeclare(exchangeName, "direct", true); //声明队列 String queueName = channel.queueDeclare().getQueue(); String routingKey = "hola"; //绑定队列，通过键 hola 将队列和交换器绑定起来 channel.queueBind(queueName, exchangeName, routingKey); //消费消息 boolean autoAck = false; String consumerTag = ""; channel.basicConsume(queueName, autoAck, consumerTag, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String routingKey = envelope.getRoutingKey(); String contentType = properties.getContentType(); System.out.println("消费的路由键：" + routingKey); System.out.println("消费的内容类型：" + contentType); long deliveryTag = envelope.getDeliveryTag(); //确认消息 channel.basicAck(deliveryTag, false); System.out.println("消费的消息体内容："); String bodyStr = new String(body, "UTF-8"); System.out.println(bodyStr); } }); while(true) { Thread.sleep(10); } } }

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52

   package org.study.rabbitmq; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer {       public static void main(String[] args) throws IOException, TimeoutException {         ConnectionFactory factory = new ConnectionFactory();         factory.setUsername("guest");         factory.setPassword("guest");         factory.setHost("localhost");         //建立到代理服务器到连接         Connection conn = factory.newConnection();         //获得信道         final Channel channel = conn.createChannel();         //声明交换器         String exchangeName = "hello-exchange";         channel.exchangeDeclare(exchangeName, "direct", true);         //声明队列         String queueName = channel.queueDeclare().getQueue();         String routingKey = "hola";         //绑定队列，通过键 hola 将队列和交换器绑定起来         channel.queueBind(queueName, exchangeName, routingKey);                    //消费消息         boolean autoAck = false;         String consumerTag = "";         channel.basicConsume(queueName, autoAck, consumerTag, new DefaultConsumer(channel) {             @Override             public void handleDelivery(String consumerTag,                                        Envelope envelope,                                        AMQP.BasicProperties properties,                                        byte[] body) throws IOException {                 String routingKey = envelope.getRoutingKey();                 String contentType = properties.getContentType();                 System.out.println("消费的路由键：" + routingKey);                 System.out.println("消费的内容类型：" + contentType);                 long deliveryTag = envelope.getDeliveryTag();                 //确认消息                 channel.basicAck(deliveryTag, false);                 System.out.println("消费的消息体内容：");                 String bodyStr = new String(body, "UTF-8");                 System.out.println(bodyStr);               }         });       while(true) {         Thread.sleep(10);       }     } }

4. 启动 RabbitMQ 服务器
   
   Java

   ./sbin/rabbitmq-server

   1	./sbin/rabbitmq-server

5. 运行 Consumer
   先运行 Consumer ，这样当生产者发送消息的时候能在消费者后端看到消息记录。
6. 运行 Producer
   接着运行 Producer ,发布一条消息，在 Consumer 的控制台能看到接收的消息：

启动第2个节点：

启动第2个节点前建议将 RabbitMQ 默认激活的插件关掉，否则会存在使用了某个插件的端口号冲突，导致节点启动不成功。

现在第2个节点和第1个节点都是独立节点，它们并不知道其他节点的存在。集群中除第一个节点外后加入的节点需要获取集群中的元数据，所以要先停止 Erlang 节点上运行的 RabbitMQ 应用程序，并重置该节点元数据，再加入并且获取集群的元数据，最后重新启动 RabbitMQ 应用程序。

停止第2个节点的应用程序：

重置第2个节点元数据：

第2节点加入第1个节点组成的集群：

启动第2个节点的应用程序

第3个节点的配置过程和第2个节点类似：

RabbitMQ 集群运维

停止某个指定的节点，比如停止第2个节点：

查看节点3的集群状态：