一、 Springboot 整合 RabbitMQ

别离以六种工作模式进行剖析

1. 新建我的项目

2. 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>

3. application.yml

spring:  rabbitmq:    host: #192.168.64.140  wht6.cn    username: admin    password: admin    #设置每回只抓取一条音讯,qos=1, 默认250    listener:      simple:        prefetch: 1

4. 主程序

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

1. 简略模式

1.1 主程序

Spring提供的Queue类,是队列的封装对象,它封装了队列的参数信息.RabbitMQ的主动配置类,会发现这些Queue实例,并在RabbitMQ服务器中定义这些队列.

import org.springframework.amqp.core.Queue;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.annotation.Bean;@SpringBootApplicationpublic class Main {    public static void main(String[] args) {        SpringApplication.run(Main.class, args);    }    @Bean    public Queue task_queue() {        /*         * 可用以下模式:          * new Queue("helloworld") - 长久,非排他,非主动删除         * new Queue("helloworld",false,false,false,null)         */        return new Queue("helloworld",false);    }}

1.2 生产者

AmqpTemplate是rabbitmq客户端API的一个封装工具,提供了简便的办法来执行音讯操作.AmqpTemplate由主动配置类主动创立

import org.springframework.amqp.core.AmqpTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;@Componentpublic class SimpleSender {    @Autowired    AmqpTemplate t;    public void send() {        // 这里向 helloworld 队列发送音讯        t.convertAndSend("helloworld", "Hello world!! "+System.currentTimeMillis());        System.out.println("音讯已发送");    }}

1.3 消费者

通过@RabbitListener从指定的队列接管音讯应用@RebbitHandler注解的办法来解决音讯

import org.springframework.amqp.rabbit.annotation.RabbitHandler;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Component@RabbitListener(queues = "helloworld")public class SimpleReceiver {    @RabbitHandler    public void receive(String msg) {        System.out.println("收到: "+msg);    }}

另外,@RabbitListener 注解中也能够间接定义队列:

 @RabbitListener(queuesToDeclare = @Queue(name = "helloworld",durable = "false"))

2. 工作模式

2.1 主程序

在主程序中创立名为task_queue长久队列

import org.springframework.amqp.core.Queue;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.annotation.Bean;@SpringBootApplicationpublic class Main {    public static void main(String[] args) {        SpringApplication.run(Main.class, args);    }    @Bean    public Queue task_queue() {        // 这个构造方法创立的队列参数为: 长久,非排他,非主动删除        return new Queue("task_queue");    }}

2.2 生产者

import java.util.Scanner;import org.springframework.amqp.core.AmqpTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;@Componentpublic class WorkSender {    @Autowired    AmqpTemplate t;        public void send() {        while (true) {            System.out.print("输出:");            String s = new Scanner(System.in).nextLine();                        //spring 默认将音讯的 DeliveryMode 设置为 PERSISTENT 长久化,            t.convertAndSend("task_queue", s);        }    }}

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

2.3 消费者

import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Componentpublic class WorkReceiver1 {    @RabbitListener(queues="task_queue")    public void receive1(String s) throws Exception {        System.out.println("receiver1 - 收到: "+s);        for (int i = 0; i < s.length(); i++) {            if (s.charAt(i) == '.') {                Thread.sleep(1000);            }        }    }    @RabbitListener(queues="task_queue")    public void receive2(String s) throws Exception {        System.out.println("receiver2 - 收到: "+s);        for (int i = 0; i < s.length(); i++) {            if (s.charAt(i) == '.') {                Thread.sleep(1000);            }        }    }}

2.4 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);        for (int i = 0; i < s.length(); i++) {            if (s.charAt(i) == '.') {                Thread.sleep(1000);            }        }        // 手动发送确认回执        c.basicAck(tag, false);    }

抓取数量

工作模式中, 为了正当地散发数据, 须要将 qos 设置成 1, 每次只接管一条音讯, 解决实现后才接管下一条音讯.
spring boot 中是通过 prefetch 属性进行设置, 改属性的默认值是 250.

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

3. 公布和订阅模式

3.1 主程序

创立 FanoutExcnahge 实例, 封装 fanout 类型交换机定义信息.spring boot 的主动配置类会主动发现交换机实例, 并在 RabbitMQ 服务器中定义该交换机.

import org.springframework.amqp.core.FanoutExchange;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.annotation.Bean;@SpringBootApplicationpublic class Main {    public static void main(String[] args) {        SpringApplication.run(Main.class, args);    }    @Bean    public FanoutExchange fanoutExchange() {        return new FanoutExchange("logs");    }}

3.2 生产者

生产者向指定的交换机 logs 发送数据.不须要指定队列名或路由键, 即便指定也有效, 因为 fanout 替换机会向所有绑定的队列发送数据, 而不是有抉择的发送.

import java.util.Scanner;import org.springframework.amqp.core.AmqpTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;@Componentpublic class Publisher {    @Autowired    AmqpTemplate t;    public void send() {        while (true) {            System.out.print("输出:");            String s = new Scanner(System.in).nextLine();            // 指定向 logs 交换机发送, 不指定队列名或路由键            t.convertAndSend("logs","",s);        }    }}

3.3 消费者

消费者须要执行以下操作:

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

spring boot 通过注解实现以上操作:

@RabbitListener(bindings = @QueueBinding( //这里进行绑定设置    value = @Queue, //这里定义随机队列,默认属性: 随机命名,非长久,排他,主动删除    exchange = @Exchange(name = "logs", declare = "false") //指定 logs 交换机,因为主程序中曾经定义,这里不进行定义))
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;@Componentpublic class Subscriber {    @RabbitListener(bindings = @QueueBinding(value = @Queue, exchange = @Exchange(name = "logs", declare = "false")))    public void receive1(String s) throws Exception {        System.out.println("receiver1 - 收到: "+s);    }    @RabbitListener(bindings = @QueueBinding(value = @Queue, exchange = @Exchange(name = "logs", declare = "false")))    public void receive2(String s) throws Exception {        System.out.println("receiver2 - 收到: "+s);    }}

4. 路由模式

与公布和订阅模式代码相似, 只是做以下三点调整:

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

4.1 主程序

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

import org.springframework.amqp.core.DirectExchange;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.annotation.Bean;@SpringBootApplicationpublic class Main {    public static void main(String[] args) {        SpringApplication.run(Main.class, args);    }    @Bean    public DirectExchange fanoutExchange() {        return new DirectExchange("direct_logs");    }}

4.2 生产者

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

import java.util.Scanner;import org.springframework.amqp.core.AmqpTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;@Componentpublic class RouteSender {    @Autowired    AmqpTemplate t;    public void send() {        while (true) {            System.out.print("输出音讯:");            String s = new Scanner(System.in).nextLine();            System.out.print("输出路由键:");            String key = new Scanner(System.in).nextLine();            // 第二个参数指定路由键            t.convertAndSend("direct_logs",key,s);        }    }}

4.3 消费者

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

@RabbitListener(bindings = @QueueBinding( // 这里做绑定设置    value = @Queue, // 定义队列, 随机命名,非长久,排他,主动删除    exchange = @Exchange(name = "direct_logs", declare = "false"), // 指定绑定的交换机,主程序中曾经定义过队列,这里不进行定义    key = {"error","info","warning"} // 设置绑定键))
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;@Componentpublic class RouteReceiver {    @RabbitListener(bindings = @QueueBinding(value = @Queue,exchange = @Exchange(name = "direct_logs", declare = "false"),key = {"error"}))    public void receive1(String s) throws Exception {        System.out.println("receiver1 - 收到: "+s);    }    @RabbitListener(bindings = @QueueBinding(value = @Queue, exchange = @Exchange(name = "direct_logs", declare = "false"),key = {"error","info","warning"}))    public void receive2(String s) throws Exception {        System.out.println("receiver2 - 收到: "+s);    }}

5. 主题模式

主题模式不过是具备非凡规定的路由模式, 代码与路由模式基本相同, 只做如下调整:

  1. 应用 topic 交换机
  2. 应用非凡的绑定键和路由键规定

5.1 主程序

import org.springframework.amqp.core.TopicExchange;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.annotation.Bean;@SpringBootApplicationpublic class Main {    public static void main(String[] args) {        SpringApplication.run(Main.class, args);    }    @Bean    public TopicExchange fanoutExchange() {        return new TopicExchange("topic_logs");    }}

5.2 生产者

import java.util.Scanner;import org.springframework.amqp.core.AmqpTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;@Componentpublic class TopicSender {    @Autowired    AmqpTemplate t;    public void send() {        while (true) {            System.out.print("输出音讯:");            String s = new Scanner(System.in).nextLine();            System.out.print("输出路由键:");            String key = new Scanner(System.in).nextLine();                        t.convertAndSend("topic_logs",key,s);        }    }}

5.3 消费者

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;@Componentpublic class TopicReceiver {    @RabbitListener(bindings = @QueueBinding(value = @Queue,exchange = @Exchange(name = "topic_logs", declare = "false"),key = {"*.orange.*"}))    public void receive1(String s) throws Exception {        System.out.println("receiver1 - 收到: "+s);    }    @RabbitListener(bindings = @QueueBinding(value = @Queue, exchange = @Exchange(name = "topic_logs", declare = "false"),key = {"*.*.rabbit","lazy.#"}))    public void receive2(String s) throws Exception {        System.out.println("receiver2 - 收到: "+s);    }}

6. RPC异步调用

6.1 主程序

主程序中定义两个队列

  • 发送调用信息的队列: rpc_queue
  • 返回后果的队列: 随机命名
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;@SpringBootApplicationpublic 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);    }} 

6.2 服务端

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

  • 会主动获取 replyTo 属性
  • 主动获取 correlationId 属性
  • replyTo 属性指定的队列发送计算结果, 并携带 correlationId 属性
import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Compoentpublic 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);    }}

6.3 客户端

应用 SPEL 表达式获取随机队列名: "#{rndQueue.name}"
发送调用数据时, 携带随机队列名和correlationId
从随机队列接管调用后果, 并获取correlationId

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;@Componentpublic 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);    }}