2. 程式人生 > >spring boot+ rabbitMQ

spring boot+ rabbitMQ

阿新 • • 發佈：2018-12-19

引入springboot和rabbitmq的依賴

複製程式碼

<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.springboot.rabbitmq</groupId>
  <artifactId>springboot-rabbitmq</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <name>springboot-rabbitmq</name>
  <description>springboot-rabbitmq</description>
  
  <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.4.1.RELEASE</version>
  </parent>
  <dependencies>
     <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>  
     <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
  </dependencies>
</project>

複製程式碼

spring-boot-starter-test是為了後面寫測試類用，
spring-boot-starter-amqp才是真正的使用rabbitmq的依賴

三、在src/main/resources裡面新增application.properties
該配置檔案主要是對rabbimq的配置資訊

複製程式碼

 

spring.application.name=springboot-rabbitmq
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.virtual-host=/

 

複製程式碼

四、新建springboot主類Application

該類初始化建立佇列、轉發器，並把佇列繫結到轉發器

複製程式碼

package com.rabbit;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class Application {
    final static String queueName = "hello";

    @Bean
    public Queue helloQueue() {
        return new Queue("hello");
    }
    
    @Bean
    public Queue userQueue() {
        return new Queue("user");
    }
    
    //===============以下是驗證topic Exchange的佇列==========
    @Bean
    public Queue queueMessage() {
        return new Queue("topic.message");
    }

    @Bean
    public Queue queueMessages() {
        return new Queue("topic.messages");
    }
  //===============以上是驗證topic Exchange的佇列==========
    
    
    //===============以下是驗證Fanout Exchange的佇列==========
    @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");
    }
    //===============以上是驗證Fanout Exchange的佇列==========
    

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

    /**
     * 將佇列topic.message與exchange繫結，binding_key為topic.message,就是完全匹配
     * @param queueMessage
     * @param exchange
     * @return
     */
    @Bean
    Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
    }

    /**
     * 將佇列topic.messages與exchange繫結，binding_key為topic.#,模糊匹配
     * @param queueMessage
     * @param exchange
     * @return
     */
    @Bean
    Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
    }
    
    @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);
    }
    
    
   
    public static void main(String[] args) throws Exception {
        SpringApplication.run(Application.class, args);
    }
}

複製程式碼

五、各種情景實現

1、最簡單的hello生產和消費實現（單生產者和單消費者）

生產者：

複製程式碼

package com.rabbit.hello;

import java.util.Date;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class HelloSender1 {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String sendMsg = "hello1 " + new Date();
        System.out.println("Sender1 : " + sendMsg);
        this.rabbitTemplate.convertAndSend("helloQueue", sendMsg);
    }

}

複製程式碼

消費者:

複製程式碼

package com.rabbit.hello;

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

@Component
@RabbitListener(queues = "helloQueue")
public class HelloReceiver1 {

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

}

複製程式碼

controller：

複製程式碼

package com.rabbit.controller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.rabbit.hello.HelloSender1;

@RestController
@RequestMapping("/rabbit")
public class RabbitTest {
    
    @Autowired
    private HelloSender1 helloSender1;
    @Autowired
    private HelloSender1 helloSender2;
    
    @PostMapping("/hello")
    public void hello() {
        helloSender1.send();
    }
}

複製程式碼

啟動程式，執行：

結果如下：

Sender1 : hello1 Thu May 11 17:23:31 CST 2017
Receiver2  : hello1 Thu May 11 17:23:31 CST 2017

 

2、單生產者-多消費者

生產者：

複製程式碼

package com.rabbit.hello;

import java.util.Date;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class HelloSender1 {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send(String msg) {
        String sendMsg = msg + new Date();
        System.out.println("Sender1 : " + sendMsg);
        this.rabbitTemplate.convertAndSend("helloQueue", sendMsg);
    }

}

複製程式碼

消費者1：

複製程式碼

package com.rabbit.hello;

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

@Component
@RabbitListener(queues = "helloQueue")
public class HelloReceiver1 {

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

}

複製程式碼

消費者2：

複製程式碼

package com.rabbit.hello;

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

@Component
@RabbitListener(queues = "helloQueue")
public class HelloReceiver2 {

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

}

