AI 助理
备案控制台
开发者社区 微服务 文章 正文

SpringBoot 整合 AMQP（RabbitMQ）

2019-08-13 9166
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： SpringBoot整合AMQP（RabbitMQ） 添加pom依赖 org.springframework.boot spring-boot-starter-amqp application.

SpringBoot整合AMQP（RabbitMQ）

  1. 添加pom依赖

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

  2. application.properties配置

    spring.rabbitmq.host=***.***.***.***
spring.rabbitmq.port=5762
spring.rabbitmq.username=admin
spring.rabbitmq.password=***

  3. 在RabbitMQ中所有的消息生产者提交的消息都会交由Exchange进行再分配，Exchange会根据不同的策略将消息分发到不同的Queue中。RabbitMQ提供了4种不同策略，分别是Direct、Fanout、Topic、Header，4种策略中前三种使用率较高

    • Direct

      DirectExchange的路由策略是将消息队列绑定到一个DirectExchange上，但一条消息到达DirectExchange时会被转发到与该条消息routing key相同的Queue上

      • DirectExchange的配置如下：

        /**
 * @author wsyjlly
 * @create 2019.07.17 - 21:33
 **/
@Configuration
public class RabbitDirectConfig {
    public final static String DIRECTNAME = "ysw-direct";
    @Bean
    Queue queue1(){
        return new Queue("queue-direct1");
    }

    @Bean

    Queue queue2(){
        return new Queue("queue-direct2");
    }

    @Bean
    Queue queue3(){
        return new Queue("queue-direct3");
    }

    @Bean
    DirectExchange directExchange(){
        return new DirectExchange(DIRECTNAME,true,false);
    }

    @Bean
    Binding binding1(){
        return BindingBuilder.bind(queue1()).to(directExchange()).with("direct1");
    }

    @Bean
    Binding binding2(){
        return BindingBuilder.bind(queue2()).to(directExchange()).with("direct2");
    }
}

        DirectExchange和Binding两个Bean的配置可以省略掉，即如果使用DirectExchange，之配置一个Queue的实例即可

      • 配置消费者

        /**
 * @author wsyjlly
 * @create 2019.07.17 - 21:42
 **/
@Component
public class DirectReceiver {
    Logger logger= LoggerFactory.getLogger(getClass());

    @RabbitListener(queues = "queue-direct1")
    public void directHandler1(String msg){
        logger.info("\033[30;4m"+"queue-direct1:"+msg+"\033[0m");
    }

    @RabbitListener(queues = "queue-direct2")
    public void directHandler2(String msg){
        logger.info("\033[30;4m"+"queue-direct2:"+msg+"\033[0m");
    }

    @RabbitListener(queues = "queue-direct3")
    public void directHandler3(String msg){
        logger.info("\033[30;4m"+"queue-direct3:"+msg+"\033[0m");
    }
}

        通过@RabbitListener注解指定一个方法是一个消费者方法，方法参数就是所接收的消息。

      • 消息发送

        通过注入RabbitTemplate对象来进行消息发送，在这里我通过定时任务使其自定发送，须开启定时任务，详细操作可查看一节

        /**
 * @author wsyjlly
 * @create 2019.07.18 - 1:13
 **/
@Component
public class RabbitmqSchedule {
    @Autowired
    RabbitTemplate rabbitTemplate;
    Logger logger = LoggerFactory.getLogger(getClass());

    @Scheduled(fixedDelay = 5000,initialDelay = 3000)
    public void direct(){
        String message = "direct-task";
        logger.info("\033[30;4m"+message+"\033[0m");
        rabbitTemplate.convertAndSend("ysw-direct","direct1",message);
        rabbitTemplate.convertAndSend("ysw-direct","direct2",message);
        rabbitTemplate.convertAndSend("queue-direct3",message);
    }
}

    • Fanout

      FanoutExchange的数据交换策略是把所有到达FanoutExchange的消息转发给所有与他绑定的Queue，在这种策略中，routingkey将不起作用。

      • FanoutExchange的配置方式如下：

        /**
 * @author wsyjlly
 * @create 2019.07.17 - 21:33
 **/
