RabbitMQ（二）--RabbitMQ整合SpringBoot

新建项目

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.4.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>cn.tedu</groupId>
    <artifactId>rabbitmq-springboot</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>rabbitmq-springboot</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

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

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

application.yml

spring:
  rabbitmq:
    host: 192.168.64.140
    username: admin
    password: admin

主程序

删除自动建立的主程序

咱们为每种模式建立一个包,在每一个包中建立各自的主程序,单独测试.

简单模式

主程序

Spring提供的Queue类,是队列的封装对象,它封装了队列的参数信息.

RabbitMQ的自动配置类,会发现这些Queue实例,并在RabbitMQ服务器中定义这些队列.

package cn.tedu.rabbitmqspringboot.m1;

import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

import javax.annotation.PostConstruct;

@SpringBootApplication
public class Main {
    @Autowired
    private Producer producer;

    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
    }

    @Bean
    public Queue helloworldQueue(){
        /**
         * 可用如下形式：
         * new Queue("helloworld") - 默认属性：持久（true），非排他（false），非自动删除（false）
         * new Queue("helloworld",false,false,false,null)
         */
        return new Queue("helloworld",false);//返回一个非持久队列
    }

    /**
     *@PostConstruct 方法会被自动执行，spring扫描建立了全部对象，并完成全部注入操做后会执行
     */
    @PostConstruct
    public void test(){
        producer.send();
        System.out.println("消息已经发送");
    }
}

生产者

AmqpTemplate是rabbitmq客户端API的一个封装工具,提供了简便的方法来执行消息操做.

AmqpTemplate由自动配置类自动建立

package cn.tedu.rabbitmqspringboot.m1;

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

@Component
public class Producer {
    // 在RabbtiAutoConfiguration 自动配置类中建立的工具对象
    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(){
        amqpTemplate.convertAndSend("helloworld", "Hello world!");
    }
}

消费者

经过@RabbitListener从指定的队列接收消息
使用@RebbitHandler注解的方法来处理消息

package cn.tedu.rabbitmqspringboot.m1;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
//@RabbitListener(queues = "helloworld")//这样使用须要用@RabbitHandler配合使用
public class Consumer {
    //@RabbitHandler//这样只能接收一个队列的消息，简单模式
 @RabbitListener(queues = "helloworld")//这种用法能够写多个方法接收多个队列，工厂模式
 public void receive(String msg){
        System.out.println("收到： " + msg);
 }
}

工做模式

主程序

在主程序中建立名为task_queue的持久队列

package cn.tedu.rabbitmqspringboot.m2;

import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

import javax.annotation.PostConstruct;

/**
 * 合理分发
 *      1.手动ack - springboot整合后默认就是手动ack模式
 *                  消费者方法执行成功后，springboot会帮助发送回执
 *      2.qos=1 - yml中配置prefetch
 * 持久化
 *      1.队列持久化
 *      2.消息持久化 - 默认是持久消息
 */
@SpringBootApplication
public class Main {
    @Autowired
    private Producer producer;

    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
    }

    @Bean
    public Queue taskQueue(){
        /**
         * 可用如下形式：
         * new Queue("helloworld") - 默认属性：持久（true），非排他（false），非自动删除（false）
         * new Queue("helloworld",false,false,false,null)
         */
        return new Queue("task_queue",true);//返回一个持久队列
    }

    /**
     *@PostConstruct 方法会被自动执行，spring扫描建立了全部对象，并完成全部注入操做后会执行
     */
    @PostConstruct
    public void test(){
        producer.send();
    }
}

生产者

package cn.tedu.rabbitmqspringboot.m2;

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

import java.util.Scanner;

@Component
public class Producer {
    // 在RabbtiAutoConfiguration 自动配置类中建立的工具对象
    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(){
//        new Thread(new Runnable() {
//            @Override
//            public void run() {
//
//            }
//        }).start();

        // lambda.匿名内部类的简写
        new Thread(() ->{
            while (true){//用单独的线程执行，不要影响主线程
                System.out.println("输入消息: ");
                String msg = new Scanner(System.in).nextLine();
                amqpTemplate.convertAndSend("task_queue",msg);
            }
        }).start();


    }
}

