RabbitMQ-SpringAMQP使用介绍

RabbitMQ

- 1. Spring AMQP
- - 1.1 引入依赖
  - 1.2 消息发送
  - 1.3 消息接收
  - 1.4 WorkQueue模型
  - - 1.4.1 实例代码
    - 1.4.2 能者多劳
    - 1.4.3 总结
  - 1.5交换机
  - 1.6 Fanout交换机（广播）
  - 1.7 Direct交换机（订阅）
  - 1.8 Topic交换机（通配符订阅）
  - 1.9 声明队列and交换机
  - 1.10 基于注解声明
  - 1.11 消息转换器
  - - 1.11.1 配置JSON转换器
    - 1.11.2 示例

RabbitMQ采用了AMQP协议，因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息，都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。

1. Spring AMQP

RabbitMQ官方提供的Java客户端编码相对复杂，一般生产环境下我们更多会结合Spring来使用。而Spring的官方刚好基于RabbitMQ提供了这样一套消息收发的模板工具：SpringAMQP。并且还基于SpringBoot对其实现了自动装配，使用起来非常方便。

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

SpringAMQP提供了三个功能：

自动声明队列、交换机及其绑定关系
基于注解的监听器模式，异步接收消息
封装了RabbitTemplate工具，用于发送消息

1.1 引入依赖

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

1.2 消息发送

配置MQ地址，在publisher服务的application.yml中添加配置：

spring:
  rabbitmq:
    host: 192.168.1.200 # 虚拟机IP
    port: 5672 # 端口
    virtual-host: /hmall # 虚拟主机
    username: feng # 用户名
    password: 123 # 密码

然后在publisher服务中编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送：

@SpringBootTest
public class SpringAmopTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void SimpleQueue(){
        //队列
        String queueName = "simpleQueue";
        //消息
        String message = "Hello World";
        //发送消息
        rabbitTemplate.convertAndSend(queueName,message);
    }
}

1.3 消息接收

在consumer端的application.xml配置MQ信息：

spring:
  rabbitmq:
    host: 192.168.1.200 # 你的虚拟机IP
    port: 5672 # 端口
    virtual-host: /hmall # 虚拟主机
    username: feng # 用户名
    password: 123321 # 密码

编写监听器

@Component
@Slf4j
public class SpringRabbitListener {

    @RabbitListener(queues = "simpleQueue")
    public void listenSimpleQueue(String message){

        log.info("Simple Queue消息： "+message);

    }
}

重启服务，即可实现监听！

1.4 WorkQueue模型

Work queues，任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。

在这里插入图片描述

1.4.1 实例代码

publisher段代码：

@Test
public void WorkQueue(){
    //队列
    String queueName = "work.queue";
    //消息
    String message = "Hello World_";
    //发送消息50次
    for (int i = 0; i < 50; i++) {
        rabbitTemplate.convertAndSend(queueName,message+i);
    }

}

consumer段消费代码：

//消费者1
@RabbitListener(queues = "work.queue")
public void listenWorkQueue1(String message){

    System.out.println("消费者1处理任务{："+message+"}"+ LocalTime.now());
}
//消费者2
@RabbitListener(queues = "work.queue")
public void listenWorkQueue2(String message){

    System.err.println("消费者2处理任务{："+message+"}"+ LocalTime.now());
}

work queue模式，消息分配模式默认是平均分配，每个消费者处理的任务数量是平均的

也就是上面案例中，消息队列一共五十条消息，消费者1和消费者2，每人会处理25条。

1.4.2 能者多劳

如果想修改work模式的分配模式，比如想按性能分配，性能好的，多处理任务，那些就需要添加配置：

application.xml：

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

这样，50条消息，由消费者1和消费者2 按性能快慢决定执行数量的多少，谁先执行完成当前任务，就可以去执行消息队列中的下一个任务，没有预分配机制；

1.4.3 总结

work queue模型的使用：

多个消费者绑定到一个消息队列，可以加快消息处理速度；
同一条消息只会被一个消费者处理；
通过设置prefetch来控制消费者预取的消息数量，处理完一条再处理下一条，实现能者多劳；

1.5交换机

交换机exchange，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。

在这里插入图片描述

Publisher：生产者，不再发送消息到队列中，而是发给交换机
Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。
Consumer：消费者，与以前一样，订阅队列，没有变化

交换机的作用：

接收publisher发送的消息
将消息按照规则路由到与之绑定的队列
不能缓存消息，路由失败，消息丢失

交换机的类型有四种：

Fanout：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
Direct：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列
Topic：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符
Headers：头匹配，基于MQ的消息头匹配，用的较少。