@Configuration
public class RabbitFanoutConfig {
    public final static String FANOUTNAME = "ysw-fanout";

    @Bean
    Queue queue4(){
        return new Queue("queue-fanout1");
    }

    @Bean
    Queue queue5(){
        return new Queue("queue-fanout2");
    }

    @Bean
    Queue queue6(){
        return new Queue("queue-fanout3");
    }

    @Bean
    FanoutExchange fanoutExchange(){
        return new FanoutExchange(FANOUTNAME,true,false);
    }

    @Bean
    Binding binding4(){
        return BindingBuilder.bind(queue4()).to(fanoutExchange());
    }

    @Bean
    Binding binding5(){
        return BindingBuilder.bind(queue5()).to(fanoutExchange());
    }

    @Bean
    Binding binding6(){
        return BindingBuilder.bind(queue6()).to(fanoutExchange());
    }
}

      • 配置消费者

        /**
 * @author wsyjlly
 * @create 2019.07.17 - 21:42
 **/
@Component
public class FanoutReceiver {
    Logger logger= LoggerFactory.getLogger(getClass());

    @RabbitListener(queues = "queue-fanout1")
    public void fanoutHandler1(String msg){
        logger.info("\033[31;4m"+"queue-fanout1:"+msg+"\033[0m");
    }

    @RabbitListener(queues = "queue-fanout2")
    public void fanoutHandler2(String msg){
        logger.info("\033[31;4m"+"queue-fanout2:"+msg+"\033[0m");
    }

    @RabbitListener(queues = "queue-fanout3")
    public void fanoutHandler3(String msg){
        logger.info("\033[31;4m"+"queue-fanout3:"+msg+"\033[0m");
    }
}

      • 消息发送

        /**
 * @author wsyjlly
 * @create 2019.07.18 - 1:13
 **/
@Component
public class RabbitmqSchedule {
    @Autowired
    RabbitTemplate rabbitTemplate;
    Logger logger = LoggerFactory.getLogger(getClass());

    Scheduled(fixedDelay = 5000,initialDelay = 4000)
    public void fanout(){
        String message = "fanout-task";
        logger.info("\033[31;4m"+message+"\033[0m");
        rabbitTemplate.convertAndSend("ysw-fanout",null,message);
    }
}

    • Topic

      TopicExchange是比较复杂也比较灵活的一种路由策略，在TopicExchange中，Queue通过routingkey绑定到TopicExchange上，当消息发送到TopicExchange后，TopicExchange根据消息的routingkey将消息路由到一个或多个Queue上。

      • TopicExchange配置如下：

        /**
 * @author wsyjlly
 * @create 2019.07.17 - 21:33
 **/
@Configuration
public class RabbitTopicConfig {
    public final static String TOPIC_NAME = "ysw-topic";
    @Bean
    Queue queue7(){
        return new Queue("queue-topic1");
    }

    @Bean
    Queue queue8(){
        return new Queue("queue-topic2");
    }

    @Bean
    Queue queue9(){
        return new Queue("queue-topic3");
    }

    @Bean
    TopicExchange topicExchange(){
        return new TopicExchange(TOPIC_NAME,true,false);
    }

    @Bean
    Binding binding7(){
         /*
         * 匹配规则
         * 绑定键binding key也必须是这种形式。以特定路由键发送的消息将会发送到所有绑定键与之匹配的队列中。但绑定键有两种特殊的情况：
         * 绑定键binding key也必须是这种形式。以特定路由键发送的消息将会发送到所有绑定键与之匹配的队列中。但绑定键有两种特殊的情况：
         * ①*（星号）仅代表一个单词
         * ②#（井号）代表任意个单词
         **/
        return BindingBuilder.bind(queue7()).to(topicExchange()).with("#.topic1");
    }

    @Bean
    Binding binding8(){
        return BindingBuilder.bind(queue8()).to(topicExchange()).with("topic2.#");
    }

    @Bean
    Binding binding9(){
        return BindingBuilder.bind(queue9()).to(topicExchange()).with("#.topic3.*");
    }
}

      • 配置消费者

        /**
 * @author wsyjlly
 * @create 2019.07.17 - 21:42
 **/
