SpringCloud-AMQP

标签：AMQP 队列 SpringCloud --- 交换机消息 msg public

SpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。

SpringAmqp官方地址：https://spring.io/projects/spring-amqp

SpringAMQP提供了三个功能：

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

父工程pom.xml

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

application.yml

spring:
  rabbitmq:
    host: 192.168.88.130
    port: 5672
    username: lmcool
    password: 1234
    virtual-host: /
Appdata:
  rabbitMQ-name: demo-mq

Basic Queue 简单队列模型

publisher.sendMessage2demomq
利用RabbitTemplate实现消息发送

package com.lmcode.mq.spring;

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.beans.factory.annotation.Value;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@SpringBootTest
@RunWith(SpringRunner.class)
public class SpringAMQPTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Value("${Appdata.rabbitMQ-name}")
    private String mqname;

    @Test
    public void sendMessage2demomq(){
        String message = "hello springamqp";
        rabbitTemplate.convertAndSend(mqname,message);
    }
}

consumer.listenSimpleQueueMessage
消费者监听消息

package com.lmcode.mq.listener;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class SpringRabbitListener {
    @RabbitListener(queues = "demo-mq")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("成功接收消息---【" + msg + "】");
    }
}

Work Queue 工作队列模型

消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。

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

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

consumer.application.yml
控制预取消息的上限，根据消费者的能力处理消息，不是平均分配

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

publisher.sendMessage2demomqWorkQueue
向队列中不停发送消息，模拟消息堆积。

    @Test
    public void sendMessage2demomqWorkQueue() throws InterruptedException {
        String message = "hello, message_";
        for (int i = 0; i < 50; i++) {
            rabbitTemplate.convertAndSend(mqname, message + i);
            Thread.sleep(20);
        }
    }

consumer.SpringRabbitListener
线程阻塞，模拟任务耗时

    @RabbitListener(queues = "demo-mq")
    public void listenWorkQueueMessage1(String msg) throws InterruptedException {
        System.out.println("成功接收消息---【" + msg + "】---消费者1---WorkQueue" + LocalTime.now());
        Thread.sleep(20);
    }

    @RabbitListener(queues = "demo-mq")
    public void listenWorkQueueMessage2(String msg) throws InterruptedException {
        System.err.println("成功接收消息---【" + msg + "】---消费者2---WorkQueue" + LocalTime.now());
        Thread.sleep(200);
    }

发布、订阅模型

Exchange：交换机一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：

Fanout：广播，将消息交给所有绑定到交换机的队列
Direct：定向，把消息交给符合指定routing key 的队列
Topic：通配符，把消息交给符合routing pattern（路由模式）的队列

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

发布、订阅模型-Fanout【扇出、广播】

广播模式的消息发送流程：

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

交换机的作用：

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

声明队列和交换机并绑定

Spring提供了一个接口Exchange，来表示所有不同类型的交换机：

consumer.config

package com.lmcode.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 {
    // 定义交换机，队列，绑定队列到交换机
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("lm.fanoutExchange");
    }
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("lm.fanoutQueue1");
    }
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("lm.fanoutQueue2");
    }
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

publisher.sendMessage2demomqFanoutQueue
消息发送给交换机，参数：交换机名，路由规则，消息

    @Test
    public void sendMessage2demomqFanoutQueue(){
        String exchangeName = "lm.fanoutExchange";
        String message = "hello, message";
        rabbitTemplate.convertAndSend(exchangeName,"",message);
    }

consumer.SpringRabbitListener

    @RabbitListener(queues = "lm.fanoutQueue1")
    public void listenFanoutQueueMessage1(String msg) throws InterruptedException {
        System.out.println("成功接收消息---【" + msg + "】---消费者1---FanoutQueue" + LocalTime.now());
    }

    @RabbitListener(queues = "lm.fanoutQueue2")
    public void listenFanoutQueueMessage2(String msg) throws InterruptedException {
        System.err.println("成功接收消息---【" + msg + "】---消费者2---FanoutQueue" + LocalTime.now());
    }

发布、订阅模型-Direct【基于注解声明队列和交换机】

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

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

每一个Queue都与Exchange设置一个BindingKey
发布者发送消息时，指定消息的RoutingKey
Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

可以指定多个key，相同的key相当于Fanout，但是性能会不好

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

基于@RabbitListener注解声明队列和交换机：@QueueBinding、@Exchange、@Queue

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

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

