消息队列之RabbitMQ

内容纲要

一、消息队列概述

1.1 消息队列MQ

MQ全称为Message Queue,消息队列是应用程序和应用程序之间的通信方法。

  • 为什么使用MQ
    在项目中,可将一些无需即时返回且耗时的操作提取出来,进行异步处理,而这种异步处理的方式大大的节省
    了服务器的请求响应时间,从而提高了系统的吞吐量
  • 开发中消息队列通常有如下应用场景:
    1、任务异步处理
    将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。提高了应用程序的响应时
    间。
    2、应用程序解耦合
    MQ相当于一个中介,生产方通过MQ与消费方交互,它将应用程序进行解耦合

1.2 AMQP和JMS

MQ是消息通信的模型;实现MQ的大致有两种主流方式:AMQP、JMS。

1.2.1 AMQP

AMQP是一种协议,更准确的说是一种binary wire-level protocol(链接协议)。这是其和JMS的本质差别,AMQP
不从API层进行限定,而是直接定义网络交换的数据格式。

1.2.2. JMS

JMS即Java消息服务(JavaMessage Service)应用程序接口,是一个Java平台中关于面向消息中间件(MOM)的
API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。

1.2.3. AMQP 与 JMS 区别

  • JMS是定义了统一的接口,来对消息操作进行统一;AMQP是通过规定协议来统一数据交互的格式
  • JMS限定了必须使用Java语言;AMQP只是协议,不规定实现方式,因此是跨语言的。
  • JMS规定了两种消息模式;而AMQP的消息模式更加丰富

1.3 消息队列产品

市场上常见的消息队列有如下:

  • ActiveMQ:基于JMS
  • ZeroMQ:基于C语言开发
  • RabbitMQ:基于AMQP协议,erlang语言开发,稳定性好
  • RocketMQ:基于JMS,阿里巴巴产品
  • Kafka:类似MQ的产品;分布式消息系统,高吞吐量

1.4. RabbitMQ

RabbitMQ是由erlang语言开发,基于AMQP(Advanced Message Queue 高级消息队列协议)协议实现的消息队
列,它是一种应用程序之间的通信方法,消息队列在分布式系统开发中应用非常广泛。
RabbitMQ官方地址:http://www.rabbitmq.com/
RabbitMQ提供了6种模式:简单模式,work模式,Publish/Subscribe发布与订阅模式,Routing路由模式,Topics
主题模式,RPC远程调用模式(远程调用,不太算MQ;不作介绍);
官网对应模式介绍:https://www.rabbitmq.com/getstarted.html

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219171810161.png

二、安装及配置RabbitMQ

2.1 安装说明

详细查看 资料/软件/安装windows RabbitMQ.pdf文档

2.2 用户以及Virtual Hosts配置

2.2.1 用户角色

RabbitMQ在安装好后,可以访问http://localhost:15672 ;其自带了guest/guest的用户名和密码;如果需要创建自
定义用户;那么也可以登录管理界面后,如下操作

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219172037422

角色说明:
1、 超级管理员(administrator)
可登陆管理控制台,可查看所有的信息,并且可以对用户,策略(policy)进行操作。
2、 监控者(monitoring)
可登陆管理控制台,同时可以查看rabbitmq节点的相关信息(进程数,内存使用情况,磁盘使用情况等)
3、 策略制定者(policymaker)
可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。
4、 普通管理者(management)
仅可登陆管理控制台,无法看到节点信息,也无法对策略进行管理。
5、 其他
无法登陆管理控制台,通常就是普通的生产者和消费者。

2.2.2. Virtual Hosts配置

像mysql拥有数据库的概念并且可以指定用户对库和表等操作的权限。RabbitMQ也有类似的权限管理;在RabbitMQ
中可以虚拟消息服务器Virtual Host,每个Virtual Hosts相当于一个相对独立的RabbitMQ服务器,每个VirtualHost
之间是相互隔离的。exchange、queue、message不能互通。 相当于mysql的db。Virtual Name一般以/开头。

  1. 创建Virtual Hosts

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219172138456

  1. 设置Virtual Hosts

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219172217012
http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219172233556

三、 RabbitMQ入门

3.1. 搭建示例工程

3.1.1. 创建工程

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219172351671
http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219172405555

3.1.2. 添加依赖

往heima-rabbitmq的pom.xml文件中添加如下依赖:

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

3.2. 编写生产者

编写消息生产者com.itheima.rabbitmq.simple.Producer