spring boot封装的 rabbitmq api 中, 发送的消息默认是持久化消息.
若是但愿发送非持久化消息, 须要在发送消息时作如下设置:

• 使用 MessagePostProcessor 前置处理器参数
• 从消息中获取消息的属性对象
• 在属性中把 DeliveryMode 设置为非持久化

//若是须要设置消息为非持久化,能够取得消息的属性对象,修改它的deliveryMode属性
    t.convertAndSend("task_queue", (Object) s, new MessagePostProcessor() {
        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            MessageProperties props = message.getMessageProperties();
            props.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
            return message;
        }
    });

消费者

package cn.tedu.rabbitmqspringboot.m2;

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

@Component
public class Consumer {

    @RabbitListener(queues = "task_queue")
    public void receive1(String msg){
        System.out.println("消费者1-收到： " + msg);
    }

    @RabbitListener(queues = "task_queue")
    public void receive2(String msg){
        System.out.println("消费者2-收到： " + msg);
    }
}

ack模式

在 spring boot 中提供了三种确认模式:

• NONE - 使用rabbitmq的自动确认
• AUTO - 使用rabbitmq的手动确认, springboot会自动发送确认回执 (默认)
• MANUAL - 使用rabbitmq的手动确认, 且必须手动执行确认操做

默认的 AUTO 模式中, 处理消息的方法抛出异常, 则表示消息没有被正确处理, 该消息会被从新发送.

设置 ack 模式

spring:
  rabbitmq:
    listener:
      simple:
        # acknowledgeMode: NONE # rabbitmq的自动确认
        acknowledgeMode: AUTO # rabbitmq的手动确认, springboot会自动发送确认回执 (默认)
        # acknowledgeMode: MANUAL # rabbitmq的手动确认, springboot不发送回执, 必须本身编码

手动执行确认操做

若是设置为 MANUAL 模式,必须手动执行确认操做

@RabbitListener(queues="task_queue")
    public void receive1(String s, Channel c, @Header(name=AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
        System.out.println("receiver1 - 收到: "+s);
        // 手动发送确认回执
        c.basicAck(tag, false);
    }

抓取数量

工做模式中, 为了合理地分发数据, 须要将 qos 设置成 1, 每次只接收一条消息, 处理完成后才接收下一条消息.

spring boot 中是经过 prefetch 属性进行设置, 改属性的默认值是 250.

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 # qos=1, 默认250

发布和订阅模式

主程序

建立 FanoutExcnahge 实例, 封装 fanout 类型交换机定义信息.

spring boot 的自动配置类会自动发现交换机实例, 并在 RabbitMQ 服务器中定义该交换机.

package cn.tedu.rabbitmqspringboot.m3;

import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

import javax.annotation.PostConstruct;


@SpringBootApplication
public class Main {
    @Autowired
    private Producer producer;

    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
    }

    @Bean
    public FanoutExchange logsExchange(){
        return new FanoutExchange("logs",false,false);//非持久，不自动删除
    }

    @PostConstruct
    public void test(){
        producer.send();
    }

}

生产者

生产者向指定的交换机 logs 发送数据.

不须要指定队列名或路由键, 即便指定也无效, 由于 fanout 交换机会向全部绑定的队列发送数据, 而不是有选择的发送.

package cn.tedu.rabbitmqspringboot.m3;

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

import java.util.Scanner;

@Component
public class Producer {

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(){

        new Thread(() ->{
            while (true){
                System.out.println("输入消息: ");
                String msg = new Scanner(System.in).nextLine();
                amqpTemplate.convertAndSend("logs","",msg);//向交换机发送消息
            }
        }).start();
    }
}

消费者

消费者须要执行如下操做:

1. 定义随机队列(随机命名,非持久,排他,自动删除)
2. 定义交换机(能够省略, 已在主程序中定义)
3. 将队列绑定到交换机

