Spring Boot (25) RabbitMQ消息队列

Posted 海盗屋

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spring Boot (25) RabbitMQ消息队列相关的知识,希望对你有一定的参考价值。

MQ全程(Message Queue)又名消息队列,是一种异步通讯的中间件。可以理解为邮局,发送者将消息投递到邮局,然后邮局帮我们发送给具体的接收者,具体发送过程和时间与我们无关,常见的MQ又kafka、activemq、zeromq、rabbitmq等等。

 

RabbitMQ

  RabbitMQ是一个遵循AMQP协议,由面向高并发的erlang语言开发而成,用在实时的对可靠性要求比较高的消息传递上,支持多种语言客户端,支持延迟队列。

 

基础概念

  Broker:消息队列的服务器实体

  Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列

  Queue:消息队列载体,每个消息都会被投入到一个或多个队列

  Binding:绑定,它主要是把exchange和queue按照路由规则绑定起来

  Routing Key:路由关键字,exchange根据这个关键字进行消息投递

  vhost:虚拟主机,一个broker里可以开设多个vhost,用作不同用户的权限分离

  producer:消息生产者,投递消息的程序

  consumer:消息消费者,接收消息的程序

  channel:消息通道,在客户端的每个连接里,可以建立多个channel,每个channel代表一个会话任务

 

常见应用场景

  1.邮箱发送:用户注册后投递消息到rabbitmq中,由消息的消费方异步的发送邮件,提升系统响应速度。

  2.流量削锋:一般在秒杀活动中应用广泛,秒杀会因为流量过大,导致应用挂掉,为了解决这个问题,一般在应用前端加入消息队列。用于控制活动人数,将超过此一定阀值的订单直接丢弃。缓解端时间的高流量压垮应用。

  3.订单超时:利用rabbitmq的延迟队列,可以很简单的实现订单超功能,比如用户在下单后30分钟未支付取消订单。

 

导入依赖

在pom.xml中添加spring-boot-starter-amqp的依赖

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

属性配置

在application.yml中配置rabbitmq的相关信息,这里配置了手动的ACK开关

spring:
  rabbitmq:
    username: david
    password: 123456
    host: localhost
    port: 5672
    virtual-host: /
    listener:
      simple:
        acknowledge-mode: manual #手动ACK 不开启自动ACK模式,目的是防止报错后为正确处理消息丢失 默认为none

 

定义队列

  

package com.spring.boot.utils;

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

@Configuration
public class RabbitConfig {
    public static final String DEFAULT_BOOK_QUEUE = "dev.book.register.default.queue";
    public static final String MANUAL_BOOK_QUEUE = "dev.book.register.manual.queue";

    @Bean
    public Queue defaultBookQueue(){
        //参数1 队列名,参数2 是否持久化处理
        return new Queue(DEFAULT_BOOK_QUEUE,true);
    }
    @Bean Queue manualBookQueue(){
        return new Queue(MANUAL_BOOK_QUEUE,true);
    }
}

 

实体类

package com.spring.boot.bean;

import java.io.Serializable;

public class Book implements Serializable{
    private Integer id;
    private String name;

    public Integer getId() {
        return id;
    }

    public void setId(Integer id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }
}

 

控制器

新建一个bookController,用于消息发送

package com.spring.boot.controller;

import com.spring.boot.bean.Book;
import com.spring.boot.utils.RabbitConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/books")
public class BookController {

    //spring boot 2.x版本推荐构造器注入 而不是属性注入
    private final RabbitTemplate rabbitTemplate;

    @Autowired
    public BookController(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @GetMapping("/defaultMessage")
    public void defaultMessage() {
        Book book = new Book();
        book.setId(1);
        book.setName("hello RabbitMQ");
        this.rabbitTemplate.convertAndSend(RabbitConfig.DEFAULT_BOOK_QUEUE, book);
        this.rabbitTemplate.convertAndSend(RabbitConfig.MANUAL_BOOK_QUEUE, book);
    }

}

 

消息消费者

默认情况下 spring-boot-data-amqp是自动ACK机制,就意味着MQ会在消息消费完毕后自动帮我们去ACK,这样依赖就存在这样一个问题:如果报错了,消息不会丢失,会无限循环消费,很容易把磁盘空间耗完,虽然可以配置消费的次数但这种做法也不太好。目前比较推荐的就是我们手动ACK然后将消费错误的消息转移到其他的消息队列中,做补偿处理

package com.spring.boot.handler;

import com.rabbitmq.client.Channel;
import com.spring.boot.bean.Book;
import com.spring.boot.utils.RabbitConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class BookHandler {

    @RabbitListener(queues={RabbitConfig.DEFAULT_BOOK_QUEUE})
    public void listenerAutoACK(Book book, Message message,Channel channel) throws IOException {
        final long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try{
            System.out.println("ListenerAutoAck 监听到的消息:" + book.toString());
            // TODO 通知MQ 已经被消费完成 可以ACK了
            channel.basicAck(deliveryTag,false);
        } catch (IOException e) {
            //TODO 处理失败,重新压入MQ
            channel.basicRecover();
            e.printStackTrace();
        }
    }

    @RabbitListener(queues={RabbitConfig.MANUAL_BOOK_QUEUE})
    public void listenerManualACK(Book book,Message message,Channel channel){
        System.out.println("listenerManualACK监听到的消息:"+book.toString());
        try{
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (IOException e) {
            //如果报错了 可以进行容错处理,比如转移当前消息进入其他队列
            e.printStackTrace();
        }
    }
}

 

测试:启动项目 输入路径 http://localhost:8088/books/defaultMessage

以上是关于Spring Boot (25) RabbitMQ消息队列的主要内容,如果未能解决你的问题,请参考以下文章

[Spring boot] Spring boot 整合RabbitMQ实现通过RabbitMQ进行项目的连接

Spring Boot 2.X - Spring Boot整合AMQP之RabbitMQ

spring-boot 集成 rabbitmq

spring-boot 集成 rabbitmq

spring boot 1.5.4 整合rabbitMQ(十七)

Spring Boot 整合 RabbitMQ