package com.itheima.rabbitmq.simple;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
  static final String QUEUE_NAME = "simple_queue";
  public static void main(String[] args) throws Exception {
    //创建连接工厂
    ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址;默认为 localhost
    connectionFactory.setHost("localhost");
    //连接端口;默认为 5672
    connectionFactory.setPort(5672);
    //虚拟主机名称;默认为 /
    connectionFactory.setVirtualHost("/itcast");
    //连接用户名;默认为guest
    connectionFactory.setUsername("heima");
    //连接密码;默认为guest
    connectionFactory.setPassword("heima");
    //创建连接
    Connection connection = connectionFactory.newConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    // 要发送的信息
    String message = "你好;小兔子!";
    /**
    * 参数1:交换机名称,如果没有指定则使用默认Default Exchage
    * 参数2:路由key,简单模式可以传递队列名称
    * 参数3:消息其它属性
    * 参数4:消息内容
    */
    channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
    System.out.println("已发送消息:" + message);
    // 关闭资源
    channel.close();
    connection.close();
 }
}

在执行上述的消息发送之后;可以登录rabbitMQ的管理控制台,可以发现队列和其消息:

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219172551478
http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219172607123

3.3. 编写消费者

抽取创建connection的工具类com.itheima.rabbitmq.util.ConnectionUtil;

package com.itheima.rabbitmq.util;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class ConnectionUtil {
  public static Connection getConnection() throws Exception {
    //创建连接工厂
    ConnectionFactory connectionFactory = new ConnectionFactory();
    //主机地址;默认为 localhost
    connectionFactory.setHost("localhost");
    //连接端口;默认为 5672
    connectionFactory.setPort(5672);
    //虚拟主机名称;默认为 /
    connectionFactory.setVirtualHost("/itcast");
    //连接用户名;默认为guest
    connectionFactory.setUsername("heima");
    //连接密码;默认为guest
    connectionFactory.setPassword("heima");
    //创建连接
    return connectionFactory.newConnection();
 }
}

编写消息的消费者com.itheima.rabbitmq.simple.Consumer

package com.itheima.rabbitmq.simple;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer {
  public static void main(String[] args) throws Exception {
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
    //创建消费者;并设置消息处理
    DefaultConsumer consumer = new DefaultConsumer(channel){
      @Override
      /**
      * consumerTag 消息者标签,在channel.basicConsume时候可以指定
      * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志
(收到消息失败后是否需要重新发送)
      * properties 属性信息
      * body 消息
      */
      public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
        //路由key
        System.out.println("路由key为:" + envelope.getRoutingKey());
        //交换机
        System.out.println("交换机为:" + envelope.getExchange());
        //消息id
        System.out.println("消息id为:" + envelope.getDeliveryTag());
        //收到的消息
        System.out.println("接收到的消息为:" + new String(body, "utf-8"));
     }
   };
    //监听消息
    /**
    * 参数1:队列名称
    * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消
息,设置为false则需要手动确认
    * 参数3:消息接收到后回调
    */
    channel.basicConsume(Producer.QUEUE_NAME, true, consumer);
    //不关闭资源,应该一直监听消息
    //channel.close();
    //connection.close();
 }
}

3.4. 小结

上述的入门案例中中其实使用的是如下的简单模式:

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219172841354

在上图的模型中,有以下概念:

  • P:生产者,也就是要发送消息的程序
  • C:消费者:消息的接受者,会一直等待消息到来。
  • queue:消息队列,图中红色部分。类似一个邮箱,可以缓存消息;生产者向其中投递消息,消费者从其中取
    出消息。

在rabbitMQ中消费者是一定要到某个消息队列中去获取消息的

四、RabbitMQ工作模式

4.1. Work queues工作队列模式

4.1.1. 模式说明

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219173037178

Work Queues 与入门程序的 简单模式 相比,多了一个或一些消费端,多个消费端共同消费同一个队列中的消息。
应用场景:对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

4.1.2. 代码

Work Queues 与入门程序的 简单模式 的代码是几乎一样的;可以完全复制,并复制多一个消费者进行多个消费者同
时消费消息的测试。
1)生产者