spring boot 经过注解完成以上操做:

@RabbitListener(bindings = @QueueBinding( //这里进行绑定设置
    value = @Queue, //这里定义随机队列,默认属性: 随机命名,非持久,排他,自动删除
    exchange = @Exchange(name = "logs", declare = "false") //指定 logs 交换机,由于主程序中已经定义,这里不进行定义
))

package cn.tedu.rabbitmqspringboot.m3;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class Consumer {
    //1.建立随机队列 2，指定交换机logs 3.绑定
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(),//队列，随即名，非持久，独占，自动删除队列
            exchange = @Exchange(name = "logs",declare = "false")//交换机,declare表示不定义交换机，只是使用
    ))
    public void receive1(String msg){
        System.out.println("消费者1-收到： " + msg);
    }

    //1.建立随机队列 2，指定交换机logs 3.绑定
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(),//队列，随即名，非持久，独占，自动删除队列
            exchange = @Exchange(name = "logs",declare = "false")//交换机,declare（false）表示不定义交换机，只是使用
    ))
    public void receive2(String msg){
        System.out.println("消费者2-收到： " + msg);
    }
}

路由模式

与发布和订阅模式代码相似, 只是作如下三点调整:

1. 使用 direct 交换机
2. 队列和交换机绑定时, 设置绑定键
3. 发送消息时, 指定路由键

主程序

主程序中使用 DirectExcnahge 对象封装交换机信息, spring boot 自动配置类会自动发现这个对象, 并在 RabbitMQ 服务器上定义这个交换机.

package cn.tedu.rabbitmqspringboot.m4;

import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

import javax.annotation.PostConstruct;


@SpringBootApplication
public class Main {
    @Autowired
    private Producer producer;

    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
    }

    @Bean
    public DirectExchange directExchange(){

        return new DirectExchange("direct_logs",false,false);//非持久，不自动删除
    }

    @PostConstruct
    public void test(){
        producer.send();
    }
}

生产者

生产者向指定的交换机发送消息, 并指定路由键.

package cn.tedu.rabbitmqspringboot.m4;

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

import java.util.Scanner;

@Component
public class Producer {

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(){

        new Thread(() ->{
            while (true){
                System.out.println("输入消息: ");
                String msg = new Scanner(System.in).nextLine();
                System.out.println("输入路由键: ");
                String key = new Scanner(System.in).nextLine();
                amqpTemplate.convertAndSend("direct_logs",key,msg);
            }
        }).start();
    }
}

消费者

消费者经过注解来定义随机队列, 绑定到交换机, 并指定绑定键:

@RabbitListener(bindings = @QueueBinding( // 这里作绑定设置
    value = @Queue, // 定义队列, 随机命名,非持久,排他,自动删除
    exchange = @Exchange(name = "direct_logs", declare = "false"), // 指定绑定的交换机,主程序中已经定义过队列,这里不进行定义
    key = {"error","info","warning"} // 设置绑定键
))

package cn.tedu.rabbitmqspringboot.m4;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class Consumer {
    //1.建立随机队列 2，指定交换机logs 3.绑定
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(),//队列，随即名，非持久，独占，自动删除队列
            exchange = @Exchange(name = "direct_logs",declare = "false"),//交换机,declare表示不定义交换机，只是使用
            key = {"error"} //设置绑定键
    ))
    public void receive1(String msg){
        System.out.println("消费者1-收到： " + msg);
    }

    //1.建立随机队列 2，指定交换机logs 3.绑定
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(),//队列，随即名，非持久，独占，自动删除队列
            exchange = @Exchange(name = "direct_logs",declare = "false"),//交换机,declare（false）表示不定义交换机，只是使用
            key = {"error","info","warning"} //设置绑定键
    ))
    public void receive2(String msg){
        System.out.println("消费者2-收到： " + msg);
    }
}

主题模式

主题模式不过是具备特殊规则的路由模式, 代码与路由模式基本相同, 只作以下调整:

1. 使用 topic 交换机
2. 使用特殊的绑定键和路由键规则

主程序