複製程式碼

controller:

複製程式碼

package com.rabbit.controller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.rabbit.hello.HelloSender1;

@RestController
@RequestMapping("/rabbit")
public class RabbitTest {
    
    @Autowired
    private HelloSender1 helloSender1;
    @Autowired
    private HelloSender1 helloSender2;
    
    @PostMapping("/hello")
    public void hello() {
        helloSender1.send("hello1");
    }
    
    /**
     * 單生產者-多消費者
     */
    @PostMapping("/oneToMany")
    public void oneToMany() {
        for(int i=0;i<10;i++){
            helloSender1.send("hellomsg:"+i);
        }
        
    }
}

複製程式碼

用post方式執行：

http://127.0.0.1:8080/rabbit/oneToMany

結果如下：

複製程式碼

Sender1 : hellomsg:0Thu May 11 17:37:59 CST 2017
Sender1 : hellomsg:1Thu May 11 17:37:59 CST 2017
Sender1 : hellomsg:2Thu May 11 17:37:59 CST 2017
Sender1 : hellomsg:3Thu May 11 17:37:59 CST 2017
Sender1 : hellomsg:4Thu May 11 17:37:59 CST 2017
Sender1 : hellomsg:5Thu May 11 17:37:59 CST 2017
Sender1 : hellomsg:6Thu May 11 17:37:59 CST 2017
Sender1 : hellomsg:7Thu May 11 17:37:59 CST 2017
Sender1 : hellomsg:8Thu May 11 17:37:59 CST 2017
Sender1 : hellomsg:9Thu May 11 17:37:59 CST 2017
Receiver2  : hellomsg:1Thu May 11 17:37:59 CST 2017
Receiver1  : hellomsg:0Thu May 11 17:37:59 CST 2017
Receiver1  : hellomsg:3Thu May 11 17:37:59 CST 2017
Receiver1  : hellomsg:4Thu May 11 17:37:59 CST 2017
Receiver1  : hellomsg:5Thu May 11 17:37:59 CST 2017
Receiver2  : hellomsg:2Thu May 11 17:37:59 CST 2017
Receiver1  : hellomsg:6Thu May 11 17:37:59 CST 2017
Receiver2  : hellomsg:7Thu May 11 17:37:59 CST 2017
Receiver2  : hellomsg:8Thu May 11 17:37:59 CST 2017
Receiver1  : hellomsg:9Thu May 11 17:37:59 CST 2017

複製程式碼

從以上結果可知，生產者傳送的10條訊息，分別被兩個消費者接收了

 

3、多生產者-多消費者

生產者1：

複製程式碼

package com.rabbit.hello;

import java.util.Date;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class HelloSender1 {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send(String msg) {
        String sendMsg = msg + new Date();
        System.out.println("Sender1 : " + sendMsg);
        this.rabbitTemplate.convertAndSend("helloQueue", sendMsg);
    }

}

複製程式碼

生產者2：

複製程式碼

package com.rabbit.hello;

import java.util.Date;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class HelloSender2 {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send(String msg) {
        String sendMsg = msg + new Date();
        System.out.println("Sender2 : " + sendMsg);
        this.rabbitTemplate.convertAndSend("helloQueue", sendMsg);
    }

}

複製程式碼

消費者1：

複製程式碼

package com.rabbit.hello;

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

@Component
@RabbitListener(queues = "helloQueue")
public class HelloReceiver1 {

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

}

複製程式碼

消費者2：

複製程式碼

package com.rabbit.hello;

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

@Component
@RabbitListener(queues = "helloQueue")
public class HelloReceiver2 {

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

}

複製程式碼

controller:

複製程式碼

    /**
     * 多生產者-多消費者
     */
    @PostMapping("/manyToMany")
    public void manyToMany() {
        for(int i=0;i<10;i++){
            helloSender1.send("hellomsg:"+i);
            helloSender2.send("hellomsg:"+i);
        }
        
    }

複製程式碼

 

用post方式執行：

http://127.0.0.1:8080/rabbit/manyToMany

結果如下：

複製程式碼