package com.itheima.rabbitmq.work;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
  static final String QUEUE_NAME = "work_queue";
  public static void main(String[] args) throws Exception {
    //创建连接
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    for (int i = 1; i <= 30; i++) {
      // 发送信息
      String message = "你好;小兔子!work模式--" + i;
      /**
      * 参数1:交换机名称,如果没有指定则使用默认Default Exchage
      * 参数2:路由key,简单模式可以传递队列名称
      * 参数3:消息其它属性
      * 参数4:消息内容
      */
      channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
      System.out.println("已发送消息:" + message);
   }
    // 关闭资源
    channel.close();
    connection.close();
 }
}

2)消费者1

package com.itheima.rabbitmq.work;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer1 {
public static void main(String[] args) throws Exception {
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
    //一次只能接收并处理一个消息
    channel.basicQos(1);
    //创建消费者;并设置消息处理
    DefaultConsumer consumer = new DefaultConsumer(channel){
      @Override
      /**
      * consumerTag 消息者标签,在channel.basicConsume时候可以指定
      * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志
(收到消息失败后是否需要重新发送)
      * properties 属性信息
      * body 消息
      */
      public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
        try {
          //路由key
          System.out.println("路由key为:" + envelope.getRoutingKey());
          //交换机
          System.out.println("交换机为:" + envelope.getExchange());
          //消息id
          System.out.println("消息id为:" + envelope.getDeliveryTag());
          //收到的消息
          System.out.println("消费者1-接收到的消息为:" + new String(body, "utf-
8"));
          Thread.sleep(1000);
          //确认消息
          channel.basicAck(envelope.getDeliveryTag(), false);
       } catch (InterruptedException e) {
          e.printStackTrace();
       }
     }
   };
    //监听消息
    /**
    * 参数1:队列名称
    * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消
息,设置为false则需要手动确认
    * 参数3:消息接收到后回调
    */
    channel.basicConsume(Producer.QUEUE_NAME, false, consumer);
 }
}

3)消费者2

package com.itheima.rabbitmq.work;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer2 {
  public static void main(String[] args) throws Exception {
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
    //一次只能接收并处理一个消息
    channel.basicQos(1);
    //创建消费者;并设置消息处理
    DefaultConsumer consumer = new DefaultConsumer(channel){
      @Override
      /**
      * consumerTag 消息者标签,在channel.basicConsume时候可以指定
      * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志
(收到消息失败后是否需要重新发送)
      * properties 属性信息
      * body 消息
      */
      public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
        try {
          //路由key
          System.out.println("路由key为:" + envelope.getRoutingKey());
          //交换机
          System.out.println("交换机为:" + envelope.getExchange());
          //消息id
          System.out.println("消息id为:" + envelope.getDeliveryTag());
          //收到的消息
          System.out.println("消费者2-接收到的消息为:" + new String(body, "utf-
8"));
          Thread.sleep(1000);
          //确认消息
          channel.basicAck(envelope.getDeliveryTag(), false);
       } catch (InterruptedException e) {
          e.printStackTrace();
       }
     }
   };
    //监听消息
    /**
    * 参数1:队列名称
    * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消
息,设置为false则需要手动确认
    * 参数3:消息接收到后回调
    */
    channel.basicConsume(Producer.QUEUE_NAME, false, consumer);
 }
}

4.1.3. 测试

启动两个消费者,然后再启动生产者发送消息;到IDEA的两个消费者对应的控制台查看是否竞争性的接收到消息。

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219173418176
http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219173431028

4.1.4. 小结

在一个队列中如果有多个消费者,那么消费者之间对于同一个消息的关系是竞争的关系。

4.2 订阅模式类型

订阅模式示例图:

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219173531522

前面2个案例中,只有3个角色:

  • P:生产者,也就是要发送消息的程序
  • C:消费者:消息的接受者,会一直等待消息到来。
  • queue:消息队列,图中红色部分

而在订阅模型中,多了一个exchange角色,而且过程略有变化:

  • P:生产者,也就是要发送消息的程序,但是不再发送到队列中,而是发给X(交换机)
  • C:消费者,消息的接受者,会一直等待消息到来。
  • Queue:消息队列,接收消息、缓存消息。
  • Exchange:交换机,图中的X。一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给
    某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。Exchange有常
    见以下3种类型:

    • Fanout:广播,将消息交给所有绑定到交换机的队列
    • Direct:定向,把消息交给符合指定routing key 的队列
    • Topic:通配符,把消息交给符合routing pattern(路由模式) 的队列

Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有
符合路由规则的队列,那么消息会丢失!

4.3. Publish/Subscribe发布与订阅模式

4.3.1. 模式说明

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219173903053

发布订阅模式:

1、每个消费者监听自己的队列。

2、生产者将消息发给broker,由交换机将消息转发到绑定此交换机的每个队列,每个绑定交换机的队列都将接收 到消息

4.3.2. 代码

1)生产者

