SpringBoot（十三）：SpringBoot整合RabbitMQ

阿新 • • 發佈：2019-01-24

如果對RabbitMQ不熟悉的，建議先看RabbitMQ系列教程。

一、環境準備

RabbitMQ 3.7.4
SpringBoot 1.5.10.RELEASE

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

application.yml

spring:
  rabbitmq:
    host: 192.168.239.128
    port: 5672
#    username: test
#    password: 123456
#    virtual-host: /vhost_test
#    publisher-confirms: true

二、簡單佇列

RabbitMQConfiguration.java 宣告佇列

package cn.saytime.config;

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

@Configuration
public class RabbitMQConfiguration {

    private static final String QUEUE_SIMPLE_NAME = "test_simple_queue";

    @Bean
    public Queue queue(){
        return new Queue(QUEUE_SIMPLE_NAME, false, false, false, null);
    }

}

消費者

package cn.saytime.listener.simple;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 消費者
 */
@RabbitListener(queues = "test_simple_queue")
@Component
public class SimpleRecv {

    @RabbitHandler
    public void process(String message) {
        System.out.println("[x] rev : " + message);
    }

}

測試生產者

package cn.saytime;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

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

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void testSimpleQueue() {
        String message = "Hello RabbitMQ !";
        amqpTemplate.convertAndSend("test_simple_queue", message);
        System.out.println("[x] send " + message + " ok");
    }

}

執行test：

[x] send Hello RabbitMQ ! ok
[x] rev : Hello RabbitMQ !

三、工作佇列

However, “Fair dispatch” is the default configuration for spring-amqp

公平分發模式在Spring-amqp中是預設的，這種情況也是日常工作中使用最為正常的，輪詢模式用的較少，區別在於prefetch預設是1，如果設定為0就是輪詢模式。

3.1 公平分發模式

RabbitMQConfiguration.java 宣告佇列

package cn.saytime.config;

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

@Configuration
public class RabbitMQConfiguration {

    private static final String QUEUE_WORK_NAME = "test_work_queue";

    @Bean
    public Queue workQueue(){
        return new Queue(QUEUE_WORK_NAME, false, false, false, null);
    }

}

消費者1

package cn.saytime.listener.workfair;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 消費者
 */
@RabbitListener(queues = "test_workfair_queue")
@Component
public class WorkRecv {

