1、实现步骤

1：jdk1.8
2：构建一个maven工程
3：导入rabbitmq的maven依赖
4：启动rabbitmq-server服务
5：定义生产者
6：定义消费者
7：观察消息的在rabbitmq-server服务中的过程

2、构建一个maven工程

image.png

3、导入rabbitmq的maven依赖

3.1、Java原生依赖

  1. <dependency>    
  2.   <groupId>com.rabbitmq</groupId>    
  3.   <artifactId>amqp-client</artifactId>    
  4.   <version>5.10.0</version>
  5. </dependency>

3.2、spring依赖

  1. <dependency>    
  2.   <groupId>org.springframework.amqp</groupId>    
  3.   <artifactId>spring-amqp</artifactId>    
  4.   <version>2.2.5.RELEASE</version></dependency><dependency>    
  5.     <groupId>org.springframework.amqp</groupId>    
  6.     <artifactId>spring-rabbit</artifactId>    
  7.     <version>2.2.5.RELEASE</version>
  8. </dependency>

3.3、springboot依赖

  1. <dependency>    
  2.   <groupId>org.springframework.boot</groupId>    
  3.   <artifactId>spring-boot-starter-amqp</artifactId>
  4. </dependency>

上面根据自己的项目环境进行选择即可。

:::info 番外：rabbitmq和spring同属一个公司开放的产品，所以他们的支持也是非常完善，这也是为什么推荐使用rabbitmq的一个原因。 :::

4、启动rabbitmq-server服务

  1. systemctl start rabbitmq-server或者docker start myrabbit

5、定义生产者

  1. package com.theory.simple;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  7. import java.io.IOException;
  8. import java.util.concurrent.TimeoutException;
  10. public class Producer {
  11.     public static void main(String[] args) {
  12.         // 所有的中间件技术都是支持基于tcp/ip协议基础之上构建新型的协议规范，只不过RabbitMQ遵循的是amqp
  13.         // IP port
  14.         // 1. 创建连接工厂
  15.         ConnectionFactory connectionFactory = new ConnectionFactory();
  16.         // 2. 设置连接属性
  17.         connectionFactory.setHost("192.168.222.111");  // ip地址
  18.         connectionFactory.setPort(5672);  // 端口号
  19.         connectionFactory.setUsername("admin");  // 账号
  20.         connectionFactory.setPassword("admin");  // 密码
  21.         connectionFactory.setVirtualHost("/");  //路由地址
  23.         Connection connection = null;
  24.         Channel channel = null;
  25.         try {
  26.             // 3. 从连接工厂中获取连接 Connection
  27.             connection = connectionFactory.newConnection("生产者");
  28.             // 4. 通过连接获取通道Channel
  29.             channel = connection.createChannel();
  30.             // 5. 通过通道创建交换机，声明队列，绑定关系，如有key，发送消息，接收消息
  31.             String queueName = "queue1";
  33.             /**
  34.              * @params1: 队列的名称，如果队列不存在，则会创建
  35.              * @params2: 是否要持久化 durable=false所谓的持久化消息就是是否存盘，如果为false即非持久化，true是持久化  非持久化也会存盘，但是重启服务器会丢失
  36.              * @params3: 排他性，是否是独占队列，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
  37.              * @params4: 是否自动删除，随着消费者消息完毕消息以后是否把队列自动删除
  38.              * @params5: 携带附属参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
  39.              */
  40.             channel.queueDeclare(queueName,false,false,false,null);
  41.             // 6. 准备发送消息内容
  42.             String message = "Hello theory";
  43.             // 7. 发送消息给队列queue
  44.             /**
  45.              * @params1: 交换机exchange
  46.              * @params2: 队列、路由key
  47.              * @params3: 消息的状态控制，即属性
  48.              * @params4: 消息主题，即发送消息的内容
  49.              */
  50.             // 面试题：可以存在没有交换机的队列吗？不可能，虽然没有指定交换机但是一定存在一个默认的交换机
  51.             channel.basicPublish("",queueName,null,message.getBytes());
  52.             System.out.println("消息发送成功！！！");
  53.         } catch (IOException e) {
  54.             e.printStackTrace();
  55.         } catch (TimeoutException e) {
  56.             e.printStackTrace();
  57.         } finally {
  58.             // 8. 关闭通道
  59.             if (channel != null && channel.isOpen()) {
  60.                 try {
  61.                     channel.close();
  62.                 } catch (IOException e) {
  63.                     e.printStackTrace();
  64.                 } catch (TimeoutException e) {
  65.                     e.printStackTrace();
  66.                 }
  67.             }
  68.             // 9. 释放连接
  69.             if (connection != null && connection.isOpen()) {
  70.                 try {
  71.                     connection.close();
  72.                 } catch (IOException e) {
  73.                     e.printStackTrace();
  74.                 }
  75.             }
  76.         }
  78.     }
  79. }