package com.itheima.rabbitmq.ps;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
* 发布与订阅使用的交换机类型为:fanout
*/
public class Producer {
  //交换机名称
  static final String FANOUT_EXCHAGE = "fanout_exchange";
  //队列名称
  static final String FANOUT_QUEUE_1 = "fanout_queue_1";
  //队列名称
  static final String FANOUT_QUEUE_2 = "fanout_queue_2";
  public static void main(String[] args) throws Exception {
    //创建连接
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    /**
    * 声明交换机
    * 参数1:交换机名称
    * 参数2:交换机类型,fanout、topic、direct、headers
    */
    channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);
    channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);
     //队列绑定交换机
    channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, "");
    channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, "");
    for (int i = 1; i <= 10; i++) {
      // 发送信息
      String message = "你好;小兔子!发布订阅模式--" + i;
      /**
      * 参数1:交换机名称,如果没有指定则使用默认Default Exchage
      * 参数2:路由key,简单模式可以传递队列名称
      * 参数3:消息其它属性
      * 参数4:消息内容
      */
      channel.basicPublish(FANOUT_EXCHAGE, "", null, message.getBytes());
      System.out.println("已发送消息:" + message);
   }
    // 关闭资源
    channel.close();
    connection.close();
 }
}

2)消费者1

package com.itheima.rabbitmq.ps;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer1 {
  public static void main(String[] args) throws Exception {
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    //声明交换机
    channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null);
    //队列绑定交换机
    channel.queueBind(Producer.FANOUT_QUEUE_1, Producer.FANOUT_EXCHAGE, "");
    //创建消费者;并设置消息处理
    DefaultConsumer consumer = new DefaultConsumer(channel){
      @Override
      /**
      * consumerTag 消息者标签,在channel.basicConsume时候可以指定
      * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志
(收到消息失败后是否需要重新发送)
      * properties 属性信息
      * body 消息
      */
      public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
        //路由key
        System.out.println("路由key为:" + envelope.getRoutingKey());
        //交换机
        System.out.println("交换机为:" + envelope.getExchange());
        //消息id
        System.out.println("消息id为:" + envelope.getDeliveryTag());
        //收到的消息
        System.out.println("消费者1-接收到的消息为:" + new String(body, "utf-
8"));
     }
   };
    //监听消息
    /**
    * 参数1:队列名称
    * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消
息,设置为false则需要手动确认
    * 参数3:消息接收到后回调
    */
    channel.basicConsume(Producer.FANOUT_QUEUE_1, true, consumer);
 }
}

3)消费者2

package com.itheima.rabbitmq.ps;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer2 {
  public static void main(String[] args) throws Exception {
  Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    //声明交换机
    channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null);
    //队列绑定交换机
    channel.queueBind(Producer.FANOUT_QUEUE_2, Producer.FANOUT_EXCHAGE, "");
    //创建消费者;并设置消息处理
    DefaultConsumer consumer = new DefaultConsumer(channel){
      @Override
      /**
      * consumerTag 消息者标签,在channel.basicConsume时候可以指定
      * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志
(收到消息失败后是否需要重新发送)
      * properties 属性信息
      * body 消息
      */
      public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
        //路由key
        System.out.println("路由key为:" + envelope.getRoutingKey());
        //交换机
        System.out.println("交换机为:" + envelope.getExchange());
        //消息id
        System.out.println("消息id为:" + envelope.getDeliveryTag());
        //收到的消息
        System.out.println("消费者2-接收到的消息为:" + new String(body, "utf-
8"));
     }
   };
    //监听消息
    /**
    * 参数1:队列名称
    * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消
息,设置为false则需要手动确认
    * 参数3:消息接收到后回调
    */
    channel.basicConsume(Producer.FANOUT_QUEUE_2, true, consumer);
 }
}

4.3.3. 测试

启动所有消费者,然后使用生产者发送消息;在每个消费者对应的控制台可以查看到生产者发送的所有消息;到达广
播的效果。
在执行完测试代码后,其实到RabbitMQ的管理后台找到 Exchanges 选项卡,点击 fanout_exchange 的交换机,可
以查看到如下的绑定:

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219174156567

4.3.4. 小结

交换机需要与队列进行绑定,绑定之后;一个消息可以被多个消费者都收到。

