RabbitMQ基础

RabbitMQ官网：https://rabbitmq.p2hp.com/

初识MQ

        MQ全称为Message Queue-消息队列，是一种应用程序对应用程序的消息通信，一端只管往队列不断发布信息，另一端只管往队列中读取消息，发布者不需要关心读取消息的谁，读取消息者不需要关心发布消息的是谁，各干各的互不干扰。

同步通讯和异步通讯

• 同步通讯

  • 优点：
    • 时效性强

  • 缺点：

    • 耦合度高，每次加入新的需求都要就该原来的代码
    • 性能下降，调用者需要等待服务提供者响应，如果调用链路过长则响应时间等于每次调用时间之和

    • 资源浪费，调用链中的每个服务在等待响应过程中，不能释放占用的资源，在高并发场景下会极度浪费系统资源

    • 级联失败，如果服务提供者出现问题，所有调用方都会出现问题，从而影响整个微服务

• 异步通讯

  • 优点：

    • 耦合度低

    • 吞吐量提升

    • 故障隔离

    • 流量削峰

  • 缺点：

    • 依赖于Broker的可靠性、安全性、吞吐能力

    • 架构复杂，排查难度高

MQ技术选型

	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司/社区	Rabbit	Apache	阿里	Apache
开发语言	Erlang	Java	Java	Scala&Java
协议支持	AMQP、XMPP、SMTP、STOPM	OpenWire、STOMP、REST、XMPP、AMQP	自定义协议	自定义协议
可用性	高	一般	高	高
单机吞吐量	一般	差	高	非常高
消息延迟	微妙	毫秒	毫秒	毫秒
消息可靠性	高	一般	高	一般

RabbitMQ基础架构图

• Broker：接收和分发消息的应用。

• Virtual Host：虚拟主机，多个不同的用户使用同一个RabbitMQ Server提供的服务时，可以划分出多个vhost，每个用户在自己的vhost创建的exchange/queue等。

• Connection：publisher / consumer 和 broker 之间TCP连接

• Channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务

• Exchange：交换机，消息到达Broker的第一站，根据分发规则，匹配查询表中的routing key，分到消息的queue中。常用的类型有：direct(point-to-point)，topic(publish-subscribe) and fanout(multicast)

• Queue：存放消息，最终被consumer取走

• Binding：exchange和queue之间的虚拟连接，binding中可以包含routing key。Binding信息保存到exchange中的查询表中，用于message的分发依据。

• Producer（Publisher）：消息发布者，发送消息

• Consumer：消息消费者，处理消息

RabbitMQ 常见消息模型

• 基本消息队列（BasicQueue）

• 工作消息队列（WorkQueue）

• 发布订阅模式

  • Fanout Exchange（广播）

  • Direct Exchange（路由）

  • Topic Exchange（主题）

SpringAMQP

        Spring AMQP 是 Spring 框架提供的一个基于 AMQP 协议的消息队列框架，用于简化 Spring 应用程序对消息队列的使用。它难点在于减少了对 AMQP 协议的细节处理，提供了一个高级别的抽象，使得生产者和消费者可以用简单的方式与消息队列进行通信。

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

        特征：

                1.监听器容器，用于异步处理入站消息

                2.用于发送和接收消息的RabbitTemplate

                3.RabbitAdmin用于自动声明式队列，交换和绑定

SpringAMQP基础配置

        1.引入依赖

<dependency>     <groupId>org.springframework.boot</groupId>     <artifactId>spring-boot-starter-amqp</artifactId> </dependency>

        2.在publisher和consumer配置

spring:     rabbitmq:         host: 127.0.0.1 #主机名         port: 5672 #端口         virtual-host: / #虚拟主机(默认)         username: root #用户名         password: 123 #密码

SpringAMQP实现RabbitMQ的基础消息队列（BasicQueue）

                基础消息队列：一对一的消息传递，‌即一个消息只能被一个消费者消费。

        1.在Publisher服务中使用RabbittTemplate的convertAndSend方法发送消息

