2. 程式人生 > >springboot RabbitMQ 配置

springboot RabbitMQ 配置

阿新 • • 發佈：2017-07-14
接收 handle mes 客戶端 fin ng- 技術 msg framework

引用自 http://www.cnblogs.com/ityouknow/p/6120544.html

自己留一份 記錄一下

　　RabbitMQ 即一個消息隊列，主要是用來實現應用程序的異步和解耦，同時也能起到消息緩沖，消息分發的作用。

　　消息中間件在互聯網公司的使用中越來越多，剛才還看到新聞阿裏將RocketMQ捐獻給了apache，當然了今天的主角還是講RabbitMQ。消息中間件最主要的作用是解耦，中間件最標準的用法是生產者生產消息傳送到隊列，消費者從隊列中拿取消息並處理，生產者不用關心是誰來消費，消費者不用關心誰在生產消息，從而達到解耦的目的。在分布式的系統中，消息隊列也會被用在很多其它的方面，比如：分布式事務的支持，RPC的調用等等。

　　以前一直使用的是ActiveMQ，在實際的生產使用中也出現了一些小問題，在網絡查閱了很多的資料後，決定嘗試使用RabbitMQ來替換ActiveMQ，RabbitMQ的高可用性、高性能、靈活性等一些特點吸引了我們，查閱了一些資料整理出此文。

RabbitMQ介紹

RabbitMQ是實現AMQP（高級消息隊列協議）的消息中間件的一種，最初起源於金融系統，用於在分布式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。RabbitMQ主要是為了實現系統之間的雙向解耦而實現的。當生產者大量產生數據時，消費者無法快速消費，那麽需要一個中間層。保存這個數據。

AMQP，即Advanced Message Queuing Protocol，高級消息隊列協議，是應用層協議的一個開放標準，為面向消息的中間件設計。消息中間件主要用於組件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、隊列、路由（包括點對點和發布/訂閱）、可靠性、安全。

RabbitMQ是一個開源的AMQP實現，服務器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用於在分布式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。

相關概念

通常我們談到隊列服務, 會有三個概念： 發消息者、隊列、收消息者，RabbitMQ 在這個基本概念之上, 多做了一層抽象, 在發消息者和 隊列之間, 加入了交換器 (Exchange). 這樣發消息者和隊列就沒有直接聯系, 轉而變成發消息者把消息給交換器, 交換器根據調度策略再把消息再給隊列。

技術分享

  • 左側 P 代表 生產者，也就是往 RabbitMQ 發消息的程序。
  • 中間即是 RabbitMQ，其中包括了 交換機 和 隊列。
  • 右側 C 代表 消費者，也就是往 RabbitMQ 拿消息的程序。

那麽，其中比較重要的概念有 4 個，分別為：虛擬主機，交換機，隊列，和綁定。

  • 虛擬主機：一個虛擬主機持有一組交換機、隊列和綁定。為什麽需要多個虛擬主機呢？很簡單，RabbitMQ當中，用戶只能在虛擬主機的粒度進行權限控制。 因此，如果需要禁止A組訪問B組的交換機/隊列/綁定，必須為A和B分別創建一個虛擬主機。每一個RabbitMQ服務器都有一個默認的虛擬主機“/”。
  • 交換機：Exchange 用於轉發消息，但是它不會做存儲 ，如果沒有 Queue bind 到 Exchange 的話，它會直接丟棄掉 Producer 發送過來的消息。
    這裏有一個比較重要的概念：路由鍵 。消息到交換機的時候，交互機會轉發到對應的隊列中，那麽究竟轉發到哪個隊列，就要根據該路由鍵。
  • 綁定：也就是交換機需要和隊列相綁定，這其中如上圖所示，是多對多的關系。

交換機(Exchange)

交換機的功能主要是接收消息並且轉發到綁定的隊列，交換機不存儲消息，在啟用ack模式後，交換機找不到隊列會返回錯誤。交換機有四種類型：Direct, topic, Headers and Fanout

  • Direct：direct 類型的行為是"先匹配, 再投送". 即在綁定時設定一個 routing_key, 消息的routing_key 匹配時, 才會被交換器投送到綁定的隊列中去.
  • Topic：按規則轉發消息（最靈活）
  • Headers：設置header attribute參數類型的交換機
  • Fanout：轉發消息到所有綁定隊列

Direct Exchange
Direct Exchange是RabbitMQ默認的交換機模式，也是最簡單的模式，根據key全文匹配去尋找隊列。

技術分享