Sender1 : hellomsg:0Fri May 12 09:08:50 CST 2017
Sender2 : hellomsg:0Fri May 12 09:08:50 CST 2017
Sender1 : hellomsg:1Fri May 12 09:08:50 CST 2017
Sender2 : hellomsg:1Fri May 12 09:08:50 CST 2017
Sender1 : hellomsg:2Fri May 12 09:08:50 CST 2017
Sender2 : hellomsg:2Fri May 12 09:08:50 CST 2017
Sender1 : hellomsg:3Fri May 12 09:08:50 CST 2017
Sender2 : hellomsg:3Fri May 12 09:08:50 CST 2017
Sender1 : hellomsg:4Fri May 12 09:08:50 CST 2017
Sender2 : hellomsg:4Fri May 12 09:08:50 CST 2017
Sender1 : hellomsg:5Fri May 12 09:08:50 CST 2017
Sender2 : hellomsg:5Fri May 12 09:08:50 CST 2017
Sender1 : hellomsg:6Fri May 12 09:08:50 CST 2017
Sender2 : hellomsg:6Fri May 12 09:08:50 CST 2017
Sender1 : hellomsg:7Fri May 12 09:08:50 CST 2017
Sender2 : hellomsg:7Fri May 12 09:08:50 CST 2017
Sender1 : hellomsg:8Fri May 12 09:08:50 CST 2017
Sender2 : hellomsg:8Fri May 12 09:08:50 CST 2017
Sender1 : hellomsg:9Fri May 12 09:08:50 CST 2017
Sender2 : hellomsg:9Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:0Fri May 12 09:08:50 CST 2017
Receiver1  : hellomsg:0Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:1Fri May 12 09:08:50 CST 2017
Receiver1  : hellomsg:1Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:2Fri May 12 09:08:50 CST 2017
Receiver1  : hellomsg:2Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:3Fri May 12 09:08:50 CST 2017
Receiver1  : hellomsg:3Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:4Fri May 12 09:08:50 CST 2017
Receiver1  : hellomsg:4Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:5Fri May 12 09:08:50 CST 2017
Receiver1  : hellomsg:5Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:6Fri May 12 09:08:50 CST 2017
Receiver1  : hellomsg:6Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:7Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:8Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:8Fri May 12 09:08:50 CST 2017
Receiver1  : hellomsg:7Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:9Fri May 12 09:08:50 CST 2017
Receiver2  : hellomsg:9Fri May 12 09:08:50 CST 2017

複製程式碼

和一對多一樣，接收端仍然會均勻接收到訊息

 

4、實體類傳輸

springboot完美的支援物件的傳送和接收，不需要格外的配置。

實體類（必須實現序列化介面）：

複製程式碼

package com.rabbit.user;

import java.io.Serializable;

public class User implements Serializable{
        private String name;
        private String pass;
        public String getName() {
            return name;
        }
        public void setName(String name) {
            this.name = name;
        }
        public String getPass() {
            return pass;
        }
        public void setPass(String pass) {
            this.pass = pass;
        }
}

複製程式碼

 

生產者：

複製程式碼

package com.rabbit.user;

import java.util.Date;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class UserSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        User user=new User();
        user.setName("hzb");
        user.setPass("123456789");
        System.out.println("user send : " + user.getName()+"/"+user.getPass());
        this.rabbitTemplate.convertAndSend("userQueue", user);
    }

}

複製程式碼

消費者：

複製程式碼

package com.rabbit.user;

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

@Component
@RabbitListener(queues = "userQueue")
public class UserReceiver {

    @RabbitHandler
    public void process(User user) {
        System.out.println("user receive  : " + user.getName()+"/"+user.getPass());
    }

}

複製程式碼

controller:

複製程式碼

    /**
     * 實體類傳輸測試
     */
    @PostMapping("/userTest")
    public void userTest() {
           userSender.send();
    }

複製程式碼

用post方式執行：

http://127.0.0.1:8080/rabbit/userTest

結果如下：

user send : hzb/123456789
user receive  : hzb/123456789

 

5、topic ExChange示例

topic 是RabbitMQ中最靈活的一種方式，可以根據binding_key自由的繫結不同的佇列

首先對topic規則配置，這裡使用兩個佇列來測試（也就是在Application類中建立和繫結的topic.message和topic.messages兩個佇列），其中topic.message的bindting_key為