1：执行发送，这个时候可以在web控制台查看到这个队列queue的信息
RabbitMQ入门案例 - Simple 简单模式 - 图2
RabbitMQ入门案例 - Simple 简单模式 - 图3
2：我们可以进行对队列的消息进行预览和测试如下：
RabbitMQ入门案例 - Simple 简单模式 - 图4
3:进行预览和获取消息进行测试
RabbitMQ入门案例 - Simple 简单模式 - 图5

6、定义消费者

  1. package com.theory.routing;
  3. import com.rabbitmq.client.*;
  5. import java.io.IOException;
  6. import java.util.concurrent.TimeoutException;
  8. public class Consumer {
  9.     public static void main(String[] args) {
  10.         // 所有的中间件技术都是支持基于tcp/ip协议基础之上构建新型的协议规范，只不过RabbitMQ遵循的是amqp
  11.         // IP port
  12.         // 1. 创建连接工厂，并设置参数
  13.         ConnectionFactory connectionFactory = new ConnectionFactory();
  14.         connectionFactory.setHost("192.168.222.111");
  15.         connectionFactory.setPort(5672);
  16.         connectionFactory.setUsername("admin");
  17.         connectionFactory.setPassword("admin");
  18.         connectionFactory.setVirtualHost("/");
  20.         Connection connection = null;
  21.         Channel channel = null;
  22.         try {
  23.             // 2. 创建连接 Connection
  24.             connection = connectionFactory.newConnection("生产者");
  25.             // 3. 通过连接获取通道Channel
  26.             channel = connection.createChannel();
  27.             // 4. 通过通道创建交换机，声明队列，绑定关系，如有key，发送消息，接收消息
  28.             channel.basicConsume("queue1", true, new DeliverCallback() {
  29.                         @Override
  30.                         public void handle(String s, Delivery delivery) throws IOException {
  31.                             System.out.println("收到消息是" + new String(delivery.getBody(), "utf-8"));
  32.                         }
  33.                     }, new CancelCallback() {
  34.                         @Override
  35.                         public void handle(String s) throws IOException {
  36.                             System.out.println("接收失败了...");
  37.                         }
  38.                     }
  39.             );
  41.             System.out.println("开始接收消息");
  42.             System.in.read();
  44.         } catch (IOException e) {
  45.             e.printStackTrace();
  46.         } catch (TimeoutException e) {
  47.             e.printStackTrace();
  48.         } finally {
  49.             // 7. 关闭通道
  50.             if (channel != null && channel.isOpen()) {
  51.                 try {
  52.                     channel.close();
  53.                 } catch (IOException e) {
  54.                     e.printStackTrace();
  55.                 } catch (TimeoutException e) {
  56.                     e.printStackTrace();
  57.                 }
  58.             }
  59.             // 8. 关闭通道
  60.             if (connection != null && connection.isOpen()) {
  61.                 try {
  62.                     connection.close();
  63.                 } catch (IOException e) {
  64.                     e.printStackTrace();
  65.                 }
  66.             }
  67.         }
  69.     }
  70. }

7、观察消息的在rabbitmq-server服务中的过程