@Component
public class TopicReceiver {
    Logger logger= LoggerFactory.getLogger(getClass());

    @RabbitListener(queues = "queue-topic1")
    public void topicHandler1(String msg){
        logger.info("\033[32;4m"+"queue-topic1:"+msg+"\033[0m");
    }

    @RabbitListener(queues = "queue-topic2")
    public void topicHandler2(String msg){
        logger.info("\033[32;4m"+"queue-topic2:"+msg+"\033[0m");
    }

    @RabbitListener(queues = "queue-topic3")
    public void topicHandler3(String msg){
        logger.info("\033[32;4m"+"queue-topic3:"+msg+"\033[0m");
    }
}

      • 消息发送

        /**
 * @author wsyjlly
 * @create 2019.07.18 - 1:13
 **/
@Component
public class RabbitmqSchedule {
    @Autowired
    RabbitTemplate rabbitTemplate;
    Logger logger = LoggerFactory.getLogger(getClass());

    @Scheduled(cron = "0-30/6 * * * * ?")
    public void topic(){
        String message = "topic-task";
        int i = 0;
        logger.info("\033[32;4m"+message+"\033[0m");
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic1.news",message + 1);//topic1
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic1.salary",message + 2);//topic1
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic2.news",message + 3);//topic2
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic2.item",message + 4);//topic2
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic2.sth.topic1",message + 5);//topic2&topic1
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic1.sth.topic2",message + 6);//topic2&topic1
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic3",message + 7);//topic3
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic3.news",message + 8);//topic3
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic1.topic3",message + 9); //topic1&topic3
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic2.topic3",message + 10);//topic2&topic3
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic3.topic1",message + 11);//topic3&topic1
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPIC_NAME,
        "topic2.topic3.topic1",message + 12);//topic1&topic2&topic3
    }
}

    • Header

      HeaderExchange是一种较少使用的路由策略，HeaderExchange会根据消息的Header将消息路由到不同的Queue上，这种策略也和routingkey无关。

      • HeaderExchange配置如下：

        /**
 * @author wsyjlly
 * @create 2019.07.17 - 21:33
 **/
@Configuration
public class RabbitHeaderConfig {
    public final static String HEADER_NAME = "ysw-header";
    @Bean
    Queue queue10(){
        return new Queue("queue-header1");
    }

    @Bean

    Queue queue11(){
        return new Queue("queue-header2");
    }

    @Bean
    Queue queue12(){
        return new Queue("queue-header3");
    }

    @Bean
    HeadersExchange headersExchange(){
        return new HeadersExchange(HEADER_NAME,true,false);
    }

    @Bean
    Binding binding10(){
        Map<String,Object> map = new HashMap<>();
        map.put("age", "18");
        map.put("name", "ysw");
        return BindingBuilder.bind(queue10()).to(headersExchange()).whereAny(map).match();
    }

    @Bean
    Binding binding11(){
        Map<String,Object> map = new HashMap<>();
        map.put("name", "ysw");
        return BindingBuilder.bind(queue11()).to(headersExchange()).where("age").exists();
    }

    @Bean
    Binding binding12(){
        Map<String,Object> map = new HashMap<>();
        map.put("age", "18");
        map.put("name", "ysw");
        return BindingBuilder.bind(queue12()).to(headersExchange()).whereAll(map).match();
    }
}

        Binding配置注释：whereAny表示消息的Header中只要有一个Header匹配上map中的key/value，就把该消息路由到名为“queue-header1”的Queue上；whereAll方法表示消息的所有Header都要匹配，才将消息路由到名为“queue-header2”的Queue上；where表示只要消息的header中包含age，无论age值为多少，都将消息路由到名为“queue-header2”的Queue上。

      • 配置消费者

        /**
 * @author wsyjlly
 * @create 2019.07.17 - 21:42
 **/
@Component
public class HeaderReceiver {
    Logger logger= LoggerFactory.getLogger(getClass());

