SpringAMQP

目录

利用SpringAMQP实现HelloWorld中的基础消息队列功能:

1．在父工程中引入spring-amqp的依赖

2．在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列

3．在consumer服务中编写消费逻辑，绑定simple.queue这个队列

模拟WorkQueue，实现一个队列绑定多个消费者：

1．在publisher服务中定义测试方法，每秒产生50条消息，发送到simple.queue

2．在consumer服务中定义两个消息监听者，都监听simple.queue队列，消费者1每秒处理50条消息，消费者2每秒处理10条消息

发布和订阅：

发布订阅-Fanout Exchange

利用SpringAMQP演示FanoutExchange的使用

1．在consumer服务中，利用代码声明队列、交换机，并将两者绑定

2．在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2

3．在publisher中编写测试方法，向itcast.fanout发送消息

发布订阅-DirectExchange

1．利用@RabbitListener声明Exchange、Queue、RoutingKey，在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2

2．在publisher中编写测试方法，向itcast. direct发送消息

发布订阅-TopicExchange

1．并利用@RabbitListener声明Exchange、Queue、RoutingKey，在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2

2．在publisher中编写测试方法，向itcast. topic发送消息

测试发送Object类型消息

?编辑?消息转换器

---

什么是AMQP?
·应用间消息通信的一种协议，与语言和平台无关。

SpringAMQP如何发送消息?
·引入amqp的starter依赖

·配置RabbitMQ地址
·利用RabbitTemplate的convertAndSend方法

---

SpringAMQP如何接收消息?

·引入amqp的starter依赖

·配置RabbitMQ地址
·定义类，添加@Component注解
·类中声明方法，添加@RabbitListener注解，方法参数就时消息

注意:消息一旦消费就会从队列删除，RabbitMQ没有消息回溯功能

---

Work模型的使用:
·多个消费者绑定到一个队列，同一条消息只会被一个
消费者处理
·通过设置prefetch来控制消费者预取的消息数量

---

交换机的作用是什么?

·接收publisher发送的消息
·将消息按照规则路由到与之绑定的队列

·不能缓存消息，路由失败，消息丢失
·FanoutExchange的会将消息路由到每个绑定的队列

声明队列、交换机、绑定关系的Bean是什么?

·Queue
·FanoutExchange

·Binding

---

描述下Direct交换机与Fanout交换机的差异?

·Fanout交换机将消息路由给每一个与之绑定的队列

·Direct交换机根据RoutingKey判断路由给哪个队列
·如果多个队列具有相同的RoutingKey，则与Fanout功能类似

基于@RabbitListener注解声明队列和交换机有哪些常见注解?

·@Queue
·@Exchange

---

SpringAMQP中消息的序列化和反序列化是怎么实现的?

·利用MessageConverter实现的，默认是JDK的序列化

·注意发送方与接收方必须使用相同的MessageConverter

SpringAMQP官网：https://spring.io/projects/spring-amqp

什么是AMQP：?

Advanced Message Queuing Protocol，是用于在应用程序或之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。

什么是SpringAMQP：

Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。包含两部分，其中spring-amqp是基础抽象，spring-rabbit是底层的默认实现。

所有流程的文件相对位置：

利用SpringAMQP实现HelloWorld中的基础消息队列功能:

1．在父工程中引入spring-amqp的依赖

在mq-demo的pom文件导入：

<!--AMQP依赖，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

<?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>

    <groupId>cn.itcast.demo</groupId>
    <artifactId>mq-demo</artifactId>
    <version>1.0-SNAPSHOT</version>
    <modules>
        <module>publisher</module>
        <module>consumer</module>
    </modules>
    <packaging>pom</packaging>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.9.RELEASE</version>
        <relativePath/>
    </parent>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.24</version>
        </dependency>
        <!--AMQP依赖，包含RabbitMQ-->
        <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>

2．在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列

1．在publisher服务中编写application.yml，添加mq连接信息:

spring:
  rabbitmq:
    host: xxx.xxx.xxx.xxx #RabbitMQ的虚拟机ip
    port: 5672 #Rabbit的端口
    username: itcast
    password: 123321
    virtual-host: / #该username的虚拟主机

