脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → SpringAMQP消息队列

SpringAMQP消息队列(SpringBoot集成RabbitMQ方式)

 更新时间：2024年04月30日 09:14:38   作者：梁山教父  
这篇文章主要介绍了SpringAMQP消息队列(SpringBoot集成RabbitMQ方式),具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教

一、初始配置

1、导入maven坐标

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

2、yml配置

spring:
    rabbitmq:
        host: 你的rabbitmq的ip
        port: 5672
        username: guest
        password: guest

二、基本消息队列

1、创建队列

访问接口：http://localhost:15672，账号密码都为guest

进入后左下角有Add queue添加队列，我已添加队列为MqTest1

2、发布消息

@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @Test
    void contextLoads() {
        String queue="MqTest1";
        String message="message1";
        rabbitTemplate.convertAndSend(queue,message);
    }
 
}

此时可以看到队列有一个消息

3、接受消息

package com.rabbitmqdemoconsumer.rabbitmq;
 
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
@Component
public class SpringRabbitLeistener {
 
    @RabbitListener(queues = "MqTest1")
    public void listenSimpleQueueMessage(String msg){
        System.out.println("接收到的消息："+msg);
    }
}

此时控制台输出接收到的消息

三、工作消息队列（Work Queue）

可以提高消息处理速度，避免队列消息堆积

1、发布消息

@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @Test
    void contextLoads() {
        String queue="MqTest1";
        String message="message1";
        for (int i=0;i<10;i++){
            rabbitTemplate.convertAndSend(queue,message);
        }
    }
 
}

此时队列有10条消息

2、接受消息

package com.rabbitmqdemoconsumer.rabbitmq;
 
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
@Component
public class SpringRabbitLeistener {
 
    @RabbitListener(queues = "MqTest1")
    public void listenSimpleQueueMessage1(String msg){
        System.out.println("consume1接收到的消息："+msg);
    }
    @RabbitListener(queues = "MqTest1")
    public void listenSimpleQueueMessage2(String msg){
        System.out.println("consume2接收到的消息："+msg);
    }
}

3、控制台输出结果

consume1接收到的消息：message1
consume2接收到的消息：message1
consume1接收到的消息：message1
consume2接收到的消息：message1
consume1接收到的消息：message1
consume2接收到的消息：message1
consume1接收到的消息：message1
consume2接收到的消息：message1
consume1接收到的消息：message1
consume2接收到的消息：message1

4、消息预取问题

但是此时有一个问题就是消息预取，比如队列有10条消息，两个消费者各自直接先预取5个消息，如果一个消费者接受消息的速度慢，一个快，就会导致一个消费者已经完成工作，另一个还在慢慢处理，会造成消息堆积消费者身上，要解决这个问题需要在yml文件配置相关配置

  rabbitmq:
    host: 43.140.244.236
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    listener:
      simple:
        prefetch: 1 #每次只能取一个，处理完才能取下一个消息

这样可以避免消息预取导致堆积

四、发布订阅模式

exchange是交换机，负责消息路由，但不存储消息，路由失败则消息丢失

五、发布订阅模式之广播模式（Fanout）

1、Fanout配置类（@Bean声明）

package com.rabbitmqdemoconsumer.config;
 
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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class FanountConfig {
    //交换机声明
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("FanountExchange");
    }
    //声明队列1
    @Bean
    public Queue Fanount_Qeueue1(){
        return new Queue("Fanount_Qeueue1");
    }
    //声明队列2
    @Bean
    public Queue Fanount_Qeueue2(){
        return new Queue("Fanount_Qeueue2");
    }
    //绑定交换机和队列
    @Bean
    public Binding bindingFanount_Qeueue1(Queue Fanount_Qeueue1,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(Fanount_Qeueue1).to(fanoutExchange);
    }
    @Bean
    public Binding bindingFanount_Qeueue2(Queue Fanount_Qeueue2,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(Fanount_Qeueue2).to(fanoutExchange);
    }
}

可以看到声明的队列

已经声明的交换机（第一个）

绑定关系

2、发送消息

首先发送10条消息，经过交换机转发到队列

@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @Test
    void contextLoads2() {
        String exchange="FanountExchange";
        String message="message";
        for (int i=0;i<10;i++){
            rabbitTemplate.convertAndSend(exchange,"",message);
        }
    }
 
}

此时可以看到两个队列各自有十条消息

3、接受消息

 //监听交换机Fanount_Qeueue1
    @RabbitListener(queues = "Fanount_Qeueue1")
    public void listenFanountQeueue1(String msg){
        System.out.println("Fanount_Qeueue1接收到的消息："+msg);
    }
    //监听交换机Fanount_Qeueue2
    @RabbitListener(queues = "Fanount_Qeueue2")
    public void listenFanountQeueue2(String msg){
        System.out.println("Fanount_Qeueue2接收到的消息："+msg);
    }

控制台结果如下（共发送20条，每个队列10条）

Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue2接收到的消息：message

六、发布订阅模式之路由模式（Direct）

会将消息根据规则路由到指定的队列

1、声明（基于@RabbitListener声明）

package com.rabbitmqdemoconsumer.rabbitmq;
 
