第一步：在專案pom.xml檔案中，新增pring-boot-starter-amqp 依賴

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

第二步：在專案application.properties檔案中，新增rabbitmq相關配置

##########RabbitMq配置##########
#是訪問port不是15672,15672是api和管理介面的port  
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=test
spring.rabbitmq.password=test
#如果要進行訊息回撥，則這裡必須要設定為true  
spring.rabbitmq.publisherconfirms=true
spring.rabbitmq.virtual-host=/
#最小訊息監聽執行緒數
spring.rabbitmq.listener.concurrency=2
#最大訊息監聽執行緒數
spring.rabbitmq.listener.max-concurrency=2
 

第三步：新增rabbitmq配置類和生產者類（生產訊息）、消費者類（消費訊息）

rabbitmq配置類

package com.fhsn.weixin.config.rabbitmq;

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

@Configuration
public class RabbitMqConfig {
    
    //宣告佇列
    @Bean
    public Queue queue1() {
        return new Queue("hello.queue1", true); // true表示持久化該佇列
    }
    
    @Bean
    public Queue queue2() {
        return new Queue("hello.queue2", true);
    }
    
    //宣告互動器
    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange("topicExchange");
    }

    //繫結
    @Bean
    public Binding binding1() {
        return BindingBuilder.bind(queue1()).to(topicExchange()).with("key.1");
    }
    
    @Bean
    public Binding binding2() {
        return BindingBuilder.bind(queue2()).to(topicExchange()).with("key.#");
    }
    
}
 

生產者類（生產訊息）

package com.fhsn.weixin.config.rabbitmq;

import java.util.UUID;

import javax.annotation.PostConstruct;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class Sender implements RabbitTemplate.ConfirmCallback, ReturnCallback {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {  
            System.out.println("訊息傳送成功:" + correlationData);  
        } else {  
            System.out.println("訊息傳送失敗:" + cause);  
        }  
        
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println(message.getMessageProperties().getCorrelationIdString() + " 傳送失敗");
        
    }

    //傳送訊息，不需要實現任何介面，供外部呼叫。
    public void send(String msg){
        
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        
        System.out.println("開始傳送訊息 : " + msg.toLowerCase());
        String response = rabbitTemplate.convertSendAndReceive("topicExchange", "key.1", msg, correlationId).toString();
        System.out.println("結束髮送訊息 : " + msg.toLowerCase());
        System.out.println("消費者響應 : " + response + " 訊息處理完成");
    }
}

 

消費者類（消費訊息）

package com.fhsn.weixin.config.rabbitmq;

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

@Component
public class Receiver {
    
    @RabbitListener(queues = "hello.queue1")
    public String processMessage1(String msg) {
        System.out.println(Thread.currentThread().getName() + " 接收到來自hello.queue1佇列的訊息：" + msg);
        return msg.toUpperCase();
    }
    
    @RabbitListener(queues = "hello.queue2")
    public void processMessage2(String msg) {
        System.out.println(Thread.currentThread().getName() + " 接收到來自hello.queue2佇列的訊息：" + msg);
    }
}

第四步：

在專案pom.xml檔案中新增 spring-boot-starter-test依賴

  <dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-test</artifactId>
        <version>1.5.9.RELEASE</version>
        <scope>test</scope>
 </dependency>

第五步：新增測試類

package com.fhsn.weixn.controller.rabbitmq;

import java.util.Date;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import com.fhsn.WeixinWebApplication;
import com.fhsn.weixin.config.rabbitmq.Sender;


@RunWith(value=SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = WeixinWebApplication.class)
public class RabbitMqTests {
    
    @Autowired
    private Sender sender;
    
    @Test
    public void sendTest() throws Exception {
        while(true){
            String msg = new Date().toString();
            sender.send(msg);
            Thread.sleep(1000);
        }
    }
}

第六步 : 右擊選中     RabbitMqTests->Run As->Junit Test 執行測試類,檢視console控制檯輸出結果如下,即表示springboot成功整合rabbitmq

開始傳送訊息 : wed may 16 14:43:40 cst 2018
SimpleAsyncTaskExecutor-1 接收到來自hello.queue1佇列的訊息：Wed May 16 14:43:40 CST 2018
SimpleAsyncTaskExecutor-1 接收到來自hello.queue2佇列的訊息：Wed May 16 14:43:40 CST 2018
結束髮送訊息 : wed may 16 14:43:40 cst 2018
消費者響應 : WED MAY 16 14:43:40 CST 2018 訊息處理完成
開始傳送訊息 : wed may 16 14:43:42 cst 2018
SimpleAsyncTaskExecutor-2 接收到來自hello.queue1佇列的訊息：Wed May 16 14:43:42 CST 2018
SimpleAsyncTaskExecutor-2 接收到來自hello.queue2佇列的訊息：Wed May 16 14:43:42 CST 2018
結束髮送訊息 : wed may 16 14:43:42 cst 2018
消費者響應 : WED MAY 16 14:43:42 CST 2018 訊息處理完成
開始傳送訊息 : wed may 16 14:43:43 cst 2018
SimpleAsyncTaskExecutor-1 接收到來自hello.queue1佇列的訊息：Wed May 16 14:43:43 CST 2018
SimpleAsyncTaskExecutor-1 接收到來自hello.queue2佇列的訊息：Wed May 16 14:43:43 CST 2018
結束髮送訊息 : wed may 16 14:43:43 cst 2018
消費者響應 : WED MAY 16 14:43:43 CST 2018 訊息處理完成
開始傳送訊息 : wed may 16 14:43:44 cst 2018
SimpleAsyncTaskExecutor-2 接收到來自hello.queue2佇列的訊息：Wed May 16 14:43:44 CST 2018
SimpleAsyncTaskExecutor-2 接收到來自hello.queue1佇列的訊息：Wed May 16 14:43:44 CST 2018
結束髮送訊息 : wed may 16 14:43:44 cst 2018
消費者響應 : WED MAY 16 14:43:44 CST 2018 訊息處理完成

參考文章來源：