logging:
  pattern:
    dateformat: MM-dd HH:mm:ss:SSS
spring:
  rabbitmq:
    host: xxx.xxx.xxx.xxx #RabbitMQ的虚拟机ip
    port: 5672 #Rabbit的端口
    username: itcast
    password: 123321
    virtual-host: / #该username的虚拟主机

?2．在publisher服务中新建一个测试类，编写测试方法:

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2SimpleQueue() {
        String queueName = "simple.queue";
        String message = "hello, spring amqp!";
        rabbitTemplate.convertAndSend(queueName,message);
    }

}

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.RabbitAccessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2SimpleQueue() {
        String queueName = "simple.queue";
        String message = "hello, spring amqp!";
        rabbitTemplate.convertAndSend(queueName,message);
    }

    

}

运行该测试用例，会看到成功运行

此时在RabbitMQ页面中可以看到simple.queue队列有一条数据

点击simple.queue，进入该服务中看，能看到发送的hello,spring amqp

3．在consumer服务中编写消费逻辑，绑定simple.queue这个队列

1．在consumer服务中编写application.yml，添加mq连接信息:

spring:
  rabbitmq:
    host: xxx.xxx.xxx.xxx #RabbitMQ的虚拟机ip
    port: 5672 #Rabbit的端口
    username: itcast
    password: 123321
    virtual-host: / #该username的虚拟主机

logging:
  pattern:
    dateformat: MM-dd HH:mm:ss:SSS
spring:
  rabbitmq:
    host: XXX.XXX.XXX.XXX #RabbitMQ的虚拟机ip
    port: 5672 #Rabbit的端口
    username: itcast
    password: 123321
    virtual-host: / #该username的虚拟主机

?2.在consumer服务中新建一个类，编写消费逻辑:

@Component
public class SpringRabbitListener {
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueue(String msg) {
        System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
    }

}

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueue(String msg) {
        System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
    }

}

?运行该代码，可以看到运行成功，并接受到了simple.queue中的消息

并且消息也就没了?

模拟WorkQueue，实现一个队列绑定多个消费者：

1．在publisher服务中定义测试方法，每秒产生50条消息，发送到simple.queue

处理消息的发送，发送50条，并有20毫米的延迟，不然太快了

@Test
public void testSendMessage2WorkQueue() throws InterruptedException {
    String queueName = "simple.queue";
    String message = "hello, message__!";
    for(int i=1;i<=50;i++){
        rabbitTemplate.convertAndSend(queueName,message + i);
        Thread.sleep(20);
    }
}

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.RabbitAccessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2SimpleQueue() {
        String queueName = "simple.queue";
        String message = "hello, spring amqp!";
        rabbitTemplate.convertAndSend(queueName,message);
    }

    @Test
    public void testSendMessage2WorkQueue() throws InterruptedException {
        String queueName = "simple.queue";
        String message = "hello, message__!";
        for(int i=1;i<=50;i++){
            rabbitTemplate.convertAndSend(queueName,message + i);
            Thread.sleep(20);
        }
    }

}

2．在consumer服务中定义两个消息监听者，都监听simple.queue队列，消费者1每秒处理50条消息，消费者2每秒处理10条消息

处理消息的接收，用两个接受者，看看处理的情况

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
    System.out.println("消费者1接收到消息：【"+ msg + "】" + LocalTime.now());
    Thread.sleep(20);
}

????????

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
    System.err.println("消费者2....接收到消息：【"+ msg + "】" + LocalTime.now());
    Thread.sleep(200);
}

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueue(String msg) {
        System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
    }


    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1接收到消息：【"+ msg + "】" + LocalTime.now());
        Thread.sleep(20);
    }


    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2....接收到消息：【"+ msg + "】" + LocalTime.now());
        Thread.sleep(200);
    }

}

启动consumer的启动类，再启动publisher的测试类。

此时可以看到，消费者1和消费者2分别处理偶数和奇数的消息。

说明就算消费者2的处理时间是消费者1的十倍，但是它仍然是和消费者1分别拿了一半的消息。

修改consumer的application.yml文件

listener:
  simple:
    prefetch: 1 #每次只取一条，等处理完成后再继续取