import org.springframework.amqp.core.ExchangeTypes;
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 SpringRabbitLeistener {
 
    /**
     * 绑定交换机和队列，并为key赋值
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "DirectQueue1"),
        exchange = @Exchange(name = "DirectExchange",type = ExchangeTypes.DIRECT),
        key = {"red","blue"}
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("listenDirectQueue1接收到的消息："+msg);
    }
 
 
    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "DirectQueue2"),
        exchange = @Exchange(name = "DirectExchange",type = ExchangeTypes.DIRECT),
        key = {"red","yellow"}
    ))
    public void listenDirectQueue2(String msg){
        System.out.println("listenDirectQueue2接收到的消息："+msg);
    }
}

此时可以看到声明的队列

声明的交换机（第一个）

绑定关系

2、发送给blue

发送消息

 
@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @Test
    void contextLoads2() {
        String exchange="DirectExchange";
        String message="HelloWorld";
        for (int i=0;i<10;i++){
            rabbitTemplate.convertAndSend(exchange,"blue",message);
        }
    }
 
}

接收消息

listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld

3、发送给red

发送消息

@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @Test
    void contextLoads2() {
        String exchange="DirectExchange";
        String message="HelloWorld";
        for (int i=0;i<10;i++){
            rabbitTemplate.convertAndSend(exchange,"blue",message);
        }
    }
 
}

接收消息

listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld

七、发布订阅模式之广播模式（Topic）

Queue与Exchange指定BindingKey可以使用通配符：

  • #：代指0个或多个单词
  • *：代指一个单词

比如：

  • bindingkey: china.# ->中国的所有消息
  • bindingkey: #.weather ->所以国家的天气

1、声明

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "TopicQueue1"),
        exchange = @Exchange(name = "TopicExchange",type = ExchangeTypes.TOPIC),
        key = {"china.#"}
    ))
public void listenTopicQueue1(String msg){
    System.out.println("listenTopicQueue1接收到的消息："+msg);
}
 
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "TopicQueue2"),
    exchange = @Exchange(name = "TopicExchange",type = ExchangeTypes.TOPIC),
    key = {"#.news"}
))
public void listenTopicQueue2(String msg){
    System.out.println("listenTopicQueue2接收到的消息："+msg);
}

队列

交换机（第四个）

绑定关系

2、发送消息（测试1）

package com.rabbitmqdemo;
 
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
 
@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @Test
    void contextLoads2() {
        String exchange="TopicExchange";
        String message="HelloWorld";
        for (int i=0;i<10;i++){
            rabbitTemplate.convertAndSend(exchange,"china.news",message);
        }
    }
 
}

接收消息

TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld

3、发送消息（测试2）

发送消息

package com.rabbitmqdemo;
 
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
 
@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @Test
    void contextLoads2() {
        String exchange="TopicExchange";
        String message="HelloWorld";
        for (int i=0;i<10;i++){
            rabbitTemplate.convertAndSend(exchange,"china.weather",message);
        }
    }
 
}

接收消息

TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

相关文章

  • JAVA生产者消费者(线程同步)代码学习示例

    JAVA生产者消费者(线程同步)代码学习示例

    这篇文章主要介绍了JAVA线程同步的代码学习示例，大家参考使用吧
    2013-11-11
  • Kafka利用Java实现数据的生产和消费实例教程

    Kafka利用Java实现数据的生产和消费实例教程

    这篇文章主要给大家介绍了关于Kafka利用Java实现数据的生产和消费的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧。
    2018-01-01
  • Mybatis + js 实现下拉列表二级联动效果

    Mybatis + js 实现下拉列表二级联动效果

    这篇文章给大家介绍基于Mybatis + js 实现下拉列表二级联动效果，实现代码分为前端界面实现和后端处理方法，代码简单易懂，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友参考下吧
    2021-06-06
  • 关于idea中ssm框架的编码问题分析

    关于idea中ssm框架的编码问题分析

    在实际开发中需要将操作系统编码、文件编码、页面编码以及tomcat服务器编码保持一致，而tomcat在默认情况下是使用UTF-8，这就使得其打印的日志文件出现中文乱码，因此在一般情况下，只需要将tomcat服务器的编码改为GBK即可
    2021-06-06
  • SpringSecurity整合jwt权限认证的全流程讲解

    SpringSecurity整合jwt权限认证的全流程讲解

    这篇文章主要介绍了SpringSecurity整合jwt权限认证的全流程讲解，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2021-06-06
  • Java中获取Class对象的3种方式代码示例

    Java中获取Class对象的3种方式代码示例

    Class对象是反射的核心,通过他可以调用类的任意方法,下面这篇文章主要给大家介绍了关于Java中获取Class对象的3种方式,文中通过代码介绍的非常详细,需要的朋友可以参考下
    2024-07-07
  • 详解Java中Duration类的使用方法

    详解Java中Duration类的使用方法

    Duration类通过秒和纳秒相结合来描述一个时间量，最高精度是纳秒。本文将通过示例详细为大家讲讲Duration类的使用，需要的可以参考一下
    2022-05-05
  • Java如何利用Mybatis进行数据权限控制详解

    Java如何利用Mybatis进行数据权限控制详解

    这篇文章主要介绍了Java如何利用Mybatis进行数据权限控制详解,数据权限控制最终的效果是会要求在同一个数据请求方法中，根据不同的权限返回不同的数据集，而且无需并且不能由研发编码控制。,需要的朋友可以参考下
    2019-06-06
  • mybatis查询实现返回List<Map>类型数据操作

    mybatis查询实现返回List<Map>类型数据操作

    这篇文章主要介绍了mybatis查询实现返回List<Map>类型数据操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
    2020-11-11
  • SpringBoot导出Excel的四种实现方式

    SpringBoot导出Excel的四种实现方式

    近期接到了一个小需求,要将系统中的数据导出为Excel,且能将Excel数据导入到系统,对于大多数研发人员来说,这算是一个最基本的操作了,本文就给大家总结一下SpringBoot导出Excel的四种实现方式,需要的朋友可以参考下
    2024-01-01

最新评论