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

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: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);
    }

}

複製程式碼

消費者2(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);
    }

}

複製程式碼

消費者B:

複製程式碼

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,收到了迴應,又傳回來了。