1.6 Fanout交换机（广播）

FanoutExchange会将消息路由到每个绑定的队列。

在这里插入图片描述

可以有多个队列
每个队列都要绑定到Exchange（交换机）
生产者发送的消息，只能发送到交换机
交换机把消息发送给绑定过的所有队列
订阅队列的消费者都能拿到消息

1.7 Direct交换机（订阅）

基于RoutingKey（路由key）发送给订阅了消息的队列。

在这里插入图片描述

在Direct模型下：

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

Direct交换机与Fanout交换机差异：

Fanout交换机将消息路由给每一个与之绑定的队列
Direct交换机根据RoutingKey判断路由给哪个队列
如果多个队列具有相同的RoutingKey，则与Fanout功能类似

1.8 Topic交换机（通配符订阅）

Topic与direct相似，都是通过RoutingKey绑定，按照RoutingKey进行路由，只不过Topic类型的Exchange在绑定bandingKey时，可以使用通配符配置：

通配符规则：

#：匹配一个或多个词
*：匹配不多不少恰好1个词

举例：

item.#：能够匹配item.spu.insert 或者 item.spu
item.*：只能匹配item.spu

Direct交换机与Topic交换机的差异：

Topic交换机接收的消息RoutingKey必须是多个单词，以 . 分割
Topic交换机与队列绑定时的bindingKey可以指定通配符
#：代表0个或多个词
*：代表1个词

1.9 声明队列and交换机

fanout交换机

package com.itheima.consumer.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.Configuration;

@Configuration
public class FanoutConfig {

    /**
     * 声明Fanout交换机
     * @return Fanout交换机类型
     */
    public FanoutExchange fanoutExchange() {
         return new FanoutExchange("fanoutExchange", true, false);
    }

    /**
     * 声明队列
     * @return Queue类型
     */
    public Queue queue() {
        return new Queue("queue", true, false, false);
    }

    /**
     * 队列绑定到交换机
     * @param queue
     * @param fanoutExchange
     * @return
     */
    public Binding binding(Queue queue ,FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }
}

Direct交换机

package com.itheima.consumer.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DirectConfig {

    /**
     * 声明交换机
     * @return Direct类型交换机
     */
    @Bean
    public DirectExchange directExchange(){
        return ExchangeBuilder.directExchange("hmall.direct").build();
    }

    /**
     * 第1个队列
     */
    @Bean
    public Queue directQueue1(){
        return new Queue("direct.queue1");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
    }
    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
    }

    /**
     * 第2个队列
     */
    @Bean
    public Queue directQueue2(){
        return new Queue("direct.queue2");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue2).to(directExchange).with("red");
    }
    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
    }
}

1.10 基于注解声明

Direct模式

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

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

Topic模式：

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

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

1.11 消息转换器

Spring的消息发送代码接收的消息体是一个Object，而在数据传输时，它会把发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。

只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：

数据体积过大
有安全漏洞
可读性差

1.11.1 配置JSON转换器

在publisher和consumer两个服务中都引入依赖：

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
    <version>2.9.10</version>
</dependency>

注意，如果项目中引入了spring-boot-starter-web依赖，则无需再次引入Jackson依赖。

配置消息转换器，在publisher和consumer两个服务的启动类中添加一个Bean即可：

@Bean
public MessageConverter messageConverter(){
    // 1.定义消息转换器
    Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
    // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
    jackson2JsonMessageConverter.setCreateMessageIds(true);
    return jackson2JsonMessageConverter;
}

消息转换器中添加的messageId可以便于我们将来做幂等性判断。

1.11.2 示例

publisher

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void SimpleQueue(){
        //队列
        String queueName = "simpleQueue";
        //消息
        Map<String,Object> map = new HashMap<>(2);
        map.put("name", "xiaoming");
        map.put("age", 18);
//        String message = "测试消息！！！";
        //发送消息
        rabbitTemplate.convertAndSend(queueName,map);
    }

consumer

@RabbitListener(queues = "simpleQueue")
public void listenSimpleQueue(Map<String, Object> map) throws InterruptedException {

    log.info("Simple Queue消息： "+map);
    System.out.println("sout=============:"+map.toString());
}
w HashMap<>(2);
        map.put("name", "xiaoming");
        map.put("age", 18);
//        String message = "测试消息！！！";
        //发送消息
        rabbitTemplate.convertAndSend(queueName,map);
    }

consumer

@RabbitListener(queues = "simpleQueue")
public void listenSimpleQueue(Map<String, Object> map) throws InterruptedException {

    log.info("Simple Queue消息： "+map);
    System.out.println("sout=============:"+map.toString());
}