spring boot：RabbitMQ消息隊(duì)列

一、概述

  • 大多數(shù)運(yùn)用中，可通過消息中間件來提升系統(tǒng)異步通訊、擴(kuò)展解耦能力。
  • 消息服務(wù)中的兩個(gè)重要概念：
    消息代理（message broker）和目的地（destination）
    當(dāng)消息發(fā)送者發(fā)送消息以后，將由消息代理接管。消息代理保證消息傳遞到指定目的地。
  • 消息代理主要有兩種形式的目的地：
    1、隊(duì)列（queue）：點(diǎn)對(duì)點(diǎn)通訊（point-to-point）
    2、主題（topic）：發(fā)布（publish）/ 訂閱（subscribe）消息通訊
  • JMS（Java Message Service）消息服務(wù)
    -- 基于JVM消息代理的規(guī)范。如ActivateMQ
  • AMQP（Advanced Message Queuing Protocol）
    -- 高級(jí)消息隊(duì)列，也是一種消息代理的規(guī)范。如RabbitMQ。

二、消息隊(duì)列的使用場(chǎng)景

1、異步處理

注冊(cè)用戶后需要發(fā)送郵件和短信提示
（1）串行方式：將注冊(cè)用戶信息寫入數(shù)據(jù)庫后，發(fā)送注冊(cè)郵件、再發(fā)送注冊(cè)短信



（2）并行方式：將注冊(cè)信息寫入數(shù)據(jù)庫后,發(fā)送郵件的同時(shí),發(fā)送短信。



（3）消息隊(duì)列：引入消息隊(duì)列后，把發(fā)送郵件、短信等不是必須的業(yè)務(wù)邏輯異步處理。

2、應(yīng)用解耦

在網(wǎng)上購物時(shí)，用戶下單后,訂單系統(tǒng)需要通知庫存系統(tǒng)
（1）傳統(tǒng)方式



缺點(diǎn)：

  • 庫存系統(tǒng)出故障，訂單就會(huì)失敗。

  • 訂單系統(tǒng)與庫存系統(tǒng)高度耦合。
    （2）消息隊(duì)列



    用戶下單后，訂單系統(tǒng)完成持久化處理，將消息寫入消息隊(duì)列，返回用戶訂單。就算庫存系統(tǒng)出現(xiàn)故障，消息系統(tǒng)也能保證訂單消息的可靠傳遞。

3、流量削峰

秒殺活動(dòng)，可能會(huì)因?yàn)榱髁刻蠖鴮?dǎo)致應(yīng)用掛掉。一般在應(yīng)用前端加入消息隊(duì)列。
作用：
1、控制活動(dòng)人數(shù)，超過一定閾值的訂單直接丟棄。
2、可以緩解短時(shí)間的高流量壓垮應(yīng)用。


三、RabbitMQ簡(jiǎn)介

RabbitMQ是一個(gè)開源用Erlang語言編寫的AMQP實(shí)現(xiàn)。

  • Message（消息）
    消息由消息頭和消息體組成，消息體是不透明的；消息頭由一系列屬性組成，如routing-key（路由鍵）、priority（優(yōu)先級(jí)）、delivery-mode（指出該消息可能需要持久化）等。
  • Exchange（交換機(jī)）
    用于接收生產(chǎn)者（publisher）發(fā)送的消息，并根據(jù)條件轉(zhuǎn)發(fā)給服務(wù)器中的隊(duì)列。
    Exchange有四種類型：direct（默認(rèn)）、fanout、topic、headers
  • Queue（消息隊(duì)列）
    保存消息直到被消費(fèi)者（consumer）取走。
  • Binding（綁定）
    用于交換器與消息隊(duì)列之間的關(guān)聯(lián)。
    Exchange和Queue之間的關(guān)聯(lián)可以是多對(duì)多的。
  • Publisher（生產(chǎn)者）
  • Consumer（消費(fèi)者）
  • Connection（連接）
  • Channel（信道）
  • Virtual host（虛擬主機(jī)）
  • Broker（消息隊(duì)列服務(wù)器實(shí)體）
rabbitMQ

四、RabbitMQ運(yùn)行機(jī)制

1、Direct Exchange

消息中的路由鍵（routing-key）如果和Binding中的binding-key一致，交換器就將消息發(fā)送到對(duì)應(yīng)的隊(duì)列中。


2、Fanout Exchange

每個(gè)發(fā)送到fanout類型的交換器的消息都會(huì)分到所有綁定的隊(duì)列上去。


3、Topic Exchange

Topic Exchange增加了一個(gè)模糊匹配的功能，讓routing-key與某個(gè)模式相匹配，符合匹配規(guī)則的隊(duì)列便于交換機(jī)綁定。
通配符#匹配0個(gè)或多個(gè)字符，*匹配一個(gè)單詞