“topic.message”，topic.messages的binding_key為“topic.#”；

生產者：

複製程式碼

package com.rabbit.topic;

import java.util.Date;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class TopicSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String msg1 = "I am topic.mesaage msg======";
        System.out.println("sender1 : " + msg1);
        this.rabbitTemplate.convertAndSend("exchange", "topic.message", msg1);
        
        String msg2 = "I am topic.mesaages msg########";
        System.out.println("sender2 : " + msg2);
        this.rabbitTemplate.convertAndSend("exchange", "topic.messages", msg2);
    }

}

複製程式碼

消費者1（topic.message）

複製程式碼

package com.rabbit.topic;

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

@Component
@RabbitListener(queues = "topic.message")
public class topicMessageReceiver {

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

}

複製程式碼

消費者２（topic.messages）

複製程式碼

package com.rabbit.topic;

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

@Component
@RabbitListener(queues = "topic.messages")
public class topicMessagesReceiver {

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

}

複製程式碼

controller:

複製程式碼

    /**
     * topic exchange型別rabbitmq測試
     */
    @PostMapping("/topicTest")
    public void topicTest() {
           topicSender.send();
    }

複製程式碼

用post方式執行：

http://127.0.0.1:8080/rabbit/topicTest

結果如下：

sender1 : I am topic.mesaage msg======
sender2 : I am topic.mesaages msg########
topicMessageReceiver  : I am topic.mesaage msg======
topicMessagesReceiver  : I am topic.mesaage msg======
topicMessagesReceiver  : I am topic.mesaages msg########

由以上結果可知：sender1傳送的訊息,routing_key是“topic.message”，所以exchange裡面的繫結的binding_key是“topic.message”，topic.＃都符合路由規則;所以sender1

傳送的訊息，兩個佇列都能接收到；

sender2傳送的訊息，routing_key是“topic.messages”，所以exchange裡面的繫結的binding_key只有topic.＃都符合路由規則;所以sender2傳送的訊息只有佇列

topic.messages能收到。

 

 

6、fanout ExChange示例

Fanout 就是我們熟悉的廣播模式或者訂閱模式，給Fanout轉發器傳送訊息，綁定了這個轉發器的所有佇列都收到這個訊息。

這裡使用三個佇列來測試（也就是在Application類中建立和繫結的fanout.A、fanout.B、fanout.C）這三個佇列都和Application中建立的fanoutExchange轉發器繫結。

生產者：

複製程式碼

package com.rabbit.fanout;

import java.util.Date;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class FanoutSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String msgString="fanoutSender :hello i am hzb";
        System.out.println(msgString);
        this.rabbitTemplate.convertAndSend("fanoutExchange","abcd.ee", msgString);
    }

}

複製程式碼

消費者A:

複製程式碼

package com.rabbit.fanout;

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

@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {

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

}

複製程式碼

消費者Ｂ：

複製程式碼

package com.rabbit.fanout;

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

@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB {

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

}

複製程式碼

消費者C:

複製程式碼

package com.rabbit.fanout;

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

@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {

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

}

複製程式碼

controller:

複製程式碼

    /**
     * fanout exchange型別rabbitmq測試
     */
    @PostMapping("/fanoutTest")
    public void fanoutTest() {
           fanoutSender.send();
    }

複製程式碼

用post方式執行：

http://127.0.0.1:8080/rabbit/fanoutTest

結果如下：

fanoutSender :hello i am hzb
FanoutReceiverC  : fanoutSender :hello i am hzb
FanoutReceiverB  : fanoutSender :hello i am hzb
FanoutReceiverA  : fanoutSender :hello i am hzb

由以上結果可知：就算fanoutSender傳送訊息的時候，指定了routing_key為"abcd.ee"，但是所有接收者都接受到了訊息

 

7、帶callback的訊息傳送

增加回調處理，這裡不再使用application.properties預設配置的方式，會在程式中顯示的使用檔案中的配置資訊。該示例中沒有新建佇列和exchange，用的是第5節中的topic.messages佇列和exchange轉發器。消費者也是第5節中的topicMessagesReceiver

rabbitmq配置類：

複製程式碼