    @RabbitListener(queues = "queue-header1")
    public void headerHandler1(byte[] msg){
        logger.info("\033[33;4m"+"queue-header1:"+new String(msg,0,msg.length)+"\033[0m");
    }

    @RabbitListener(queues = "queue-header2")
    public void headerHandler2(byte[] msg){
        logger.info("\033[33;4m"+"queue-header2:"+new String(msg,0,msg.length)+"\033[0m");
    }

    @RabbitListener(queues = "queue-header3")
    public void headerHandler3(byte[] msg){
        logger.info("\033[33;4m"+"queue-header3:"+new String(msg,0,msg.length)+"\033[0m");
    }
}

      • 消息发送

        /**
 * @author wsyjlly
 * @create 2019.07.18 - 1:13
 **/
@Component
public class RabbitmqSchedule {
    @Autowired
    RabbitTemplate rabbitTemplate;
    Logger logger = LoggerFactory.getLogger(getClass());

    @Scheduled(cron = "0-30/3 * * * * ?")
    public void header(){
        String message = "header-task";
        logger.info("\033[33;4m"+message+"\033[0m");

        Message message1 = MessageBuilder.withBody("name=name".getBytes())
        .setHeader("name", "aaa").build();
        Message message2 = MessageBuilder.withBody("name=ysw".getBytes())
        .setHeader("name", "ysw").build();
        Message message3 = MessageBuilder.withBody("age=19".getBytes())
        .setHeader("age", "19").build();
        Message message4 = MessageBuilder.withBody("age=18".getBytes())
        .setHeader("age", "18").build();
        Message message5 = MessageBuilder.withBody("name=ysw&age=18".getBytes())
        .setHeader("name", "ysw").setHeader("age","18").build();
        Message message6 = MessageBuilder.withBody("name=ysw&age=19".getBytes())
        .setHeader("name", "ysw").setHeader("age","19").build();
        Message message7 = MessageBuilder.withBody("name=aaa&age=18".getBytes())
        .setHeader("name", "aaa").setHeader("age","18").build();

        rabbitTemplate.convertAndSend(RabbitHeaderConfig.HEADER_NAME,
        null,message1);
        rabbitTemplate.convertAndSend(RabbitHeaderConfig.HEADER_NAME,
        null,message2);
        rabbitTemplate.convertAndSend(RabbitHeaderConfig.HEADER_NAME,
        null,message3);
        rabbitTemplate.convertAndSend(RabbitHeaderConfig.HEADER_NAME,
        null,message4);
        rabbitTemplate.convertAndSend(RabbitHeaderConfig.HEADER_NAME,
        null,message5);
        rabbitTemplate.convertAndSend(RabbitHeaderConfig.HEADER_NAME,
        null,message6);
        rabbitTemplate.convertAndSend(RabbitHeaderConfig.HEADER_NAME,
        null,message7);
    }
}
文章标签：
云消息队列 MQ
消息中间件
Java
关键词：
云消息队列 MQ springboot
amqp云消息队列 MQ
云消息队列 MQ amqp
springboot云消息队列 MQ
Spring boot rabbitmq
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
swyan
目录
相关文章
如夜了我衣衫太薄便归家靠路灯
|
3月前
|
消息中间件 Java 网络架构
springboot 整合rabbitmq，实现发送提醒消息
springboot 整合rabbitmq，实现发送提醒消息
如夜了我衣衫太薄便归家靠路灯
225 4
热爱技术的小郑
|
3月前
|
消息中间件 Java 测试技术
消息中间件RabbitMQ---SpringBoot整合RabbitMQ【三】
这篇文章是关于如何在SpringBoot应用中整合RabbitMQ的消息中间件。内容包括了在SpringBoot项目中添加RabbitMQ的依赖、配置文件设置、启动类注解，以及如何通过单元测试来创建交换器、队列、绑定，并发送和接收消息。文章还介绍了如何配置消息转换器以支持对象的序列化和反序列化，以及如何使用注解`@RabbitListener`来接收消息。
热爱技术的小郑
49 2
消息中间件RabbitMQ---SpringBoot整合RabbitMQ【三】
VipSoft
|
3月前
|
网络协议 Java 物联网
MQTT(EMQX) - SpringBoot 整合MQTT 连接池 Demo - 附源代码 + 在线客服聊天架构图
MQTT(EMQX) - SpringBoot 整合MQTT 连接池 Demo - 附源代码 + 在线客服聊天架构图
VipSoft
766 2
如夜了我衣衫太薄便归家靠路灯
|
3月前
|
消息中间件 Java Maven
springboot 整合rocketMQ，实现发送提醒消息
springboot 整合rocketMQ，实现发送提醒消息
如夜了我衣衫太薄便归家靠路灯
99 4
码农飞哥
|
4月前
|
消息中间件 Java 测试技术
【RocketMQ系列八】SpringBoot集成RocketMQ-实现普通消息和事务消息
【RocketMQ系列八】SpringBoot集成RocketMQ-实现普通消息和事务消息
码农飞哥
317 1
三分钟热度的鱼
|
4月前
|
消息中间件 物联网 API
消息队列 MQ使用问题之如何在物联网项目中搭配使用 MQTT、AMQP 与 RabbitMQ
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
142 0
weixin_836869520
|
4月前
|
消息中间件 Java 数据安全/隐私保护
Spring Boot与RabbitMQ的集成
Spring Boot与RabbitMQ的集成
weixin_836869520
85 0
聚娃科技开发者团队
|
4月前
|
消息中间件 Java RocketMQ
Spring Boot与RocketMQ的集成
Spring Boot与RocketMQ的集成
聚娃科技开发者团队
583 0
weixin_836869520
|
4月前
|
消息中间件 Java Spring
实现Spring Boot与RabbitMQ消息中间件的无缝集成
实现Spring Boot与RabbitMQ消息中间件的无缝集成
weixin_836869520
56 0
小目标青年
|
消息中间件 网络协议 Java
Springboot 整合RabbitMq ，用心看完这一篇就够了
Springboot 整合RabbitMq ，用心看完这一篇就够了
小目标青年
1872 0
Springboot 整合RabbitMq ，用心看完这一篇就够了