　　第一個 X - Q1 就有一個 binding key，名字為 orange； X - Q2 就有 2 個 binding key，名字為 black 和 green。當消息中的 路由鍵 和 這個 binding key 對應上的時候，那麽就知道了該消息去到哪一個隊列中。

　　Ps：為什麽 X 到 Q2 要有 black，green，2個 binding key呢，一個不就行了嗎？ - 這個主要是因為可能又有 Q3，而Q3只接受 black 的信息，而Q2不僅接受black 的信息，還接受 green 的信息。

Topic Exchange

　　Topic Exchange 轉發消息主要是根據通配符。 在這種交換機下，隊列和交換機的綁定會定義一種路由模式，那麽，通配符就要在這種路由模式和路由鍵之間匹配後交換機才能轉發消息。

　　在這種交換機模式下：

  • 路由鍵必須是一串字符，用句號（.） 隔開，比如說 agreements.us，或者 agreements.eu.stockholm 等。
  • 路由模式必須包含一個 星號（*），主要用於匹配路由鍵指定位置的一個單詞，比如說，一個路由模式是這樣子：agreements..b.*，那麽就只能匹配路由鍵是這樣子的：第一個單詞是 agreements，第四個單詞是 b。 井號（#）就表示相當於一個或者多個單詞，例如一個匹配模式是agreements.eu.berlin.#，那麽，以agreements.eu.berlin開頭的路由鍵都是可以的。

具體代碼發送的時候還是一樣，第一個參數表示交換機，第二個參數表示routing key，第三個參數即消息。如下：

rabbitTemplate.convertAndSend("testTopicExchange","key1.a.c.key2", " this is  RabbitMQ!");

topic 和 direct 類似, 只是匹配上支持了"模式", 在"點分"的 routing_key 形式中, 可以使用兩個通配符:

  • *表示一個詞.
  • #表示零個或多個詞.

Headers Exchange

headers 也是根據規則匹配, 相較於 direct 和 topic 固定地使用 routing_key , headers 則是一個自定義匹配規則的類型.
在隊列與交換器綁定時, 會設定一組鍵值對規則, 消息中也包括一組鍵值對( headers 屬性), 當這些鍵值對有一對, 或全部匹配時, 消息被投送到對應隊列.

Fanout Exchange

Fanout Exchange 消息廣播的模式，不管路由鍵或者是路由模式，會把消息發給綁定給它的全部隊列，如果配置了routing_key會被忽略。

springboot集成RabbitMQ

springboot集成RabbitMQ非常簡單，如果只是簡單的使用配置非常少，springboot提供了spring-boot-starter-amqp項目對消息各種支持。

簡單使用

1、配置pom包，主要是添加spring-boot-starter-amqp的支持

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

2、配置文件

配置rabbitmq的安裝地址、端口以及賬戶信息

spring.application.name=spirng-boot-rabbitmq
spring.rabbitmq.host=192.168.0.86
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

3、隊列配置

@Configuration
public class RabbitConfig {

    @Bean
    public Queue Queue() {
        return new Queue("hello");
    }

}

3、發送者

rabbitTemplate是springboot 提供的默認實現

public class HelloSender {
    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String context = "hello " + new Date();
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("hello", context);
    }
}

4、接收者

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {

    @RabbitHandler
    public void process(String hello) {
        System.out.println("Receiver  : " + hello);
    }

}

5、測試

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

    @Autowired
    private HelloSender helloSender;

    @Test
    public void hello() throws Exception {
        helloSender.send();
    }

}

註意，發送者和接收者的queue name必須一致，不然不能接收

多對多使用

一個發送者，N個接收者或者N個發送者和N個接收者會出現什麽情況呢？

一對多發送
對上面的代碼進行了小改造,接收端註冊了兩個Receiver,Receiver1和Receiver2，發送端加入參數計數，接收端打印接收到的參數，下面是測試代碼，發送一百條消息，來觀察兩個接收端的執行效果

@Test
public void oneToMany() throws Exception {
    for (int i=0;i<100;i++){
        neoSender.send(i);
    }
}

結果如下：

Receiver 1: spirng boot neo queue ****** 11
Receiver 2: spirng boot neo queue ****** 12
Receiver 2: spirng boot neo queue ****** 14
Receiver 1: spirng boot neo queue ****** 13
Receiver 2: spirng boot neo queue ****** 15
Receiver 1: spirng boot neo queue ****** 16
Receiver 1: spirng boot neo queue ****** 18
Receiver 2: spirng boot neo queue ****** 17
Receiver 2: spirng boot neo queue ****** 19
Receiver 1: spirng boot neo queue ****** 20