package com.rabbit.callback;

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Scope;

public class RabbitConfig {
    
    @Value("${spring.rabbitmq.host}")
    private String addresses;
    
    @Value("${spring.rabbitmq.port}")
    private String port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;

    @Value("${spring.rabbitmq.publisher-confirms}")
    private boolean publisherConfirms;
    
    @Bean
    public ConnectionFactory connectionFactory() {

        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(addresses+":"+port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        /** 如果要進行訊息回撥，則這裡必須要設定為true */
        connectionFactory.setPublisherConfirms(publisherConfirms);
        return connectionFactory;
    }
    
    @Bean
    /** 因為要設定回撥類，所以應是prototype型別，如果是singleton型別，則回撥類為最後一次設定 */
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplatenew() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        return template;
    }

}

複製程式碼

生產者：

複製程式碼

package com.rabbit.callback;

import java.util.Date;
import java.util.UUID;

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

@Component
public class CallBackSender implements  RabbitTemplate.ConfirmCallback{
    @Autowired
    private RabbitTemplate rabbitTemplatenew;
    public void send() {
        
        rabbitTemplatenew.setConfirmCallback(this);
        String msg="callbackSender : i am callback sender";
        System.out.println(msg );
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());  
        System.out.println("callbackSender UUID: " + correlationData.getId());  
        this.rabbitTemplatenew.convertAndSend("exchange", "topic.messages", msg, correlationData);  
    }

    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        // TODO Auto-generated method stub
        System.out.println("callbakck confirm: " + correlationData.getId());
    }
}

複製程式碼

消費者：第5節中的topicMessagesReceiver

controller:

    @PostMapping("/callback")
    public void callbak() {
        callBackSender.send();
    }

用post方式執行：

http://127.0.0.1:8080/rabbit/callback

結果如下：

callbackSender : i am callback sender
callbackSender UUID: cd0c80a6-4c65-4bf9-b4f8-f3b1180755d6
callbakck confirm: cd0c80a6-4c65-4bf9-b4f8-f3b1180755d6
topicMessagesReceiver  : callbackSender : i am callback sender

從上面可以看出callbackSender發出的UUID，收到了迴應，又傳回來了。

相關推薦

spring boot Rabbitmq集成,延時消息隊列實現

-c 轉發 包裝 let err 接下來 rec ger str 本篇主要記錄Spring boot 集成Rabbitmq，分為兩部分， 第一部分為創建普通消息隊列， 第二部分為延時消息隊列實現： spring boot提供對mq消息隊列支持amqp相關包，引入即可：

Spring Boot RabbitMQ 延遲消息實現完整版

throws 曾經 nfa header 面試 you ble 程序 記得 概述 曾經去網易面試的時候，面試官問了我一個問題，說 下完訂單後，如果用戶未支付，需要取消訂單，可以怎麽做 我當時的回答是，用定時任務掃描DB表即可。面試官不是很滿意，提出： 用定時任務無法

Spring Boot -- RabbitMQ

Spring Boot -- RabbitMQ 1. pom.xml 2. application.properties 3. config 4. 生產者 5. 消費者 6. 測試 發現物件訊息 一個生產者多個消費者 多個

Spring Boot + RabbitMQ死信機制實現延遲佇列

本文轉載自部落格：http://blog.csdn.net/nexttake/article/details/78607486 ------------------------------------------------------------------------------------

Spring Boot RabbitMQ 延遲訊息實現完整版

概述 曾經去網易面試的時候，面試官問了我一個問題，說 下完訂單後，如果使用者未支付，需要取消訂單，可以怎麼做 我當時的回答是，用定時任務掃描DB表即可。面試官不是很滿意，提出： 用定時任務無法做到準實時通知，有沒有其他辦法？ 我當時的回答是： 可以用

spring boot+ rabbitMQ

引入springboot和rabbitmq的依賴 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLo

spring boot rabbitmq 重試機制

spring.rabbitmq.listener.simple.retry.max-attempts=5 最大重試次數 spring.rabbitmq.listener.simple.retry.en

RabbitMq 集成 spring boot 消息隊列 入門Demo

