RabbitMQ

RabbitMQ是實作了高級佇列訊息協議（AMQP: Advanced Message Queuing Protocol）的開源訊息代理軟體，它也被稱作為面向訊息的中間件，RabbitMQ服務器使用了Erlang語言撰寫而成，它具有高性能，可伸縮的特性，

1.安裝

各個平臺的安裝請參考官網鏈接：RabbitMQ安裝

2.基本概念

2.1 RabbitMQ角色定義

Producer：訊息生產者，生產并發送訊息，
Consumer：訊息消費者，接受并處理訊息，
Borker：訊息佇列服務器的物體，他負責接受生產者產生的訊息，并將訊息發送至接收者或者其它的Broker，
Exchange：訊息交換機，訊息被創建后第一個到達的地方，通過它指定的路由規則，將訊息分發到不同的訊息佇列中去，
Routing Key：路由關鍵字，Exchange根據這個關鍵字進行訊息的投遞，
Queue：訊息佇列，訊息被發送及路由之后到達的地方，到達Queue的訊息即進入了等待消費的狀態，每一個訊息都可以被發送到一個或多個交換機，
Binding：系結，連接Exchange和Queue，將它們按照路由規則系結起來，
Virtual Host：虛擬主機，是對Broker的虛擬劃分，將消費者，生產者及它們依賴的AMQP相關的結構進行隔離，
Connection：連接，生產者，消費者及Broker之間進行通信的物理網路，
Channel：訊息通道，用于連接生產者和消費者，在每一個連接里，可以建立多個Channel，每一個Channel代表一次回話任務，使用Channel可以隔離同一個連接中的不同的互動內容，

2.2 Exchange型別

Direct交換機：完全根據訊息的Routing Key進行訊息投遞，
Topic交換機：對Routing Key進行模式匹配后進行訊息投遞，使用#匹配一個或者多個詞，使用*匹配正好一個詞，
Fanout交換機：不需要任何Key，采用廣播模式的方式將一個訊息投遞到與該交換機系結的所有佇列，

2.3 訊息持久化

Exchange持久化：宣告時指定durable等于1（1是持久化），
Queue持久化：宣告時指定durable等于1（1是持久化），
訊息持久化：宣告時指定delivery_mode等于2（1是非持久化），

3.訊息投遞大致流程

客戶端連接訊息佇列服務器并打開一個Channel，
客戶端宣告一個Exchange并設定相關屬性，
客戶端宣告一個Queue并設定相關屬性，
客戶端使用Routing Key在Exchange和Queue之間建立系結關系，
客戶端投遞訊息到Exchange，
Exchange根據根據訊息的Key和已經設定的Binding進行訊息路由，將訊息投遞到一個或多個Queue里，

4.Spring Boot應用整合RabbitMQ

4.1 準備

安裝并開啟RabbitMQ服務

4.2 構建SpringBoot應用

新建一個Spring Boot工程

添加依賴：spring-boot-starter-amqp

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

配置RabbitMQ

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    userName: admin
    password: admin
    virtual-host: /

RabbitMQ配置類

@Configuration
public class RabbitConfig {

    /**
     * 宣告交換機
     * String: 交換機名稱
     * Boolean: 是否持久化
     * Boolean: 是否自動洗掉
     */
    /** Direct Exchange **/
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("DIRECT_EXCHANGE", true, false);
    }
    /** Topic Exchange **/
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("TOPIC_EXCHANGE", true, false);
    }
    /** Fanout Exchange **/
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("FANOUT_EXCHANGE", true, false);
    }

    /**
     * 宣告訊息佇列
     * String: 訊息佇列名稱
     * Boolean: 是否持久化
     */
    @Bean
    public Queue helloQueue() {
        return new Queue("HELLO_QUEUE", true);
    }
    @Bean
    public Binding helloBinding() {
        return BindingBuilder.bind(helloQueue()).to(directExchange()).with("mq.hello");
    }
}

RabbitMQ生產者

@Component
public class producer {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        rabbitTemplate.convertAndSend("HELLO_QUEUE", "hello mq!");
    }
}

RabbitMQ消費者

@Component
@RabbitListener(queues = "HELLO_QUEUE")
public class consumer {

    @RabbitHandler
    public void process(String message) {
        System.out.println(message);
    }
}

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/41959.html

標籤：Java

上一篇：散串列漫談

下一篇：Java連載138-資料庫洗掉資料以及編譯預處理

Spring Boot & RabbitMQ