【消息中间件RabbitMQ 】：简单模式和Work queues_rabbitmq 简单模式和工作队列模式

在上一篇我们成功在服务器上搭建了 RabbitMQ，那就相当于中间红框部分。

想要使用 RabbitMQ，还必须用到两个角色——Produces 和 Consumer ，所以下面我们具体是实现 RabbitMQ 发送和接收消息。

1、工作模式

RabbitMQ 提供了六种工作模式，分别用于不同场景，如下所示：


六种模式分别对应：简单模式、work queues 、Publish/Subscribe 发布与订阅模式、Routing路由模式、Topics 主题模式、RPC 远程调用模式。

图中的 P 表示 消息生产者，C 表示消息消费者，X 表示交换机，红色的表示队列。

2、简单模式

简单模式是 RabbitMQ 中最简单的一种模式。这种模式下，只有两个角色，一个消息生产者，一个消息消费者。

2.1、消息提供方

首先通过 Maven 创建一个 java 项目，然后通过如下步骤完成消息生产者的搭建：

• 引入依赖：引入 RabbitMQ 所需依赖

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

• 编写代码：在主函数中编写代码

  public static void main(String[] args) throws Exception {
     //1.创建连接工厂
     ConnectionFactory factory = new ConnectionFactory();
     //2. 设置参数
     factory.setHost("ip地址");//ip  默认值 localhost
     factory.setPort(5672); //端口 默认值 5672
     factory.setVirtualHost("/");//虚拟机 默认值/
     factory.setUsername("用户名");//用户名 默认 guest
     factory.setPassword("密码");//密码 默认值 guest
     //3. 创建连接 Connection
     Connection connection = factory.newConnection();
     //4. 创建Channel
     Channel channel = connection.createChannel();
     //5. 创建队列Queue，因为是简单模式，所以不需要使用交换机
     channel.queueDeclare("hello",false,false,false,null);
     //6. 发送消息
     String str = "hello world !!! hello rabbitmq !!!";
     channel.basicPublish("","hello_world",null,str.getBytes());
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19

因为 RabbitMQ 消息发送端口 5672，所以我们需要在阿里云控制台添加有关安全组规则，然后执行主函数。执行成功之后，在 RabbitMQ 的 web 管理页面可以查消息是否发送成功：

这样一个消息提供方就创建完成。下面我们对主函数中的某些代码进行解释：

• ConnectionFactory factory = new ConnectionFactory();，创建一个建连接工厂，设置一些参数，来指定 RabbitMQ Server 的地址等信息。

• Connection connection = factory.newConnection();，创建一个连接，用于与 RabbitMQ Server 进行连接。

• Channel channel = connection.createChannel();，创建一个管道。

• channel.queueDeclare("hello",false,false,false,null);，创建了一个队列。queueDeclare 方法有 5 个参数，具体意思如下：

  • 队列名称
  • 队列是否持久化，既是否保存消息。消息可以存储在内存和磁盘上，默认是存在内存中的（值为false）
  • 队列是否只供一个消费者进行消费，默认情况下只允许一个消费者消费（值为false）
  • 当最后一个消费者端开连接以后，该队列是否自动删除，true 为自动删除
  • 其他参数，因为快速使用的缘故在此填为 null。

• channel.basicPublish("","hello",null,ste.getBytes());，发送消息。basicPublish 方法有 4 个参数，具体意思如下：

  • 交换机名称，因为没有使用交换机，所以填“ ”
  • 路由名称，
  • 其他的参数信息
  • 发送消息数据

2.2、消息消费方

为了与消息生产者进行区分，所以通过 Maven 创建一个新的项目。搭建消息消费方的步骤和消息生产者差不多：

• 引入依赖：同上
• 编写代码：在主函数中编写代码

  public static void main(String[] args) throws  Exception{
      //1.创建连接工厂
      ConnectionFactory factory = new ConnectionFactory();
      //2. 设置参数
      factory.setHost("120.26.166.77");//ip  默认值 localhost
      factory.setPort(5672); //端口 默认值 5672
      factory.setVirtualHost("/");//虚拟机 默认值/
      factory.setUsername("long");//用户名 默认 guest
      factory.setPassword("czl212121");//密码 默认值 guest
      //3. 创建连接 Connection
      Connection connection = factory.newConnection();
      //4. 创建Channel
      Channel channel = connection.createChannel();
      //5. 创建队列Queue，其实在创建队列之前，我们先要使用Exchange（交换机），但是由于是入门案例，就可以省略。
      channel.queueDeclare("hello",false,false,false,null);
  
      //推送的消息如何进行消费的接口回调
      DeliverCallback deliverCallback=(consumerTag, delivery)->{
          System.out.println("consumerTag："+consumerTag);
          System.out.println(new String(delivery.getBody()));
      };
  
      //取消消费的一个回调接口 如在消费的时候队列被删除掉了
      CancelCallback cancelCallback=(consumerTag)->{
          System.out.println("消息消费中断");
      };
  
      channel.basicConsume("hello",true,deliverCallback,cancelCallback);
  }
  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

执行主函数之后：

在 web 管理页面查看，队列中的消息已经被消息完：

在主函数中的代码与消息提供方中的代码相差不大，唯一的区别是由消息发送变为了消息接受。

channel.basicConsume("hello",true,deliverCallback,cancelCallback);表示接受消息，basicConsume 中的四个参数分别表示：

• 队列名称
• 消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
• 消费成功的回调函数
• 消费失败的回调函数

3、Work queues 工作队列模式

工作队列也是 RabbitMQ 中的一种模式，它和简单模式相差不大，代码几乎是一样的，唯一的区别是 工作队列模式中存在多个消息消费者。

为了得到具体的效果，我们对简单模式中的案例进行简单的修改：

• 消息生产者：将消息的发送改为在控制台输入发送，如下只贴出了修改之后的部分代码

  //String str = "hello world !!! hello rabbitmq";
  //channel.basicPublish("","hello",null,str.getBytes());
  //用于替代上述代码，完成由控制台输入消息内容
  Scanner scanner = new Scanner(System.in);
  while (!scanner.hasNext("#")){
      String str = scanner.nextLine();
      channel.basicPublish("","hello",null,str.getBytes());
      System.out.println("消息已发送："+str);
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9

• 消息消费者：同时运行两个消息消费者

当消息生产者发送消息时，会产生如下效果：

从上图可以看出，在工作队列模式中，队列是通过轮训分发消息，简单理解就是雨露均沾，一人一条消息。