热门文章

最新文章

  • 1
    RabbitMQ：四种ExChange用法
  • 2
    未读消息（小红点），前端与 RabbitMQ实时消息推送实践，贼简单~
  • 3
    SpringBoot整合RabbitMQ实现消息的发送与接收，确认消息，延时消息
  • 4
    安装配置rabbitmq
  • 5
    3分钟实现SpringBoot集成RabbitMQ，实现消息队列服务！
  • 6
    rabbitmq开启web界面教程
  • 7
    RabbitMQ队列中unacked消息持续时间很久
  • 8
    一些你会忘记的RabbitMQ集群和仲裁队列小知识点
  • 9
    rabbitmq逻辑与开发
  • 10
    转载RabbitMQ入门（3）--发布和订阅
  • 1
    RabbitMQ高级
    44
  • 2
    rabbitmq与kafka的区别
    89
  • 3
    用docker方式 安装rabbitmq 并配置MQTT
    290
  • 4
    MQTT 5.0 报文解析 02：PUBLISH 与 PUBACK
    372
  • 5
    如何实现一个简单易用的 RocketMQ SDK
    74
  • 6
    写了10000字：全面学习RocketMQ中间件
    762
  • 7
    Spring Cloud RocketMQ：构建可靠消息驱动的微服务架构
    251
  • 8
    Serverless 应用引擎产品使用之在阿里云函数计算中，使用了RocketMQ的触发器，并且发送和接收消息都没有问题，但是消息轨迹中没有体现出来消费的情况如何解决
    62
  • 9
    Spring Boot+RocketMQ 实现多实例分布式环境下的事件驱动
    159
  • 10
    RabbitMQ组件介绍
    89

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台

    • 相关实验场景

    更多
  • 部署RabbitMQ
  • Spring-cloud-bus-rocketmq入门与实践
  • Spring-cloud-stream-binder-rocketmq入门与实践
  • Rocketmq-spring入门与实践
  • RocketMQ中使用Java客户端发送消息和消费的应用
  • RocketMQ一站式入门使用
    • 下一篇
    阿里云OSS设置跨域访问