logging:
  pattern:
    dateformat: MM-dd HH:mm:ss:SSS
spring:
  rabbitmq:
    host: 192.168.80.129 #RabbitMQ的虚拟机ip
    port: 5672 #Rabbit的端口
    username: itcast
    password: 123321
    virtual-host: / #该username的虚拟主机
    listener:
      simple:
        prefetch: 1 #每次只取一条，等处理完成后再继续取

重新再运行consumer的启动类和publisher的测试类。此时可以看到消费者2不再跟消费者1一人一半消息了。

发布和订阅：

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange(交换机)。

常见exchange类型包括:
1、Fanout:广播
2、Direct:路由

3、Topic:话题

注意: exchange负责消息路由，而不是存储，路由失败则消息丢失

发布订阅-Fanout Exchange

Fanout Exchange会将接收到的消息路由到每一个跟其绑定的queue

利用SpringAMQP演示FanoutExchange的使用

1．在consumer服务中，利用代码声明队列、交换机，并将两者绑定

在consumer服务常见一个类，添加@Configuration注解，并声明FanoutExchange、Queue和绑定关系对象Binding，代码如下:

@Configuration
public class FanoutConfig {
    //交换机名字:itcast.fanout
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }

    //队列名称： fanout.queue1
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
    //绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    //队列名称： fanout.queue2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
    //绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }

  
}

package cn.itcast.mq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutConfig {
    //交换机名字:itcast.fanout
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }

    //队列名称： fanout.queue1
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
    //绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    //队列名称： fanout.queue2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
    //绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }

  
}

启动consumer的启动类

可以在RabbitMQ的网站上看到这两个队列

可以在RabbitMQ的网站上看到该交换机

点击该交换机，可以看到里面绑定了两个队列

2．在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg)  {
    System.out.println("消费者接收到fanout.queue1的消息：【"+ msg + "】");
}

@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg)  {
    System.out.println("消费者接收到fanout.queue2的消息：【"+ msg + "】");
}

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {
//    @RabbitListener(queues = "simple.queue")
//    public void listenSimpleQueue(String msg) {
//        System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
//    }
//

//    @RabbitListener(queues = "simple.queue")
//    public void listenWorkQueue1(String msg) throws InterruptedException {
//        System.out.println("消费者1接收到消息：【"+ msg + "】" + LocalTime.now());
//        Thread.sleep(20);
//    }

//    @RabbitListener(queues = "simple.queue")
//    public void listenWorkQueue2(String msg) throws InterruptedException {
//        System.err.println("消费者2....接收到消息：【"+ msg + "】" + LocalTime.now());
//        Thread.sleep(200);
//    }

    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg)  {
        System.out.println("消费者接收到fanout.queue1的消息：【"+ msg + "】");
    }

    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg)  {
        System.out.println("消费者接收到fanout.queue2的消息：【"+ msg + "】");
    }

}

3．在publisher中编写测试方法，向itcast.fanout发送消息

@Test
public void testSendFanoutExchange(){
    //交换机名称
    String exchangeName = "itcast.fanout";
    //消息
    String message = "hello,everyone!";
    //发送消息
    rabbitTemplate.convertAndSend(exchangeName,"",message);
}

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.RabbitAccessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2SimpleQueue() {
        String queueName = "simple.queue";
        String message = "hello, spring amqp!";
        rabbitTemplate.convertAndSend(queueName,message);
    }

    @Test
    public void testSendMessage2WorkQueue() throws InterruptedException {
        String queueName = "simple.queue";
        String message = "hello, message__!";
        for(int i=1;i<=50;i++){
            rabbitTemplate.convertAndSend(queueName,message + i);
            Thread.sleep(20);
        }
    }

    @Test
    public void testSendFanoutExchange(){
        //交换机名称
        String exchangeName = "itcast.fanout";
        //消息
        String message = "hello,everyone!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName,"",message);
    }

}

?运行publisher的测试类和consumer的启动类，可以看到两个队列都收到了

发布订阅-DirectExchange

Direct Exchange 会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式（routes)。

1、每一个Queue都与Exchange设置一个BindingKey

2、发布者发送消息时，指定消息的RoutingKey
3、Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