发布订阅模式与工作队列模式的区别
1、工作队列模式不用定义交换机,而发布/订阅模式需要定义交换机。
2、发布/订阅模式的生产方是面向交换机发送消息,工作队列模式的生产方是面向队列发送消息(底层使用默认交换
机)。
3、发布/订阅模式需要设置队列和交换机的绑定,工作队列模式不需要设置,实际上工作队列模式会将队列绑 定到
默认的交换机 。

4.4. Routing路由模式

4.4.1. 模式说明

路由模式特点:

  • 队列与交换机的绑定,不能是任意绑定了,而是要指定一个 RoutingKey (路由key)
  • 消息的发送方在 向 Exchange发送消息时,也必须指定消息的 RoutingKey
  • Exchange不再把消息交给每一个绑定的队列,而是根据消息的 Routing Key 进行判断,只有队列的
    Routingkey 与消息的 Routing key 完全一致,才会接收到消息

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219174411206

图解:

  • P:生产者,向Exchange发送消息,发送消息时,会指定一个routing key。
  • X:Exchange(交换机),接收生产者的消息,然后把消息递交给 与routing key完全匹配的队列
  • C1:消费者,其所在队列指定了需要routing key 为 error 的消息
  • C2:消费者,其所在队列指定了需要routing key 为 info、error、warning 的消息

4.4.2. 代码

在编码上与 Publish/Subscribe发布与订阅模式 的区别是交换机的类型为:Direct,还有队列绑定交换机的时候需
要指定routing key。

1)生产者

package com.itheima.rabbitmq.routing;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
* 路由模式的交换机类型为:direct
*/
public class Producer {
  //交换机名称
  static final String DIRECT_EXCHAGE = "direct_exchange";
  //队列名称
  static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
  //队列名称
  static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";
  public static void main(String[] args) throws Exception {
    //创建连接
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    /**
    * 声明交换机
    * 参数1:交换机名称
    * 参数2:交换机类型,fanout、topic、direct、headers
    */
    channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);
    channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);
    //队列绑定交换机
    channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert");
    channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update");
    // 发送信息
    String message = "新增了商品。路由模式;routing key 为 insert " ;
    /**
    * 参数1:交换机名称,如果没有指定则使用默认Default Exchage
    * 参数2:路由key,简单模式可以传递队列名称
    * 参数3:消息其它属性
    * 参数4:消息内容
    */
    channel.basicPublish(DIRECT_EXCHAGE, "insert", null, message.getBytes());
    System.out.println("已发送消息:" + message);
    // 发送信息
    message = "修改了商品。路由模式;routing key 为 update" ;
    /**
    * 参数1:交换机名称,如果没有指定则使用默认Default Exchage
    * 参数2:路由key,简单模式可以传递队列名称
    * 参数3:消息其它属性
    * 参数4:消息内容
    */
    channel.basicPublish(DIRECT_EXCHAGE, "update", null, message.getBytes());
    System.out.println("已发送消息:" + message);
    // 关闭资源
    channel.close();
    connection.close();
 }
}

2)消费者1

package com.itheima.rabbitmq.routing;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer1 {
  public static void main(String[] args) throws Exception {
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    //声明交换机
    channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null);
    //队列绑定交换机
    channel.queueBind(Producer.DIRECT_QUEUE_INSERT, Producer.DIRECT_EXCHAGE,
"insert");
    //创建消费者;并设置消息处理
    DefaultConsumer consumer = new DefaultConsumer(channel){
      @Override
      /**
      * consumerTag 消息者标签,在channel.basicConsume时候可以指定
      * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志
(收到消息失败后是否需要重新发送)
      * properties 属性信息
      * body 消息
      */
      public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
        //路由key
        System.out.println("路由key为:" + envelope.getRoutingKey());
        //交换机
        System.out.println("交换机为:" + envelope.getExchange());
        //消息id
        System.out.println("消息id为:" + envelope.getDeliveryTag());
        //收到的消息
        System.out.println("消费者1-接收到的消息为:" + new String(body, "utf-
8"));
     }
   };
    //监听消息
    /**
    * 参数1:队列名称
    * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消
息,设置为false则需要手动确认
    * 参数3:消息接收到后回调
    */
    channel.basicConsume(Producer.DIRECT_QUEUE_INSERT, true, consumer);
 }
}

3)消费者2