五、spring boot集成RabbitMQ

1、 準(zhǔn)備

  • 創(chuàng)建交換器（Exchange）
    smallbear.direct、smallbear.fanout、smallbear.topic

  • 創(chuàng)建消息隊(duì)列（Queue）
    qq.music、qq.vedio、163.music、163.vedio、com

  • 交換機(jī)綁定消息隊(duì)列
    （1） smallbear.direct



    （2）smallbear.fanout



    （3）smallbear.topic

2. 構(gòu)建項(xiàng)目

  • 添加依賴
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
  • 配置
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
  • RabbitMQ配置
@Configuration
public class RabbitConfig {

    /**
     * 消息默認(rèn)是通過序列化傳到rabbitmq中的，這里設(shè)置成json格式
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

3. 測(cè)試

  • direct測(cè)試
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void directTest() {
        Map<String, Object> map = new HashMap<>();
        map.put("sender", "小明");
        map.put("msg", "今晚出去玩不？");
        rabbitTemplate.convertAndSend("smallbear.direct", "163.vedio", map);
    }

    @Test
    public void directReceiveTest() {
        Object o = rabbitTemplate.receiveAndConvert("163.vedio");
        System.out.println(o); // {msg=今晚出去玩不？, sender=小明}
    }

運(yùn)行directTest()后可在163.vedio隊(duì)列中看到以下結(jié)果


運(yùn)行directReceiveTest()獲取163.vedio隊(duì)列中的消息。

  • fanout測(cè)試
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Book {
    private String name;
    private String author;
    private Double price;
}

    @Test
    public void fanoutTest() {
        Book book = new Book("龍族", "江南", 133.32);
        rabbitTemplate.convertAndSend("smallbear.fanout", "", book);
    }

運(yùn)行fanoutTest()，四個(gè)消息隊(duì)列都接收到以下消息

  • topic測(cè)試
    @Test void topicTest() {
        rabbitTemplate.convertAndSend("smallbear.topic", "google.com", "Google Chrome");
        rabbitTemplate.convertAndSend("smallbear.topic", "baidu.com", "百度一下，你就知道");
    }

運(yùn)行topicTest()，com隊(duì)列獲取到兩條消息


4. @RabbitListner監(jiān)聽

  • 開啟RabbitMQ
@EnableRabbit
@SpringBootApplication
public class RabbitmqDemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(RabbitmqDemoApplication.class, args);
    }
}
  • 添加Rabbit監(jiān)聽器
@Data
public class User {
    private String name;
    private String telephone;
    private String email;
}

@Service
public class RegistryService {

    @Autowired
    private MessageConverter messageConverter;

    @RabbitListener(queues = "send.telephone")
    public void sendTelephone(User user) {
        System.out.println("新用戶為" + user.getName() + "發(fā)送短信到" + user.getTelephone());
    }

    @RabbitListener(queues = "send.email")
    public void sendEmail(Message message) {
        System.out.println("獲取頭部信息:" + message.getMessageProperties());
        User user = (User) messageConverter.fromMessage(message);
        System.out.println("新用戶為" + user.getName() + "發(fā)送郵件到" + user.getEmail());
    }
}
  • 編寫Controller
@RestController
@RequestMapping("user")
public class UserController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostMapping("registry")
    public String registry(User user) {
        System.out.println("用戶" + user.getName() + "注冊(cè)成功");
        // 新增隊(duì)列send.email和send.telephone，并與smallbear.direct交換器綁定
        rabbitTemplate.convertAndSend("smallbear.direct", "send.telephone", user);
        rabbitTemplate.convertAndSend("smallbear.direct", "send.email", user);
        return "SUCCESS";
    }
}

  • 測(cè)試



    輸出結(jié)果

用戶小李注冊(cè)成功
獲取頭部信息:MessageProperties [headers={__TypeId__=com.smallbear.rabbitmqdemo.entity.User}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=smallbear.direct, receivedRoutingKey=send.email, deliveryTag=1, consumerTag=amq.ctag-Hxt7ShNqlFOu6Ww-qlb1bA, consumerQueue=send.email]
新用戶為小李發(fā)送郵件到12345678@qq.com
新用戶為小李發(fā)送短信到13111111111

5. AmqpAdmin設(shè)置Queue、Exchange和Biinding

    @Autowired
    private AmqpAdmin amqpAdmin;

    @Test void amqpAdminTest() {
        amqpAdmin.declareExchange(new DirectExchange("exchange-demo"));
        amqpAdmin.declareQueue(new Queue("queue-demo", true));
        amqpAdmin.declareBinding(new Binding("queue-demo", Binding.DestinationType.QUEUE, "exchange-demo", "queue-demo", null));
    }
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容