一、消息中間件的應用場景

異步處理

場景：使用者注冊，資訊寫入資料庫後，需要給使用者發送注冊成功的郵件，再發送注冊成功的郵件。

1.同步調用：注冊成功後，順序執行發送郵件方法，發送短信方法，最後響應使用者

2.并行調用：注冊成功後，用多線程的方式并發執行發郵件和發短信方法，最後響應使用者

3.消息隊列：注冊成功後，将要發送的消息用很短的時間寫入消息隊列中，之後響應使用者；發送郵件的服務和發送短息的服務就可以從消息隊列中異步讀去，然後發送任務。

應用解耦

場景：購物下單後，調用庫存系統，更新庫存。

1.耦合的方式：訂單系統，寫調用庫存系統的邏輯。

2.解耦的方式：訂單系統，将下達的消息寫入消息隊列，庫存系統從消息隊列中讀取消息，更新庫存。

流量削峰

秒殺場景中，我們可以設定一個定長的消息隊列，秒殺開始，誰快誰先進入隊列，然後快速傳回使用者是否秒到，之後在平穩的處理秒殺後的業務。

二、消息服務中間件概述

1. 大多應用中，可通過消息服務中間件來提升系統異步通信、擴充解耦能力
1. 消息服務中兩個重要概念:
  
  消息代理(message broker)和目的地(destination) 當消息發送者發送消息以後，将由消息代理接管，消息代理保證消息傳遞到指定目

的地。

1. 消息隊列主要有兩種形式的目的地
- 1. 隊列(queue):點對點消息通信(point-to-point)
- 1. 主題(topic):釋出(publish)/訂閱(subscribe)消息通信
1. 點對點式:
- 1. 消息發送者發送消息，消息代理将其放入一個隊列中，消息接收者從隊列中擷取消息内容，消息讀取後被移出隊列
- 1. 消息隻有唯一的發送者和接受者，但并不是說隻能有一個接收者
1. 釋出訂閱式:
- 發送者(釋出者)發送消息到主題，多個接收者(訂閱者)監聽(訂閱)這個主題，那麼就會在消息到達時同時收到消息
1. JMS(Java Message Service)JAVA消息服務:
- 基于JVM消息代理的規範。ActiveMQ、HornetMQ是JMS實作
1. AMQP(Advanced Message Queuing Protocol)
- 進階消息隊列協定，也是一個消息代理的規範，相容JMS
- RabbitMQ是AMQP的實作

1. Spring支援
- spring-jms提供了對JMS的支援
- spring-rabbit提供了對AMQP的支援
- 需要ConnectionFactory的實作來連接配接消息代理
- 提供JmsTemplate、RabbitTemplate來發送消息
- @JmsListener(JMS)、@RabbitListener(AMQP)注解在方法上監聽消息代理發布的消息
- @EnableJms、@EnableRabbit開啟支援
1. Spring Boot自動配置
- JmsAutoConfiguration
- RabbitAutoConfiguration

三、RabbitMQ簡介

RabbitMQ是一個由erlang開發的AMQP(Advanved Message Queue Protocol)的開源實作。

1. 核心概念

Message ：消息，消息是不具名的，它由消息頭和消息體組成。消息體是不透明的，而消息頭則由一系列的可選屬性組成，這些屬性包括routing-key(路由鍵)、priority(相對于其他消息的優先權)、delivery-mode(指出該消息可能需要持久性存儲)等。
Publisher ：消息的生産者，也是一個向交換器釋出消息的用戶端應用程式。
Exchange ：交換器，用來接收生産者發送的消息并将這些消息路由給伺服器中的隊列。Exchange有4種類型:direct(預設)，fanout, topic, 和headers，不同類型的Exchange轉發消息的政策有所差別
Queue ：消息隊列，用來儲存消息直到發送給消費者。它是消息的容器，也是消息的終點。一個消息可投入一個或多個隊列。消息一直在隊列裡面，等待消費者連接配接到這個隊列将其取走。
Binding ：綁定，用于消息隊列和交換器之間的關聯。一個綁定就是基于路由鍵将交換器和消息隊列連接配接起來的路由規則，是以可以将交換器了解成一個由綁定構成的路由表。Exchange 和Queue的綁定可以是多對多的關系。
Connection ：網絡連接配接，比如一個TCP連接配接。
Channel ：信道，多路複用連接配接中的一條獨立的雙向資料流通道。信道是建立在真實的TCP連接配接内的虛拟連接配接，AMQP 指令都是通過信道發出去的，不管是釋出消息、訂閱隊列還是接收消息，這些動作都是通過信道完成。因為對于作業系統來說建立和銷毀 TCP 都是非常昂貴的開銷，所以引入了信道的概念，以複用一條 TCP 連接配接。
Consumer ：消息的消費者，表示一個從消息隊列中取得消息的用戶端應用程式。
Virtual Host ：虛拟主機，表示一批交換器、消息隊列和相關對象。虛拟主機是共享相同的身份認證和加密環境的獨立伺服器域。每個 vhost 本質上就是一個 mini 版的 RabbitMQ 伺服器，擁有自己的隊列、交換器、綁定和權限機制。vhost 是 AMQP 概念的基礎，必須在連接配接時指定， RabbitMQ 預設的 vhost 是 / 。
Broker ：表示消息隊列伺服器實體。

四、RabbitMQ運作機制

AMQP 中的消息路由

AMQP 中消息的路由過程和 Java 開發者熟悉的 JMS 存在一些差别，AMQP 中增加了 Exchange 和 Binding 的角色。生産者把消息釋出到 Exchange 上，消息最終到達隊列并被消費者接收，而 Binding 決定交換器的消息應該發送到那個隊列。