1．利用@RabbitListener声明Exchange、Queue、RoutingKey，在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue1"),
        exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
        key = {"red","blue"}))
public void ListenDirectQueue1(String msg){
    System.out.println("消费者接收到direct.queue1的消息：【"+ msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue2"),
        exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
        key = {"red","yellow"}))
public void ListenDirectQueue2(String msg){
    System.out.println("消费者接收到direct.queue2的消息：【"+ msg + "】");
}

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {
//    @RabbitListener(queues = "simple.queue")
//    public void listenSimpleQueue(String msg) {
//        System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
//    }
//

//    @RabbitListener(queues = "simple.queue")
//    public void listenWorkQueue1(String msg) throws InterruptedException {
//        System.out.println("消费者1接收到消息：【"+ msg + "】" + LocalTime.now());
//        Thread.sleep(20);
//    }

//    @RabbitListener(queues = "simple.queue")
//    public void listenWorkQueue2(String msg) throws InterruptedException {
//        System.err.println("消费者2....接收到消息：【"+ msg + "】" + LocalTime.now());
//        Thread.sleep(200);
//    }

    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg)  {
        System.out.println("消费者接收到fanout.queue1的消息：【"+ msg + "】");
    }

    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg)  {
        System.out.println("消费者接收到fanout.queue2的消息：【"+ msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red","blue"}))
    public void ListenDirectQueue1(String msg){
        System.out.println("消费者接收到direct.queue1的消息：【"+ msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red","yellow"}))
    public void ListenDirectQueue2(String msg){
        System.out.println("消费者接收到direct.queue2的消息：【"+ msg + "】");
    }

}

运行consumer的启动类，在RabbitMQ的网页中可以看到多出了itcast.direct的交换机

队列也多出两个direct.queue

?可以看到两个队列分别绑定了两个key

2．在publisher中编写测试方法，向itcast. direct发送消息

@Test
public void testSendDirectExchange(){
    //交换机名称
    String exchangeName = "itcast.direct";
    //消息
    String message = "hello,red!";
    //发送消息
    rabbitTemplate.convertAndSend(exchangeName,"red",message);
}

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.RabbitAccessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2SimpleQueue() {
        String queueName = "simple.queue";
        String message = "hello, spring amqp!";
        rabbitTemplate.convertAndSend(queueName,message);
    }

    @Test
    public void testSendMessage2WorkQueue() throws InterruptedException {
        String queueName = "simple.queue";
        String message = "hello, message__!";
        for(int i=1;i<=50;i++){
            rabbitTemplate.convertAndSend(queueName,message + i);
            Thread.sleep(20);
        }
    }

    @Test
    public void testSendFanoutExchange(){
        //交换机名称
        String exchangeName = "itcast.fanout";
        //消息
        String message = "hello,everyone!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName,"",message);
    }

    @Test
    public void testSendDirectExchange(){
        //交换机名称
        String exchangeName = "itcast.direct";
        //消息
        String message = "hello,red!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName,"red",message);
    }


}

通过修改message的内容，绑定了不同key的队列会收到不同的消息

发布订阅-TopicExchange

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且以.分割。

Queue与Exchange指定BindingKey时可以使用通配符:

#:代指0个或多个单词
*:代指一个单词

1．并利用@RabbitListener声明Exchange、Queue、RoutingKey，在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "topic.queue1"),
        exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),
        key = "china.#"
))
public void ListenTopictQueue1(String msg){
    System.out.println("消费者接收到topic.queue1的消息：【"+ msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "topic.queue2"),
        exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),
        key = "#.news"
))
public void ListenTopictQueue2(String msg){
    System.out.println("消费者接收到topic.queue2的消息：【"+ msg + "】");
}

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {
//    @RabbitListener(queues = "simple.queue")
//    public void listenSimpleQueue(String msg) {
//        System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
//    }
//

//    @RabbitListener(queues = "simple.queue")
//    public void listenWorkQueue1(String msg) throws InterruptedException {
//        System.out.println("消费者1接收到消息：【"+ msg + "】" + LocalTime.now());
//        Thread.sleep(20);
//    }

//    @RabbitListener(queues = "simple.queue")
//    public void listenWorkQueue2(String msg) throws InterruptedException {
//        System.err.println("消费者2....接收到消息：【"+ msg + "】" + LocalTime.now());
//        Thread.sleep(200);
//    }

    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg)  {
        System.out.println("消费者接收到fanout.queue1的消息：【"+ msg + "】");
    }

    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg)  {
        System.out.println("消费者接收到fanout.queue2的消息：【"+ msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red","blue"}))
    public void ListenDirectQueue1(String msg){
        System.out.println("消费者接收到direct.queue1的消息：【"+ msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red","yellow"}))
    public void ListenDirectQueue2(String msg){
        System.out.println("消费者接收到direct.queue2的消息：【"+ msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void ListenTopictQueue1(String msg){
        System.out.println("消费者接收到topic.queue1的消息：【"+ msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void ListenTopictQueue2(String msg){
        System.out.println("消费者接收到topic.queue2的消息：【"+ msg + "】");
    }

   
}

运行后，在RabbitMQ网页中，可以看到，队列中出现了两个topic.queue

在交换机中也出现了itcas.topic

?点击该交换机，可以看到绑定了两个topic.queue，并且绑定了不同的key。

2．在publisher中编写测试方法，向itcast. topic发送消息

@Test
public void testSendTopicExchange(){
    //交换机名称
    String exchangeName = "itcast.topic";
    //消息
    String message = "天气!";
    //发送消息
    rabbitTemplate.convertAndSend(exchangeName,"china.weather",message);
}

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.RabbitAccessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2SimpleQueue() {
        String queueName = "simple.queue";
        String message = "hello, spring amqp!";
        rabbitTemplate.convertAndSend(queueName,message);
    }

    @Test
    public void testSendMessage2WorkQueue() throws InterruptedException {
        String queueName = "simple.queue";
        String message = "hello, message__!";
        for(int i=1;i<=50;i++){
            rabbitTemplate.convertAndSend(queueName,message + i);
            Thread.sleep(20);
        }
    }

    @Test
    public void testSendFanoutExchange(){
        //交换机名称
        String exchangeName = "itcast.fanout";
        //消息
        String message = "hello,everyone!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName,"",message);
    }

    @Test
    public void testSendDirectExchange(){
        //交换机名称
        String exchangeName = "itcast.direct";
        //消息
        String message = "hello,red!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName,"red",message);
    }

    @Test
    public void testSendTopicExchange(){
        //交换机名称
        String exchangeName = "itcast.topic";
        //消息
        String message = "天气!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName,"china.weather",message);
    }

}

可以看到，通过修改不同的message，会由不同的队列接收

测试发送Object类型消息

说明:在SpringAMQP的发送方法中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息，SpringAMQP会帮我们序列化为字节后发送。

编写一个队列，在FanoutConfig里写

@Bean
public Queue objectQueue(){
    return new Queue("object.queue");
}

package cn.itcast.mq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutConfig {
    //交换机名字:itcast.fanout
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }

    //队列名称： fanout.queue1
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
    //绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    //队列名称： fanout.queue2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
    //绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }

    @Bean
    public Queue objectQueue(){
        return new Queue("object.queue");
    }
}

?运行consumer的启动类，在RabbitMQ网页中可以看到object.queue的队列

编写消息

@Test
public void testSendObjectQueue(){
    Map<String,Object> msg = new HashMap<>();
    msg.put("name","小明");
    msg.put("age",21);
    rabbitTemplate.convertAndSend("object.queue",msg);
}

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.RabbitAccessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2SimpleQueue() {
        String queueName = "simple.queue";
        String message = "hello, spring amqp!";
        rabbitTemplate.convertAndSend(queueName,message);
    }

    @Test
    public void testSendMessage2WorkQueue() throws InterruptedException {
        String queueName = "simple.queue";
        String message = "hello, message__!";
        for(int i=1;i<=50;i++){
            rabbitTemplate.convertAndSend(queueName,message + i);
            Thread.sleep(20);
        }
    }

    @Test
    public void testSendFanoutExchange(){
        //交换机名称
        String exchangeName = "itcast.fanout";
        //消息
        String message = "hello,everyone!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName,"",message);
    }

    @Test
    public void testSendDirectExchange(){
        //交换机名称
        String exchangeName = "itcast.direct";
        //消息
        String message = "hello,red!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName,"red",message);
    }

    @Test
    public void testSendTopicExchange(){
        //交换机名称
        String exchangeName = "itcast.topic";
        //消息
        String message = "天气!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName,"china.weather",message);
    }

    @Test
    public void testSendObjectQueue(){
        Map<String,Object> msg = new HashMap<>();
        msg.put("name","小明");
        msg.put("age",21);
        rabbitTemplate.convertAndSend("object.queue",msg);
    }

}

启动该测试类，在RabbitMQ网页中可以看到object.queue已经有新的消息

点击该队列，看到消息变成了奇怪的序列化

?消息转换器

Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。
如果要修改只需要定义一个MessageConverter类型的Bean即可。推荐用JSON方式序列化

在mq-demo的pom文件中添加：

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

<?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>

    <groupId>cn.itcast.demo</groupId>
    <artifactId>mq-demo</artifactId>
    <version>1.0-SNAPSHOT</version>
    <modules>
        <module>publisher</module>
        <module>consumer</module>
    </modules>
    <packaging>pom</packaging>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.9.RELEASE</version>
        <relativePath/>
    </parent>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.24</version>
        </dependency>
        <!--AMQP依赖，包含RabbitMQ-->
        <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>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
    </dependencies>
</project>

在publisher的启动类声明一个bean

@Bean
public MessageConverter messageConverter(){
    return new Jackson2JsonMessageConverter();
}

package cn.itcast.mq;

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class PublisherApplication {
    public static void main(String[] args) {
        SpringApplication.run(PublisherApplication.class);
    }

    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

?启动测试类，可以看到该内容是正常的了

在consumer中，同样是声明bean，再写一个监听的函数

@Bean
public MessageConverter messageConverter(){
    return new Jackson2JsonMessageConverter();
}

package cn.itcast.mq;

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class ConsumerApplication {
    public static void main(String[] args) {
        SpringApplication.run(ConsumerApplication.class, args);
    }

    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

@RabbitListener(queues = "object.queue")
public void listenObjectQueue(Map<String,Object> msg){
    System.out.println("接收到object.queue的消息：" + msg);
}

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {
//    @RabbitListener(queues = "simple.queue")
//    public void listenSimpleQueue(String msg) {
//        System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
//    }
//

//    @RabbitListener(queues = "simple.queue")
//    public void listenWorkQueue1(String msg) throws InterruptedException {
//        System.out.println("消费者1接收到消息：【"+ msg + "】" + LocalTime.now());
//        Thread.sleep(20);
//    }

//    @RabbitListener(queues = "simple.queue")
//    public void listenWorkQueue2(String msg) throws InterruptedException {
//        System.err.println("消费者2....接收到消息：【"+ msg + "】" + LocalTime.now());
//        Thread.sleep(200);
//    }

    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg)  {
        System.out.println("消费者接收到fanout.queue1的消息：【"+ msg + "】");
    }

    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg)  {
        System.out.println("消费者接收到fanout.queue2的消息：【"+ msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red","blue"}))
    public void ListenDirectQueue1(String msg){
        System.out.println("消费者接收到direct.queue1的消息：【"+ msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red","yellow"}))
    public void ListenDirectQueue2(String msg){
        System.out.println("消费者接收到direct.queue2的消息：【"+ msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void ListenTopictQueue1(String msg){
        System.out.println("消费者接收到topic.queue1的消息：【"+ msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void ListenTopictQueue2(String msg){
        System.out.println("消费者接收到topic.queue2的消息：【"+ msg + "】");
    }

    @RabbitListener(queues = "object.queue")
    public void listenObjectQueue(Map<String,Object> msg){
        System.out.println("接收到object.queue的消息：" + msg);
    }
}

?启动consumer的启动类，可以看到成功接收到消息

代码文件点击下载https://pan.baidu.com/s/1wXL7uWpLXu7X-OWuUZRUKA?pwd=tnxl?上一篇：RabbitMQ