    @RabbitHandler
    public void process(String message){
        System.out.println("[1] rev : " + message);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

消費者2

package cn.saytime.listener.workfair;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 消費者
 */
@RabbitListener(queues = "test_workfair_queue")
@Component
public class WorkRecv2 {

    @RabbitHandler
    public void process(String message){
        System.out.println("[2] rev : " + message);
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

測試生產者

package cn.saytime;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

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

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void testWorkFairQueue(){
        for (int i = 0; i < 20; i++) {
            String message = "Hello RabbitMQ " + i;
            // 傳送訊息
            amqpTemplate.convertAndSend("test_workfair_queue", message);
            System.out.println(" [x] Sent '" + message + "'");
            try {
                Thread.sleep(i*100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

執行test：

 [x] Sent 'Hello RabbitMQ 0'
 [x] Sent 'Hello RabbitMQ 1'
[2] rev : Hello RabbitMQ 1
[1] rev : Hello RabbitMQ 0
 [x] Sent 'Hello RabbitMQ 2'
 [x] Sent 'Hello RabbitMQ 3'
 [x] Sent 'Hello RabbitMQ 4'
 [x] Sent 'Hello RabbitMQ 5'
[1] rev : Hello RabbitMQ 2
 [x] Sent 'Hello RabbitMQ 6'
[2] rev : Hello RabbitMQ 3
[1] rev : Hello RabbitMQ 4
 [x] Sent 'Hello RabbitMQ 7'
 [x] Sent 'Hello RabbitMQ 8'
[1] rev : Hello RabbitMQ 5
 [x] Sent 'Hello RabbitMQ 9'
[2] rev : Hello RabbitMQ 6
[1] rev : Hello RabbitMQ 7
 [x] Sent 'Hello RabbitMQ 10'
[1] rev : Hello RabbitMQ 8
 [x] Sent 'Hello RabbitMQ 11'
[2] rev : Hello RabbitMQ 9
[1] rev : Hello RabbitMQ 10
 [x] Sent 'Hello RabbitMQ 12'
[1] rev : Hello RabbitMQ 11
 [x] Sent 'Hello RabbitMQ 13'
[2] rev : Hello RabbitMQ 12
[1] rev : Hello RabbitMQ 13
 [x] Sent 'Hello RabbitMQ 14'
[1] rev : Hello RabbitMQ 14
 [x] Sent 'Hello RabbitMQ 15'
[1] rev : Hello RabbitMQ 15
 [x] Sent 'Hello RabbitMQ 16'
[2] rev : Hello RabbitMQ 16
 [x] Sent 'Hello RabbitMQ 17'
[1] rev : Hello RabbitMQ 17
 [x] Sent 'Hello RabbitMQ 18'
[2] rev : Hello RabbitMQ 18
 [x] Sent 'Hello RabbitMQ 19'
[1] rev : Hello RabbitMQ 19

公平分發模式測試正常。

3.2 輪詢分發模式

修改application.yml

spring:
  rabbitmq:
    host: 192.168.239.128
    port: 5672
#    username: test
#    password: 123456
#    virtual-host: /vhost_test
#    publisher-confirms: true
     listener:
      simple:
        prefetch: 0

 [x] Sent 'Hello RabbitMQ 0'
 [x] Sent 'Hello RabbitMQ 1'
[1] rev : Hello RabbitMQ 0
[2] rev : Hello RabbitMQ 1
 [x] Sent 'Hello RabbitMQ 2'
 [x] Sent 'Hello RabbitMQ 3'
 [x] Sent 'Hello RabbitMQ 4'
 [x] Sent 'Hello RabbitMQ 5'
[1] rev : Hello RabbitMQ 2
 [x] Sent 'Hello RabbitMQ 6'
[2] rev : Hello RabbitMQ 3
[1] rev : Hello RabbitMQ 4
 [x] Sent 'Hello RabbitMQ 7'
 [x] Sent 'Hello RabbitMQ 8'
[1] rev : Hello RabbitMQ 5
 [x] Sent 'Hello RabbitMQ 9'
[2] rev : Hello RabbitMQ 6
[1] rev : Hello RabbitMQ 7
 [x] Sent 'Hello RabbitMQ 10'
[1] rev : Hello RabbitMQ 8
 [x] Sent 'Hello RabbitMQ 11'
[2] rev : Hello RabbitMQ 9
[1] rev : Hello RabbitMQ 10
 [x] Sent 'Hello RabbitMQ 12'
[1] rev : Hello RabbitMQ 11
 [x] Sent 'Hello RabbitMQ 13'
[2] rev : Hello RabbitMQ 12
[1] rev : Hello RabbitMQ 13
 [x] Sent 'Hello RabbitMQ 14'
[1] rev : Hello RabbitMQ 14
 [x] Sent 'Hello RabbitMQ 15'
[1] rev : Hello RabbitMQ 15
 [x] Sent 'Hello RabbitMQ 16'
[2] rev : Hello RabbitMQ 16
 [x] Sent 'Hello RabbitMQ 17'
[1] rev : Hello RabbitMQ 17
 [x] Sent 'Hello RabbitMQ 18'
[2] rev : Hello RabbitMQ 18
 [x] Sent 'Hello RabbitMQ 19'
[1] rev : Hello RabbitMQ 19

可以看到消費者1消費偶數，消費者2消費奇數，表示輪詢分發正常。

四、訂閱模式

定義交換機、佇列、以及佇列與交換機的繫結關係。

package cn.saytime.config;

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

@Configuration
public class RabbitMQConfiguration {

    private static final String EXCHANGE_FANNOUT_NAME = "test_exchange_fanout";
    private static final String QUEUE_PS_SMS_NAME = "test_queue_fanout_sms";
    private static final String QUEUE_PS_EMAIL_NAME = "test_queue_fanout_email";

    @Bean("fanoutExchange")
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange(EXCHANGE_FANNOUT_NAME);
    }
    @Bean
    public Queue fanoutSmsQueue(){
        return new Queue(QUEUE_PS_SMS_NAME, false, false, false, null);
    }
    @Bean
    public Queue fanoutEmailQueue(){
        return new Queue(QUEUE_PS_EMAIL_NAME, false, false, false, null);
    }
    @Bean
    public Binding smsQueueExchangeBinding(FanoutExchange fanoutExchange, Queue fanoutSmsQueue){
        return BindingBuilder.bind(fanoutSmsQueue).to(fanoutExchange);
    }
    @Bean
    public Binding emailQueueExchangeBinding(FanoutExchange fanoutExchange, Queue fanoutEmailQueue){
        return BindingBuilder.bind(fanoutEmailQueue).to(fanoutExchange);
    }

}

消費者1 sms

package cn.saytime.listener.ps;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * sms消費者
 */
@RabbitListener(queues = "test_queue_fanout_sms")
@Component
public class SmsRecv {

    @RabbitHandler
    public void process(String message){
        System.out.println("[sms] rev : " + message);
    }

}

消費者2

package cn.saytime.listener.ps;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * email消費者
 */
@RabbitListener(queues = "test_queue_fanout_email")
@Component
public class EmailRecv {

    @RabbitHandler
    public void process(String message){
        System.out.println("[email] rev : " + message);
    }

}

測試訊息：

package cn.saytime;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

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

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void testFanoutQueue() {
        String message = "Hello, fanout message ";
        // 傳送訊息
        amqpTemplate.convertAndSend("test_exchange_fanout", "", message);
        System.out.println(" [x] Sent '" + message + "'");
    }

}

執行test結果：

 [x] Sent 'Hello, fanout message '
[sms] rev : Hello, fanout message 
[email] rev : Hello, fanout message

五、路由模式

跟訂閱模式一樣，只不過Configuration裡面配置的是DirectExchange，並設定路由鍵。

繫結關係：

BindingBuilder.bind(queue).to(exchange).with("info");

傳送訊息：

amqpTemplate.convertSendAndReceive(exchange, routingkey, message);

六、主題模式

跟路由模式一樣，只不過路由鍵可以模糊匹配，配置的是TopicExchange.

SpringBoot（十三）：SpringBoot整合RabbitMQ

如果對RabbitMQ不熟悉的，建議先看RabbitMQ系列教程。一、環境準備 RabbitMQ 3.7.4 SpringBoot 1.5.10.RELEASE <dependency>

springBoot（24）：集成rabbitmq

springboot rabbitmq 註意：springboot支持的amqp規範的中間件只有rabbitmq第一步：添加依賴 <dependency> <groupId>org.springframework.boot</

SpringBoot（四）：SpringBoot整合JdbcTemplate

個人其實偏向於使用類似於JdbcTemplate這種的框架，返回資料也習慣於接受Map/List形式，而不是轉化成物件，一是前後臺分離轉成json方便，另外是返回資料格式，資料欄位可以通過SQL控制，而不是返回整個物件欄位資料，或者通過VO方式。當然更多人習慣於採用Bea

SpringBoot（八）：SpringBoot整合Log4j

SpringBoot預設使用日誌框架logback 一、依賴  <dependency>

SpringBoot（五）：SpringBoot整合MyBatis

怎麼說了，寫部落格雖然是一件很費時間的事情，而且還是個菜鳥，但是如果寫的東西能夠幫助到別人，還是值得開心的。回顧：上篇寫了JdbcTemplate，但是想到使用Mybatis，JPA的人估計不少，所以這篇寫一下SpringBoot整合Mybatis，JPA的話以

SpringBoot（六）：SpringBoot使用CROS解決跨域問題

跨域問題其實之前就有說過，不明白的可以參考我之前的文章，這章主要講解SpringBoot CROS的支援。這裡主要按照官方文件方式講述。 SpringBoot CROS 參考： http://docs.spring.io/spring-boot/docs/1.5.4.R

SpringBoot（九）：SpringBoot使用FastJson

阿里巴巴FastJson是一個Json處理工具包，包括“序列化”和“反序列化”兩部分，它具備如下特徵：速度最快，測試表明，fastjson具有極快的效能，超越任其他的Java Json parser。包括自稱最快的JackJson；功能強大，完全支援Java Bean

SpringBoot（四）：SpringBoot與資料訪問（未完待續）

1、簡介對於資料訪問層，無論是SQL還是NOSQL，SpringBoot預設採用整合SpringData（SpringData是Spring 的一個子專案。用於簡化資料庫訪問，支援NoSQL 和關係資料儲存。其主要目標是使資料庫的訪問變得方便快捷）的方式進行

SpringBoot（十）：SpringBoot使用FastJson

阿里巴巴FastJson是一個Json處理工具包，包括“序列化”和“反序列化”兩部分，它具備如下特徵：速度最快，測試表明，fastjson具有極快的效能，超越任其他的Java Json p

RabbitMQ學習（三）：Spring整合RabbitMQ

1、maven依賴   <dependency>

SpringBoot（九）：SpringBoot使用Druid資料來源

Druid是阿里巴巴開源平臺上一個資料庫連線池實現，它結合了C3P0、DBCP、PROXOOL等DB池的優點，同時加入了日誌監控，可以很好的監控DB池連線和SQL的執行情況，可以說是針對監控而生的DB連線池(據說是目前最好的連線池) 一、依賴為

學習之路-RabbitMQ（三）：SpringBoot整合RabbitMQ

一：引入RabbitMQ的相關jar包： <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp

企業分布式微服務雲SpringCloud SpringBoot mybatis （十三）Spring Boot整合MyBatis

ech 字段 osc 操作 with public assert 連接 ref Spring中整合MyBatis就不多說了，最近大量使用Spring Boot，因此整理一下Spring Boot中整合MyBatis的步驟。搜了一下Spring Boot整合MyBatis的文

springboot（八）：整合多資料來源

資料來源是如何劃分：　　按包名（業務）和註解劃分；例如：com.test.learn01----使用datasource1； com.test.learn02----使用datasource2。專案目錄結構：配置檔案中新增兩個資料來源建立資料來源DataSource1

每天學點SpringCloud（十三）：SpringCloud-Stream整合RabbitMQ

我們知道，當微服務越來越來多的時候，僅僅是feign的http呼叫方式已經滿足不了我們的使用場景了。這個時候系統就需要接入訊息中介軟體了。相比較於傳統的Spring專案、SpringBoot專案使用訊息中介軟體的很多配置不同，SpringCloud Stream抽象了中介軟體產品的不同，在SpringClou

Spring boot入門（三）：SpringBoot整合結合AdminLTE(Freemarker)，利用generate自動生成程式碼，利用DataTable和PageHelper進行分頁顯示

　　關於SpringBoot和PageHelper，前篇部落格已經介紹過Spring boot入門（二）：Spring boot整合MySql,Mybatis和PageHelper外掛，前篇部落格大致講述了SpringBoot如何整合Mybatis和Pagehelper,但是沒有做出實際的範例，本篇部落格是連

SpringBoot學習筆記（五）：SpringBoot整合lombok工具、SpringBoot整合Shiro安全框架

SpringBoot整合lombok工具什麼是lombok？自動生成setget方法，建構函式，列印日誌官網：http://projectlombok.org/features/index. 平時我們寫的一些重複程式碼，比如每個實體類的setter，getter方法，給每個類寫

SpringBoot學習筆記（四）：SpringBoot整合Mybatis-Plus+程式碼生成

簡介官網：http://baomidou.oschina.io/mybatis-plus-doc/ 平時業務程式碼不復雜的時候我們寫什麼程式碼寫的最多，就是我們的SQL語句啊，配置那麼多的Mapper.xml，還要配置什麼resultMap這些東西，還要去管理paramtype

SpringBoot學習筆記（三）：SpringBoot整合Mybatis、SpringBoot事務管理、SpringBoot多資料來源

SpringBoot整合Mybatis 第一步我們需要在pom.xml裡面引入mybatis相關的jar包 <dependency> <groupId>org.mybatis.spring.boot</groupId> <artif

SpringBoot學習筆記（二）：SpringBoot訪問靜態檔案、捕獲全域性異常、整合Thymeleaf、整合JSP

SpringBoot訪問靜態檔案什麼是靜態檔案？不需要通過web容器去得到的檔案，直接通過路徑就能得到的檔案，比如專案的css,js,img等檔案。所有的資原始檔都應該在src/main/resources（maven專案中會自動建立這個資料夾）資料夾下面，但在SpringBo

SpringBoot（十三）：SpringBoot整合RabbitMQ

一、環境準備

二、簡單佇列

三、工作佇列

3.1 公平分發模式

3.2 輪詢分發模式

四、訂閱模式

五、路由模式

六、主題模式

相關推薦