package cn.tedu.rabbitmqspringboot.m5;

import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

import javax.annotation.PostConstruct;


@SpringBootApplication
public class Main {
    @Autowired
    private Producer producer;

    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
    }

    @Bean
    public TopicExchange directExchange(){

        return new TopicExchange("topic_logs",false,false);//非持久，不自动删除
    }

    @PostConstruct
    public void test(){
        producer.send();
    }
}

生产者

package cn.tedu.rabbitmqspringboot.m5;

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

import java.util.Scanner;

@Component
public class Producer {

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(){

        new Thread(() ->{
            while (true){
                System.out.println("输入消息: ");
                String msg = new Scanner(System.in).nextLine();
                System.out.println("输入路由键: ");
                String key = new Scanner(System.in).nextLine();
                amqpTemplate.convertAndSend("topic_logs",key,msg);
            }
        }).start();
    }
}

消费者

package cn.tedu.rabbitmqspringboot.m5;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class Consumer {
    //1.建立随机队列 2，指定交换机logs 3.绑定
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(),//队列，随即名，非持久，独占，自动删除队列
            exchange = @Exchange(name = "topic_logs",declare = "false"),//交换机,declare表示不定义交换机，只是使用
            key = {"*.orange.*"} //设置绑定键
    ))
    public void receive1(String msg){
        System.out.println("消费者1-收到： " + msg);
    }

    //1.建立随机队列 2，指定交换机logs 3.绑定
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(),//队列，随即名，非持久，独占，自动删除队列
            exchange = @Exchange(name = "topic_logs",declare = "false"),//交换机,declare（false）表示不定义交换机，只是使用
            key = {"*.*.rabbit","lazy.#"} //设置绑定键
    ))
    public void receive2(String msg){
        System.out.println("消费者2-收到： " + msg);
    }
}

RPC异步调用

主程序

主程序中定义两个队列

• 发送调用信息的队列: rpc_queue
• 返回结果的队列: 随机命名

package cn.tedu.m6;

import java.util.UUID;

import org.springframework.amqp.core.Queue;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class Main {

    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
    }
    @Bean
    public Queue sendQueue() {
        return new Queue("rpc_queue",false);
    }
    @Bean
    public Queue rndQueue() {
        return new Queue(UUID.randomUUID().toString(), false);
    }
}

服务端

从rpc_queue接收调用数据, 执行运算求斐波那契数,并返回计算结果.
@Rabbitlistener注解对于具备返回值的方法:

• 会自动获取 replyTo 属性
• 自动获取 correlationId 属性
• 向 replyTo 属性指定的队列发送计算结果, 并携带 correlationId 属性

package cn.tedu.m6;

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

@Component
public class RpcServer {
    @RabbitListener(queues = "rpc_queue")
    public long getFbnq(int n) {
        return f(n);
    }

    private long f(int n) {
        if (n==1 || n==2) {
            return 1;
        }
        return f(n-1) + f(n-2);
    }
}

客户端

使用 SPEL 表达式获取随机队列名: "#{rndQueue.name}"

发送调用数据时, 携带随机队列名和correlationId

从随机队列接收调用结果, 并获取correlationId

package cn.tedu.m6;

import java.util.UUID;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
public class RpcClient {
    @Autowired
    AmqpTemplate t;
    
    @Value("#{rndQueue.name}")
    String rndQueue;
    
    public void send(int n) {
        // 发送调用信息时, 经过前置消息处理器, 对消息属性进行设置, 添加返回队列名和关联id
        t.convertAndSend("rpc_queue", (Object)n, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties p = message.getMessageProperties();
                p.setReplyTo(rndQueue);
                p.setCorrelationId(UUID.randomUUID().toString());
                return message;
            }
        });
    }
    
    //从随机队列接收计算结果
    @RabbitListener(queues = "#{rndQueue.name}")
    public void receive(long r, @Header(name=AmqpHeaders.CORRELATION_ID) String correlationId) {
        System.out.println("nn"+correlationId+" - 收到: "+r);
    }
    
}