根據返回結果得到以下結論

一個發送者，N個接受者,經過測試會均勻的將消息發送到N個接收者中

多對多發送

復制了一份發送者，加入標記，在一百個循環中相互交替發送

@Test
    public void manyToMany() throws Exception {
        for (int i=0;i<100;i++){
            neoSender.send(i);
            neoSender2.send(i);
        }
}

結果如下：

Receiver 1: spirng boot neo queue ****** 20
Receiver 2: spirng boot neo queue ****** 20
Receiver 1: spirng boot neo queue ****** 21
Receiver 2: spirng boot neo queue ****** 21
Receiver 1: spirng boot neo queue ****** 22
Receiver 2: spirng boot neo queue ****** 22
Receiver 1: spirng boot neo queue ****** 23
Receiver 2: spirng boot neo queue ****** 23
Receiver 1: spirng boot neo queue ****** 24
Receiver 2: spirng boot neo queue ****** 24
Receiver 1: spirng boot neo queue ****** 25
Receiver 2: spirng boot neo queue ****** 25

結論：和一對多一樣，接收端仍然會均勻接收到消息

高級使用

對象的支持

springboot以及完美的支持對象的發送和接收，不需要格外的配置。

//發送者
public void send(User user) {
    System.out.println("Sender object: " + user.toString());
    this.rabbitTemplate.convertAndSend("object", user);
}

...

//接受者
@RabbitHandler
public void process(User user) {
    System.out.println("Receiver object : " + user);
}

結果如下：

Sender object: User{name=‘neo‘, pass=‘123456‘}
Receiver object : User{name=‘neo‘, pass=‘123456‘}

Topic Exchange

topic 是RabbitMQ中最靈活的一種方式，可以根據routing_key自由的綁定不同的隊列

首先對topic規則配置，這裏使用兩個隊列來測試

@Configuration
public class TopicRabbitConfig {

    final static String message = "topic.message";
    final static String messages = "topic.messages";

    @Bean
    public Queue queueMessage() {
        return new Queue(TopicRabbitConfig.message);
    }

    @Bean
    public Queue queueMessages() {
        return new Queue(TopicRabbitConfig.messages);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("exchange");
    }

    @Bean
    Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
    }

    @Bean
    Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
    }
}

使用queueMessages同時匹配兩個隊列，queueMessage只匹配"topic.message"隊列

public void send1() {
    String context = "hi, i am message 1";
    System.out.println("Sender : " + context);
    this.rabbitTemplate.convertAndSend("exchange", "topic.message", context);
}

public void send2() {
    String context = "hi, i am messages 2";
    System.out.println("Sender : " + context);
    this.rabbitTemplate.convertAndSend("exchange", "topic.messages", context);
}

發送send1會匹配到topic.#和topic.message 兩個Receiver都可以收到消息，發送send2只有topic.#可以匹配所有只有Receiver2監聽到消息

Fanout Exchange

Fanout 就是我們熟悉的廣播模式或者訂閱模式，給Fanout交換機發送消息，綁定了這個交換機的所有隊列都收到這個消息。

Fanout 相關配置

@Configuration
public class FanoutRabbitConfig {

    @Bean
    public Queue AMessage() {
        return new Queue("fanout.A");
    }

    @Bean
    public Queue BMessage() {
        return new Queue("fanout.B");
    }

