关于rabbitmq: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();
    }
}

消费者

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

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

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

路由模式

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

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

主程序

主程序中应用 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);
    }
}

主题模式

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

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

主程序

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

关于rabbitmq:RabbitMQ二RabbitMQ整合SpringBoot

新建我的项目

pom.xml

application.yml

主程序

简略模式

主程序

生产者

消费者

工作模式

主程序

生产者

消费者

ack模式

设置 ack 模式

手动执行确认操作

抓取数量

公布和订阅模式

主程序

生产者

消费者

路由模式

主程序

生产者

消费者

主题模式

主程序

生产者

消费者

RPC异步调用

主程序

服务端

客户端

评论

发表回复取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

关于rabbitmq:RabbitMQ二RabbitMQ整合SpringBoot

新建我的项目

pom.xml

application.yml

主程序

简略模式

主程序

生产者

消费者

工作模式

主程序

生产者

消费者

ack模式

设置 ack 模式

手动执行确认操作

抓取数量

公布和订阅模式

主程序

生产者

消费者

路由模式

主程序

生产者

消费者

主题模式

主程序

生产者

消费者

RPC异步调用

主程序

服务端

客户端

评论

发表回复 取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

发表回复取消回复