consumer.SpringRabbitListener

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "lm.directQueue1"),
            exchange = @Exchange(name = "lm.directExchange", type = ExchangeTypes.DIRECT),
            key = {"key1", "key2"}
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("成功接收消息---【" + msg + "】---消费者1---directQueue1" + LocalTime.now());
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "lm.directQueue2"),
            exchange = @Exchange(name = "lm.directExchange", type = ExchangeTypes.DIRECT),
            key = {"key1", "key3"}
    ))
    public void listenDirectQueue2(String msg){
        System.err.println("成功接收消息---【" + msg + "】---消费者2---directQueue2" + LocalTime.now());
    }

publisher.sendMessage2demomqDirectQueue

    @Test
    public void sendMessage2demomqDirectQueue(){
        String exchangeName = "lm.directExchange";
        String message1 = "hello, key1";
        String message2 = "hello, key2";
        String message3 = "hello, key3";
        rabbitTemplate.convertAndSend(exchangeName,"key1",message1);
        rabbitTemplate.convertAndSend(exchangeName,"key2",message2);
        rabbitTemplate.convertAndSend(exchangeName,"key3",message3);
    }

发布、订阅模型-Topic

描述下Direct交换机与Topic交换机的差异？

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

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

Routingkey 一般都是有一个或多个单词组成，多个单词之间以.分割

解释：

Queue1：绑定的是china.# ，因此凡是以 china.开头的routing key 都会被匹配到。包括china.news和china.weather
Queue2：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配。包括china.news和japan.news

在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
在publisher中编写测试方法，向itcast. topic发送消息

publisher.sendMessage2demomqTopicQueue

    @Test
    public void sendMessage2demomqTopicQueue() {
        String exchangeName = "lm.topicExchange";
        String message1 = "hello, china.news";
        String message2 = "hello, china.weather";
        String message3 = "hello, japan.news";
        String message4 = "hello, japan.weather";
        rabbitTemplate.convertAndSend(exchangeName, "china.news", message1);
        rabbitTemplate.convertAndSend(exchangeName, "china.weather", message2);
        rabbitTemplate.convertAndSend(exchangeName, "japan.news", message3);
        rabbitTemplate.convertAndSend(exchangeName, "japan.weather", message4);
    }

consumer.SpringRabbitListener

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "lm.topicQueue1"),
            exchange = @Exchange(name = "lm.topicExchange", type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void listenTopicQueue1(String msg){
        System.out.println("成功接收消息---【" + msg + "】---消费者1---topicQueue1" + LocalTime.now());
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "lm.topicQueue2"),
            exchange = @Exchange(name = "lm.topicExchange", type = ExchangeTypes.TOPIC),
            key = "japan.#"
    ))
    public void listenTopicQueue2(String msg){
        System.err.println("成功接收消息---【" + msg + "】---消费者2---topicQueue2" + LocalTime.now());
    }
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "lm.topicQueue4"),
            exchange = @Exchange(name = "lm.topicExchange", type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void listenTopicQueue3(String msg){
        System.out.println("成功接收消息---【" + msg + "】---消费者3---topicQueue3" + LocalTime.now());
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "lm.topicQueue4"),
            exchange = @Exchange(name = "lm.topicExchange", type = ExchangeTypes.TOPIC),
            key = "#.weather"
    ))
    public void listenTopicQueue4(String msg){
        System.err.println("成功接收消息---【" + msg + "】---消费者4---topicQueue4" + LocalTime.now());
    }

消息转化器【配置JSON转换器】

Spring会把发送的消息序列化为字节发送给MQ，接收消息时把字节反序列化为Java对象；Spring对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。

使用java JDK的序列化方式：

可读性差
可能出现注入问题，有安全漏洞
数据体积过大，传输消息速度慢，而且额外占用内存空间，性能差

@Test
public void testSendMap() throws InterruptedException {
    Map<String,Object> msg = new HashMap<>();
    msg.put("name", "Jack");
    msg.put("age", 21);
    rabbitTemplate.convertAndSend("simple.queue","", msg);
}

发送消息后查看控制台：

显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。

如果要修改只需要定义一个MessageConverter类型的Bean即可。

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

consumer/publisher.pom.xml

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

consumer/publisher.Application.MessageConverter
配置消息转换器

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

标签：AMQP,队列,SpringCloud,---,交换机,消息,msg,public
From： https://www.cnblogs.com/lm02/p/18156280

Basic Queue 简单队列模型

Work Queue 工作队列模型

发布、订阅模型

发布、订阅模型-Fanout【扇出、广播】

声明队列和交换机并绑定

发布、订阅模型-Direct【基于注解声明队列和交换机】

发布、订阅模型-Topic

消息转化器【配置JSON转换器】

相关文章

赞助商

阅读排行