Exchange 類型

Exchange分發消息時根據類型的不同分發政策有差別，目前共四種類型：direct、fanout、topic、headers 。headers 比對 AMQP 消息的 header 而不是路由鍵， headers 交換器和 direct 交換器完全一緻，但性能差很多，目前幾乎用不到了，是以直接看另外三種類型：

五、RabbitMQ安裝

我們使用 docker 來安裝 RabbitMQ。

我們在 docker hub上選擇官方的帶management管理界面的最新版本。

#擷取rabbitmq鏡像
docker pull rabbitmq:3-management
#啟動 rabbitmq鏡像，5672是mq通信端口，15672是mq的web管理界面端口
run -d -p 5672:5672 -p 15672:15672 --name myrabbitmq 鏡像ID

通路127.0.0.1:15672 ，用賬号：guest 密碼：guest 登入，界面如下：

對rabbitmq的詳細使用在這裡，就不講解了，我們這節的重點是整合rabbitmq。

六、整合RabbitMQ

建立項目引入rabbitmq依賴。

1. pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.gf</groupId>
    <artifactId>springboot-rabbitmq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>springboot-rabbitmq</name>
    <description>Demo project for Spring Boot</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.5.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <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.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

2. MyAMQPConfig

package com.gf.config;

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.amqp.support.converter.MessageConverter;

/**
 * 自定義消息轉換器，預設是jdk的序列化轉換器，我們自定義為json的
 */
@Configuration
public class MyAMQPConfig {

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

}

3. springboot 測試類

我們測試建立管理配置、發送消息、接收消息

package com.gf;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringbootRabbitmqApplicationTests {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Autowired
    AmqpAdmin amqpAdmin;

    @Test
    public void contextLoads() {
    }

    @Test
    public void create(){
        //建立Exchange
        amqpAdmin.declareExchange( new DirectExchange( "exchange.direct") );
        amqpAdmin.declareExchange( new FanoutExchange( "exchange.fanout") );
        amqpAdmin.declareExchange( new TopicExchange( "exchange.topic") );

        //建立Queue
        amqpAdmin.declareQueue( new Queue( "direct.queue" , true ) );
        amqpAdmin.declareQueue( new Queue( "fanout.queue" , true ) );

        //綁定Queue
        amqpAdmin.declareBinding( new Binding( "direct.queue" , Binding.DestinationType.QUEUE , "exchange.direct" , "direct.queue" , null ) );
        amqpAdmin.declareBinding( new Binding( "fanout.queue" , Binding.DestinationType.QUEUE , "exchange.direct" , "fanout.queue" , null ) );
        amqpAdmin.declareBinding( new Binding( "direct.queue" , Binding.DestinationType.QUEUE , "exchange.fanout" , "" , null ) );
        amqpAdmin.declareBinding( new Binding( "fanout.queue" , Binding.DestinationType.QUEUE , "exchange.fanout" , "" , null ) );
        amqpAdmin.declareBinding( new Binding( "direct.queue" , Binding.DestinationType.QUEUE , "exchange.topic" , "direct.#" , null ) );
        amqpAdmin.declareBinding( new Binding( "fanout.queue" , Binding.DestinationType.QUEUE , "exchange.topic" , "direct.*" , null ) );


    }

    @Test
    public void send2Direct() {
        Map<String , Object> map = new HashMap<>();
        map.put( "msg" , "這是一條點對點消息" );
        map.put( "data" , Arrays.asList("helloworld" , 123 , true) );

        rabbitTemplate.convertAndSend( "exchange.direct" , "direct.queue" , map );

    }

    @Test
    public void send2Topic() {
        Map<String , Object> map = new HashMap<>();
        map.put( "msg" , "這是一條廣播消息" );
        map.put( "data" , Arrays.asList("topic消息" , 123 , true) );

        rabbitTemplate.convertAndSend( "exchange.fanout" , "", map );

    }

    @Test
    public void receive() {
        Object o = rabbitTemplate.receiveAndConvert( "direct.queue" );
        o.getClass();
        System.out.println(o.getClass());
        System.out.println(o);
    }

}

監聽消息

4. 啟動類

package com.gf;

import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * 自動配置
 * 1. RabbitAutoConfiguration
 * 2. 自動配置了連接配接工廠ConnectionFactory
 * 3. RabbitProperties 封裝了RabbitMQ的配置
 * 4. RabbitTemplate : 給RabbitMQ發送和接受消息
 * 5. AmqpAdmin : RabbitMQ系統管理功能元件
 * 6. @EnableRabbit + @RabbitListener
 */
@EnableRabbit
@SpringBootApplication
public class SpringbootRabbitmqApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringbootRabbitmqApplication.class, args);
    }
}

5. MQService

package com.gf.service;


import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

@Service
public class MQService {

    @RabbitListener(queues = "fanout.queue")
    public void receive(Message message) {
        System.out.println("收到消息 : " + new String(message.getBody()));

    }

}

源碼下載下傳：

https://github.com/gf-huanchupk/SpringBootLearning

Spring Boot 整合 rabbitmq二、消息服務中間件概述

一、消息中間件的應用場景

異步處理

應用解耦

流量削峰

二、消息服務中間件概述

三、RabbitMQ簡介

1. 核心概念

四、RabbitMQ運作機制

AMQP 中的消息路由

Exchange 類型

五、RabbitMQ安裝

六、整合RabbitMQ

1. pom.xml

2. MyAMQPConfig

3. springboot 測試類

4. 啟動類

5. MQService

繼續閱讀

nginx location中斜線的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

CentOS 7,docker安裝

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

【Docker】端口映射問題操作步驟

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method