@RunWith(SpringRunner.class) @SpringBootTest public class SpringAmqtTest{     @Autowired     private RabbitTemplate rabbitTemplate;          @Test     public void testSendMessage2SimpleQueue(){         String queueName = "simple.queue"; // 队列名称         String message = "123";// 消息内容         rabbitTemplate.convertAndSend(queueName, message);// 给simple.queue队列发送消息     } }

        2.在Consumer服务中绑定simple.queue队列，接收消息并编写消费逻辑

@Component public class SpringRabbitListener {     // 使用Rabbit监听器监听名称为simple.queue的队列。可同时监听多个队列。     @RabbitListener(queues = "simple.queue")      // 方法中的msg参数类型是根据消息发送者的内容来定的，发送和接收要保持一致。如果是一个对象那接收就是一个对象     public void listenSimpleQueueMessage(String msg) throws Exception {         System.out.println("spring 消费者接收到消息：" + msg);     } }

SpringAMQP实现RabbitMQ的工作消息队列（WorkQueue）

        工作消息队列：一对多的消息发布方式，‌即一个消息可以被多个消费者同时消费。‌
        特点：是一个消息只会被一个消费者消费，‌但多个消费者可以同时处理不同的消息，‌从而提高消息的处理速度。

        弊端：当多个消费者绑定一个队列时，消费者之间的消息分配是按照轮询的方式来消费消息，即consumer1消费者消费了1条消息后，轮到consumer2消费者消费消息，在consumer1处理消息完成前，consumer2消费者是不会处理消息的。这就导致如果其中一个消费者处理速度很慢，可能会影响其他的消费者闲置，不能有效地利用系统资源。

        解决方案：

                1.消息推送给能力较强的消费者（通过prefetch count）

                在RabbitMQ中，可以通过设置prefetch count来限制消费者每次从队列中取出的消息数量。这样做可以保证一个消费者不会因为处理速度慢而积累过多的消息，从而使其他消费者闲置。

                2.基于权重的分发（通过消费者权重）

                在RabbitMQ中，可以给每个消费者设置一个权重，这样RabbitMQ就会根据权重来分发消息。权重高的消费者会得到更多的消息，权重低的消费者会得到更少的消息。

        实现WorkQueue模式

// publisher @Component public class publisherService {     @Autowirte     private RabbitTemplate rabbitTemplate;      public void publisherSendMessage(){         String queneName = "work.queue";         String message = "123";         for (int i = 0; i < 50; i++) {             rabbitTemplate.convertAndSend(queueName, message);// 发送消息         }     } }

// consumer监听队列。 可绑定多个消费者 @Component public class ListenerQueue {      @RabbitListener(queues = "work.queue")     public void listenerMessage1(String msg) throws Exception {         System.out.println("spring 消费者1111接收到消息：" + msg);     }      @RabbitListener(queues = "work.queue")     public void listenerMessage2(String msg) throws Exception {         System.out.println("spring 消费者2222222222接收到消息：" + msg);     } }

SpringAMQP实现RabbitMQ的发布者订阅模式

        允许将同一消息发送给多个消费者。实现方式是加入了交换机（exchange），exchange负责消息路由，而不是存储，路由失败则消息丢失，queue能存储消息。

        交换机作用：

                1.接收publisher发送的消息

                2.将消息按照规则路由到与之绑定的队列

                3.不能缓存消息，路由失败则消息丢失

        常见的Exchange类型包括：

                FanoutExchange（广播）：会将接收到的消息路由到每个绑定的队列

                DirectExchange（路由）：会将接收到的消息根据规则路由到指定的队列，因此称为路由模式（routes）