package com.itheima.rabbitmq.routing;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer2 {
  public static void main(String[] args) throws Exception {
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
     Channel channel = connection.createChannel();
    //声明交换机
    channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(Producer.DIRECT_QUEUE_UPDATE, true, false, false, null);
    //队列绑定交换机
    channel.queueBind(Producer.DIRECT_QUEUE_UPDATE, Producer.DIRECT_EXCHAGE,
"update");
    //创建消费者;并设置消息处理
    DefaultConsumer consumer = new DefaultConsumer(channel){
      @Override
      /**
      * consumerTag 消息者标签,在channel.basicConsume时候可以指定
      * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志
(收到消息失败后是否需要重新发送)
      * properties 属性信息
      * body 消息
      */
      public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
        //路由key
        System.out.println("路由key为:" + envelope.getRoutingKey());
        //交换机
        System.out.println("交换机为:" + envelope.getExchange());
        //消息id
        System.out.println("消息id为:" + envelope.getDeliveryTag());
        //收到的消息
        System.out.println("消费者2-接收到的消息为:" + new String(body, "utf-
8"));
     }
   };
    //监听消息
    /**
    * 参数1:队列名称
    * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消
息,设置为false则需要手动确认
    * 参数3:消息接收到后回调
    */
    channel.basicConsume(Producer.DIRECT_QUEUE_UPDATE, true, consumer);
 }
}

4.4.3. 测试

启动所有消费者,然后使用生产者发送消息;在消费者对应的控制台可以查看到生产者发送对应routing key对应队
列的消息;到达按照需要接收的效果。
在执行完测试代码后,其实到RabbitMQ的管理后台找到 Exchanges 选项卡,点击 direct_exchange 的交换机,可
以查看到如下的绑定:

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219175124726

4.4.4. 小结

Routing模式要求队列在绑定交换机时要指定routing key,消息会转发到符合routing key的队列。

4.5. Topics通配符模式

4.5.1. 模式说明

Topic 类型与 Direct 相比,都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型 Exchange
以让队列在绑定 Routing key 的时候使用通配符!

Routingkey 一般都是有一个或多个单词组成,多个单词之间以”.”分割,例如: item.insert

通配符规则:

#:匹配一个或多个单词

*:匹配不多不少恰好一个词

举例:

item.# :能够匹配 item.insert.abc 或者 item.insert
item.* :只能匹配 item.insert

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219175531802
http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200219175546381

图解:

  • 红色Queue:绑定的是 usa.# ,因此凡是以 usa. 开头的 routing key 都会被匹配到
  • 黄色Queue:绑定的是 #.news ,因此凡是以 .news 结尾的 routing key 都会被匹配

4.5.2. 代码

1)生产者
使用topic类型的Exchange,发送消息的routing key有3种: item.insertitem.updateitem.delete

package com.itheima.rabbitmq.topic;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
* 通配符Topic的交换机类型为:topic
*/
public class Producer {
  //交换机名称
  static final String TOPIC_EXCHAGE = "topic_exchange";
  //队列名称
  static final String TOPIC_QUEUE_1 = "topic_queue_1";
  //队列名称
  static final String TOPIC_QUEUE_2 = "topic_queue_2";
  public static void main(String[] args) throws Exception {
    //创建连接
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    /**
    * 声明交换机
    * 参数1:交换机名称
    * 参数2:交换机类型,fanout、topic、topic、headers
    */
    channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
    // 发送信息
    String message = "新增了商品。Topic模式;routing key 为 item.insert " ;
    channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, message.getBytes());
    System.out.println("已发送消息:" + message);
    // 发送信息
    message = "修改了商品。Topic模式;routing key 为 item.update" ;
    channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes());
    System.out.println("已发送消息:" + message);
    // 发送信息
    message = "删除了商品。Topic模式;routing key 为 item.delete" ;
    channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, message.getBytes());
    System.out.println("已发送消息:" + message);
    // 关闭资源
    channel.close();
    connection.close();
 }
}

2)消费者1
接收两种类型的消息:更新商品和删除商品

package com.itheima.rabbitmq.topic;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer1 {
  public static void main(String[] args) throws Exception {
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    //声明交换机
    channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null);
    //队列绑定交换机
    channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE,
"item.update");
    channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE,