spring boot rabbitmq 入門集成spring boot 集成 RabbitMq還是很方便的。現在來一個簡單的例子來集成rabbitmq。入門demo。主要概念：其中比較重要的概念有 4 個，分別為：虛擬主機，交換機，隊列，和綁定。虛擬主機：一個虛擬主機持有一組交換機、隊列和綁定。為什麽需要多

Spring Boot 揭秘與實戰（六） 消息隊列篇 - RabbitMQ

-i XML @service 我的文章 tro [] 創意 org ota 文章目錄 1. 什麽是 RabitMQ 2. Spring Boot 整合 RabbitMQ 3. 實戰演練4. 源代碼 3.1. 一個簡單的實戰開始 3.1.1. Configuration

rabbitMq與spring boot搭配實現監聽

address app caching prefix 前段時間 ever 不用 理解 its 　　在我前面有一篇博客說到了rabbitMq實現與zk類似的watch功能，但是那一篇博客沒有代碼實例，後面自己補了一個demo，便於理解。demo中主要利用spring boot

spring boot 1.5.4 整合rabbitMQ（十七）

rabbitmq springboot springboot1.5.4 springboot整合jsp springboot整合rabbitmq 上一篇：spring boot 1.5.4 整合redis、攔截器、過濾器、監聽器、靜態資源配置（十六） 關於rabbitMQ原理，請參閱博客：

spring boot實戰(第十二篇)整合RabbitMQ

this direct 還需要 添加屬性 創建 還需 topic start routing 前言 本篇主要講述Spring Boot與RabbitMQ的整合，內容非常簡單，純API的調用操作。 操作之間需要加入依賴Jar [html] view plain cop

Spring Boot (25) RabbitMQ消息隊列

res bin context 程序 了解 可靠 版本 prope message MQ全程（Message Queue）又名消息隊列，是一種異步通訊的中間件。可以理解為郵局，發送者將消息投遞到郵局，然後郵局幫我們發送給具體的接收者，具體發送過程和時間與我們無關，常見的MQ

Spring-boot集成RabbitMQ踩過的坑

inf 需要 info named logfile 遠程 nis 報錯 img 1.java.net.SocketException: socket closed 　　官方文檔已經說明，新建user和guest的賬戶是沒有遠程登錄的權限的 需要對登錄所用賬戶授權 　　　解決

【Spring Boot】（30）、SpringBoot整合RabbitMQ

1、安裝 1.1、Erlang： Erlang下載地址，下載後安裝即可。 1.2、RabbitMQ安裝 RabbitMQ下載地址，下載後安裝即可。 注意：Erlang的版本要與RabbitMQ版本需要匹配才行。 RabbitMQ Mini

Spring Boot項目配置RabbitMQ集群

user spa 配置 per work number host nbsp onf //具體參看了配置的源碼 org.springframework.boot.autoconfigure.amqp.RabbitProperties //RabbitMQ單機

Spring Boot專案配置RabbitMQ叢集

//具體參看了配置的原始碼 org.springframework.boot.autoconfigure.amqp.RabbitProperties //RabbitMQ單機 spring:   rabbitmq:     host: localh

Spring Boot AMQP+RabbitMq

主要解決兩個問題： 1如何傳送/釋出訊息(將java物件序列化為json)，通過使用RabbitTemplate 2如何讀取消費訊息（json反序列化為java物件）使用@RabbitListener註解 amqp是一個規範，一種定義了兩個系統間通過訊息進行互動的協議，而RabbitMQ

Spring Boot（十三）RabbitMQ安裝與集成

png println jin api 為什麽 廣播 exception 傳輸 測試框架 一、前言 RabbitMQ是一個開源的消息代理軟件（面向消息的中間件），它的核心作用就是創建消息隊列，異步接收和發送消息，MQ的全程是：Message Queue中文的意思是消息隊列。

spring-boot（八）RabbitMQ的使用

RabbitMQ 即一個訊息佇列，主要是用來實現應用程式的非同步和解耦，同時也能起到訊息緩衝，訊息分發的作用。 訊息中介軟體在網際網路公司的使用中越來越多，剛才還看到新聞阿里將RocketMQ捐獻給了apache，當然了今天的主角還是講RabbitMQ。訊息中介軟體最主要的作用是解耦，中介軟體最