                TopicExchange（话题）：会将接收到消息根据规则路由到指定的队列，与Direct类似，区别在于RoutingKey必须是多个单词的列表，并且以 . 分割。

发布者订阅模式之FanoutExchange

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

// 模拟FanoutExchange // 1.声明FanoutExchange交换机 // 2.声明队列（多个队列就生成多个） // 3.绑定队列和交换机（多个队列可以绑定一个交换机）  @Configuration public class FanoutConfig {     // 声明FanoutExchange交换机     @Bean     public FanoutExchange fanoutExchange(){         return new FanoutExchange("itcast.fanout");     }     // 声明队列1     @Bean     public Queue fanoutQueue1(){         return new Queue("fanout.queue1");     }     // 把队列1跟交换机绑定，在绑定时，与队列方法名必须一致，如fanoutQueue1     @Bean     public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){         return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);     }          // 声明队列2     @Bean     public Queue fanoutQueue2() {         return new Queue("fanout.queue2");     }     // 把队列2跟交换机绑定     @Bean     public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {         return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);     } }  // consumer监听队列 public class ListenerQueue {     @RabbitListener(queues = "fanout.queue1")     public void listenerFanoutQueue1(String msg){         System.out.println("spring 消费者1111接收到消息：" + msg);     }          @RabbitListener(queues = "fanout.queue2")     public void listenerFanoutQueue2(String msg){         System.out.println("spring 消费者22222222接收到消息：" + msg);     } }  // publisher发送消息 @RunWith(SpringRunner.class) @SpringBootTest public class PublisherTest(){     @Autowirt     private RabbitTemplate rabbitTeplate;     @Test     public void fanoutSend(){         String exchangeName = "itcast.fanout";// 交换机的名称，与声明交换机的名称保持一致         String msg = "123";         // 三个参数分别是：交换机名称、RoutingKey、消息内容         rabbitTeplate.convertAndSend(exchangeName, "", msg);     } }

发布者订阅模式之DirectExchange

        DirectExchange会将接收到的消息根据规则路由到指定的队列，因此称为路由模式（routes）

        特点：

                1.发布者发送消息时，指定消息的RoutingKey

                2.每个Queue都与Exchange设置一个BindingKey

                3.Exchange将消息路由到BindingKey与RoutingKey一致的Queue上

// consumer监听队列 // 可通过@RabbitListener注解，同时声明并绑定队列、交换机、BindingKey。 // bindings 可以绑定多个 @component public class ListenerQueue {     @RabbitListener(bindings = @QueueBinding(         value = @Queue(name = "direct.queue1"),         exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),         key = { "blue", "red"} // key可以配置一个或多个     ))     public void listenerDirectQueue1(String msg) {         // ......     }          @RabbitListener(bindings = @QueueBinding(         value = @Queue(name = "direct.queue2"),         exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),         key = { "yellow", "red"}     ))     public void listenerDirectQueue2(String msg) {         // ......     } }  // publisher发送消息 @RunWith(SpringRunner.class) @SpringBootTest public class PublisherTest(){     @Autowirt     private RabbitTemplate rabbitTeplate;     @Test     public void directSend(){         String exchangeName = "itcast.direct";// 交换机的名称，与声明交换机的名称保持一致         String msg = "123";         // 三个参数分别是：交换机名称、RoutingKey、消息内容         rabbitTeplate.convertAndSend(exchangeName, "red", msg);     } }

发布者订阅模式之TopicExchange

        TopicExchange会将接收到消息根据根据规则路由到指定的队列，与Direct类似，区别在于RoutingKey必须是多个单词的列表，并且以 . 分割。

        Queue和Exchange指定BindingKey时可以使用通配符：

                #：代指0个或多个单词

                *：代指一个单词

注意：只能在BindingKey中使用 # 和 * 通配符，RoutingKey必须是多个单词并用 . 分割

// consumer监听队列 @Component public class ListenerQueue {     @RabbitListener(bindings = @QueueBinding(         value = @Queue(name = "topic.queue1"),         exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),         key = "chain.#"     ))     public void topicQueue1(String msg) {         // .......     }          @RabbitListener(bindings = @QueueBinding(         value = @Queue(name = "topic.queue2"),         exchange = @Exchange(name = "itcast.topic" type = ExchangeTypes.TOPIC),         key = "#.news"     ))     public void topicQueue2(String msg) {         // .......     } }  // publisher发送消息 @RunWith(SpringRunner.class) @SpringBootTest public class PublisherTest(){     @Autowirt     private RabbitTemplate rabbitTeplate;     @Test     public void topicSend(){         String exchangeName = "itcast.topic";// 交换机的名称，与声明交换机的名称保持一致         String msg = "123";         // 三个参数分别是：交换机名称、RoutingKey、消息内容         rabbitTeplate.convertAndSend(exchangeName, "chain.weather", msg);     } }