"item.delete");
    //创建消费者;并设置消息处理
    DefaultConsumer consumer = new DefaultConsumer(channel){
      @Override
      /**
      * consumerTag 消息者标签,在channel.basicConsume时候可以指定
      * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志
(收到消息失败后是否需要重新发送)
      * properties 属性信息
      * body 消息
      */
      public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
        //路由key
        System.out.println("路由key为:" + envelope.getRoutingKey());
        //交换机
        System.out.println("交换机为:" + envelope.getExchange());
        //消息id
        System.out.println("消息id为:" + envelope.getDeliveryTag());
        //收到的消息
        System.out.println("消费者1-接收到的消息为:" + new String(body, "utf-
8"));
     }
   };
    //监听消息
    /**
    * 参数1:队列名称
    * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消
息,设置为false则需要手动确认
    * 参数3:消息接收到后回调
    */
    channel.basicConsume(Producer.TOPIC_QUEUE_1, true, consumer);
 }
}

3)消费者2
接收所有类型的消息:新增商品,更新商品和删除商品。

package com.itheima.rabbitmq.topic;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer2 {
  public static void main(String[] args) throws Exception {
    Connection connection = ConnectionUtil.getConnection();
    // 创建频道
    Channel channel = connection.createChannel();
    //声明交换机
    channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
    // 声明(创建)队列
    /**
    * 参数1:队列名称
    * 参数2:是否定义持久化队列
    * 参数3:是否独占本次连接
    * 参数4:是否在不使用的时候自动删除队列
    * 参数5:队列其它参数
    */
    channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null);
    //队列绑定交换机
    channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.*");
    //创建消费者;并设置消息处理
    DefaultConsumer consumer = new DefaultConsumer(channel){
      @Override
      /**
      * consumerTag 消息者标签,在channel.basicConsume时候可以指定
      * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志
(收到消息失败后是否需要重新发送)
      * properties 属性信息
      * body 消息
      */
      public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
        //路由key
        System.out.println("路由key为:" + envelope.getRoutingKey());
        //交换机
        System.out.println("交换机为:" + envelope.getExchange());
        //消息id
        System.out.println("消息id为:" + envelope.getDeliveryTag());
        //收到的消息
        System.out.println("消费者2-接收到的消息为:" + new String(body, "utf-
8"));
     }
   };
    //监听消息
    /**
    * 参数1:队列名称
    * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消
息,设置为false则需要手动确认
    * 参数3:消息接收到后回调
    */
    channel.basicConsume(Producer.TOPIC_QUEUE_2, true, consumer);
 }
}

4.5.3. 测试

启动所有消费者,然后使用生产者发送消息;在消费者对应的控制台可以查看到生产者发送对应routing key对应队
列的消息;到达按照需要接收的效果;并且这些routing key可以使用通配符。

在执行完测试代码后,其实到RabbitMQ的管理后台找到 Exchanges 选项卡,点击 topic_exchange 的交换机,可
以查看到如下的绑定:

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200220085640038

4.5.4. 小结

Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 和 Routing路由模式 的功能;只是Topic在配置routing
key 的时候可以使用通配符,显得更加灵活

4.6. 模式总结

RabbitMQ工作模式:

1、简单模式 HelloWorld 一个生产者、一个消费者,不需要设置交换机(使用默认的交换
机)
2、工作队列模式 Work Queue 一个生产者、多个消费者(竞争关系),不需要设置交换机(使用默认的交换机)
3、发布订阅模式 Publish/subscribe 需要设置类型为fanout的交换机,并且交换机和队列进行绑定,当发送消息到
交换机后,交换机会将消息发送到绑定的队列
4、路由模式 Routing 需要设置类型为direct的交换机,交换机和队列进行绑定,并且指定routing key,当发送消息
到交换机后,交换机会根据routing key将消息发送到对应的队列
5、通配符模式 Topic 需要设置类型为topic的交换机,交换机和队列进行绑定,并且指定通配符方式的routing key,
当发送消息到交换机后,交换机会根据routing key将消息发送到对应的队列

五、Spring Boot整合RabbitMQ

5.1. 简介

在Spring项目中,可以使用Spring-Rabbit去操作RabbitMQ https://github.com/spring-projects/spring-amqp
尤其是在spring boot项目中只需要引入对应的amqp启动器依赖即可,方便的使用RabbitTemplate发送消息,使用
注解接收消息。
一般在开发过程中:

生产者工程:

  1. application.yml文件配置RabbitMQ相关信息;
  2. 在生产者工程中编写配置类,用于创建交换机和队列,并进行绑定
  3. 注入RabbitTemplate对象,通过RabbitTemplate对象发送消息到交换机