    @Bean
    public Queue CMessage() {
        return new Queue("fanout.C");
    }

    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    @Bean
    Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(AMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(BMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(CMessage).to(fanoutExchange);
    }

}

這裏使用了A、B、C三個隊列綁定到Fanout交換機上面，發送端的routing_key寫任何字符都會被忽略：

public void send() {
        String context = "hi, fanout msg ";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}

結果如下：

Sender : hi, fanout msg 
...
fanout Receiver B: hi, fanout msg 
fanout Receiver A  : hi, fanout msg 
fanout Receiver C: hi, fanout msg

結果說明，綁定到fanout交換機上面的隊列都收到了消息

springboot RabbitMQ 配置

相關推薦

springboot RabbitMQ 配置

接收 handle mes 客戶端 fin ng- 技術 msg framework 引用自 http://www.cnblogs.com/ityouknow/p/6120544.html 自己留一份 記錄一下 　　RabbitMQ 即一個消息隊列，主要是用來實現應

springboot+rabbitmq整合示例程

param resource pom del actor .cn pri 完全 pan 關於什麽是rabbitmq，請看另一篇文： http://www.cnblogs.com/boshen-hzb/p/6840064.html 一、新建maven工程：springboot

springBoot+RabbitMQ例子

boot code cer .... ebo set print return cor demo目錄 貼代碼 1.ProducerConfig.java package com.test.config; import org.springframework.amqp.

rabbitmq配置

exceptio onf 沒有 默認 announce slist public change exception package me.aizou.tour.guide.config;import lombok.extern.slf4j.Slf4j;import me.a

springboot(三)配置日誌

end classpath warn filename %d war append val cnblogs 工程代碼：https://github.com/showkawa/springBoot_2017/tree/master/spb-demo1.添加pom.xml依賴

springboot 外部配置<上篇>

port 默認 -- cnblogs 應用 def less set man SpringBoot允許在外部進行配置，讓你在不同的環境中運行相同的代碼。你可以通過屬性文件、YAML文件、環境變量和命令行來進行外部配置。屬性值可以直接通過@Value註入，並可以通過Sprin

springboot 外部配置<中篇> YAML文件代替屬性文件

set onf another ade list() factory oot 生產 環境 YAML是JSON格式的擴展集，是一種非常方便的用來定義屬性的文件格式。當你的項目有SnakeYAML庫時，Spring Boot將會自動支持它。 1、使用YAML文件 SpringF

springboot讀取配置文件的三種方式

spring boot 讀取外部properties文件方式一：@Configuration//@PropertySource("classpath:jdbc.properties")//@PropertySource("classpath:config/jdbc222222.properties")@Pr

springboot中配置主從redis

oca 配置文件 ping bean gap ons ng- class 如果 測試redis的主從配置 redis實例 文件夾名稱如下 redis_master_s redis_slaver1_s redis_slaver2_s redis.conf文件 master

springboot + mybatis配置分頁插件

XML control framework cti sarg location @override html apache 1：首先配置springboot +mybatis框架 參考：http://www.cnblogs.com/liyafei/p/791154

SpringBoot自動化配置的註解開關原理

err 過濾 turn prot 自動化 res addall fault mes 我們以一個最簡單的例子來完成這個需求：定義一個註解EnableContentService，使用了這個註解的程序會自動註入ContentService這個bean。 @Retention(R

Springboot讀取配置文件及自定義配置文件

val request wire odin 自定義配置文件 方式 -s bin ssi 1.創建maven工程，在pom文件中添加依賴 1 <parent> 2 <groupId>org.springframework.boot

springboot日誌配置Logback

mat classpath path 節點 %d .text fixed ger tps springboot可以通過application.yml或者application.properties中配置如下屬性指定日誌配置 #logback日誌配置xml存放路徑配置logg

springboot讀取配置文件解決亂碼問題

ati cat per force set enable parent class default 首先在application.properties裏面寫好要讀取的配置文件：name=小郭age=23sex=男然後繼續加入以下代碼：banner.charset=UTF-8

springBoot之配置文件的讀取以及過濾器和攔截器的使用

boolean nco 定義 www cheng handle mod HR out 前言 在之前的學習springBoot中，成功的實現了Restful風格的基本服務。但是想將之前的工程作為一個項目來說，那些是僅僅不夠的。可能還需要獲取自定義的配置以及添加過濾器和攔截器。

SpringBoot常用配置

類名 設定 conf text 指定日期 over RKE mobile 日期 前言：springboot集成了主流的第三方框架，但是需要使用springboot那一套配置方式。但是我這裏只列舉了非常非常常用的，可以看已發的幾篇博客，慢慢會補充。當然官方文檔裏也有相應的配置

SpringBoot bootstrap 配置文件沒有生效

body enc ring pre 結合 frame 今天 end ram 今天單獨使用SpringBoot，發現其中的bootstrap.properties文件無法生效，改成yaml格式也無濟於事。 最後調查發現原來是因為SpringBoot本身並不支持，需要和

SpringBoot之配置google kaptcha

back put 是否 [] produce 邊框 驗證 servlet ray 項目中引入POM： <dependency> <groupId>com.google.code.kaptcha</groupId> <

SpringBoot讀取配置文件（從classpath/file讀取yml/properties文件)

CA rop 使用配置 data ret fig color oot 讀取配置 一、讀取properties文件 使用配置項@PropertySource 二、讀取yml文件 啟動類添加下面代碼： @Bean public static P

SpringBoot 中配置 Druid

AS not 方法 view http web DC class import 這裏只舉了一個例子，其他按照官方文檔進行修改，使用 registration.addInitParameter 這個方法。 import com.alibaba.druid.support.ht