SpringBoot+RabbitMq實現生產者與消費者的多種情景

阿新 • • 發佈：2018-11-30

一、新建maven工程：springboot-rabbitmq

二、引入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>

三、在src/main/resources裡面新增application.properties

該配置檔案主要是對rabbimq的配置資訊

spring.application.name=springbootrabbitmq
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

Receiver1 : hellomsg:7Fri 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，收到了迴應，又傳回來了。

SpringBoot+RabbitMq實現生產者與消費者的多種情景

SpringBoot+RabbitMq實現生產者與消費者的多種情景

使用Python多線程實現生產者與消費者模型

佇列實現生產者與消費者

RabbitMQ學習——生產者與消費者入門例子

RabbitMQ實現生產者和消費者（帶註釋）

RabbitMQ的生產者與消費者

利用wait()和notify()實現生產者與消費者問題

作業系統-使用訊號量實現生產者與消費者（C++實現）

實現生產者與消費者模式

java多執行緒--ReentrantLock實現生產者與消費者模式

多執行緒實現生產者與消費者模式

IOS多執行緒使用GCD與訊號量實現生產者與消費者模式

訊息中介軟體——RabbitMQ（五）快速入門生產者與消費者，SpringBoot整合RabbitMQ！

生產者與消費者的Java實現

生產者與消費者程式碼實現 java

淺談Java簡單實現的生產者與消費者問題

java簡單實現非同步佇列:使用生產者與消費者模型

使用Condition物件實現執行緒同步，模擬生產者與消費者問題。

Springboot + rabbitMQ實現延遲佇列（消費者）

生產者與消費者問題C語言實現

SpringBoot+RabbitMq實現生產者與消費者的多種情景

相關推薦