消费者工程:

  1. application.yml文件配置RabbitMQ相关信息
  2. 创建消息处理类,用于接收队列中的消息并进行处理

5.2. 搭建生产者工程

5.2.1. 创建工程

创建生产者工程springboot-rabbitmq-producer

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200220085953896

5.2.2. 添加依赖

修改pom.xml文件内容为如下:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.1.4.RELEASE</version>
  </parent>
  <groupId>com.itheima</groupId>
  <artifactId>springboot-rabbitmq-producer</artifactId>
  <version>1.0-SNAPSHOT</version>
  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
  </dependencies>
</project>

5.2.3. 启动类

package com.itheima.rabbitmq;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class ProducerApplication {
  public static void main(String[] args) {
    SpringApplication.run(ProducerApplication.class);
 }
}

5.2.4. 配置RabbitMQ

1)配置文件
创建application.yml,内容如下:

spring:
rabbitmq:
 host: localhost
 port: 5672
 virtual-host: /itcast
 username: heima
 password: heima

2)绑定交换机和队列
创建RabbitMQ队列与交换机绑定的配置类com.itheima.rabbitmq.config.RabbitMQConfig

package com.itheima.rabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
  //交换机名称
  public static final String ITEM_TOPIC_EXCHANGE = "item_topic_exchange";
  //队列名称
  public static final String ITEM_QUEUE = "item_queue";
  //声明交换机
  @Bean("itemTopicExchange")
  public Exchange topicExchange(){
    return
ExchangeBuilder.topicExchange(ITEM_TOPIC_EXCHANGE).durable(true).build();
 }
  //声明队列
  @Bean("itemQueue")
  public Queue itemQueue(){
    return QueueBuilder.durable(ITEM_QUEUE).build();
 }
  //绑定队列和交换机
  @Bean
  public Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue,
                  @Qualifier("itemTopicExchange") Exchange
exchange){
    return BindingBuilder.bind(queue).to(exchange).with("item.#").noargs();
 }
}

5.3. 搭建消费者工程

5.3.1. 创建工程

创建消费者工程springboot-rabbitmq-consumer

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200220090338433

5.3.2. 添加依赖

修改pom.xml文件内容为如下:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.1.4.RELEASE</version>
  </parent>
  <groupId>com.itheima</groupId>
  <artifactId>springboot-rabbitmq-consumer</artifactId>
  <version>1.0-SNAPSHOT</version>
  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
  </dependencies>
</project>

5.3.3. 启动类

package com.itheima.rabbitmq;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class ConsumerApplication {
  public static void main(String[] args) {
    SpringApplication.run(ConsumerApplication.class);
 }
}

5.3.4. 配置RabbitMQ

创建application.yml,内容如下:

spring:
rabbitmq:
 host: localhost
 port: 5672
 virtual-host: /itcast
 username: heima
 password: heima

5.3.5. 消息监听处理类

编写消息监听器com.itheima.rabbitmq.listener.MyListener

package com.itheima.rabbitmq.listener;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class MyListener {
  /**
  * 监听某个队列的消息
  * @param message 接收到的消息
  */
  @RabbitListener(queues = "item_queue")
  public void myListener1(String message){
    System.out.println("消费者接收到的消息为:" + message);
 }
}

5.4. 测试

在生产者工程springboot-rabbitmq-producer中创建测试类,发送消息:

package com.itheima.rabbitmq;
import com.itheima.rabbitmq.config.RabbitMQConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMQTest {
  @Autowired
  private RabbitTemplate rabbitTemplate;
  @Test
  public void test(){
    rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE,
"item.insert", "商品新增,routing key 为item.insert");
    rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE,
"item.update", "商品修改,routing key 为item.update");
    rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE,
"item.delete", "商品删除,routing key 为item.delete");
 }
}

先运行上述测试程序(交换机和队列才能先被声明和绑定),然后启动消费者;在消费者工程springboot-rabbitmq-
consumer中控制台查看是否接收到对应消息。
另外;也可以在RabbitMQ的管理控制台中查看到交换机与队列的绑定:

http://cdn.imodou.com.cn/wp-content/uploads/2021/04/image-20200220090600405

赞(2) 打赏
未经允许不得转载:魔豆IT » 消息队列之RabbitMQ
分享到: 更多 (0)

评论 1

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址
  1. #1

    work 模式下不竞争资源 是第一个执行完再执行第二个

    liyongzhe6个月前 (04-20)回复