SpringBoot使用RabbitMQ消息队列

简介: springboot使用rabbitmq消息队列

消息队列中间件是分布式系统中重要的组件,主要解决应用耦合,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构使用较多的消息队列有ActiveMQ,RabbitMQ,Kafka,RocketMQ等,这里主要讲解RabbitMQ的简单使用

一、创建SpringBoot项目,并引入依赖

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

二、Queue方式

1、定义配置类

/**
 * @author Gjing
 **/
@Configuration
public class RabbitMqConfiguration {

    /**
     * 声明一个名为simple的队列
     */
    @Bean
    public Queue testQueue() {
        return new Queue("simple");
    }
}    

2、声明一个生产者

/**
 * @author Gjing
 **/
@Component
public class Producer {

    @Resource
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String message = "hello";
        this.rabbitTemplate.convertAndSend("simple", message);
    }
}

3、声明消费者

/**
 * @author Gjing
 **/
@Slf4j
@Component
public class Consumer {

    @RabbitListener(queues = "simple")
    public void receive(String message) {
        log.info("消费者1收到消息:{}", message);
    }
}

4、创建Controller进行调用

/**
 * @author Gjing
 **/
@RestController
public class DemoController {

    @Resource
    private Producer producer;

    @PostMapping("/message")
    public void send() {
        for (int i = 0; i < 10; i++) {
            this.producer.send();
        }
    }
}

5、执行结果

1560475877_1_

三、topic exchange方式

1、定义配置类

/**
 * @author Gjing
 **/
@Configuration
public class RabbitMqConfiguration {

    /**
     * 声明一个名为topic.message1的队列
     */
    @Bean
    public Queue topicQueue() {
        return new Queue("topic.message1");
    }

    /**
     * 声明一个名为topic.message2的队列
     */
    @Bean
    public Queue topicQueue2() {
        return new Queue("topic.message2");
    }

    /**
     * 声明一个名为exchange的交换机
     */
    @Bean
    public TopicExchange exchange() {
        return new TopicExchange("exchange");
    }

    /**
     * 将topic.message1的队列绑定到exchange交换机
     */
    @Bean
    public Binding bindMessage1() {
        return BindingBuilder.bind(topicQueue()).to(exchange()).with("topic.message1");
    }

    /**
     * 将topic.message2的队列绑定到exchange交换机
     */
    @Bean
    public Binding bindMessage2() {
        return BindingBuilder.bind(topicQueue2()).to(exchange()).with("topic.message2");
    }
}

2、定义生产者

/**
 * @author Gjing
 **/
@Component
public class TopicProducer {

    @Resource
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String message1 = "I am topic.message1";
        String message2 = "I am topic.message2";
        this.rabbitTemplate.convertAndSend("exchange", "topic.message1", message1);
        this.rabbitTemplate.convertAndSend("exchange", "topic.message2", message2);
    }
}

3、定义消费者1

/**
 * @author Gjing
 **/
@Component
@Slf4j
public class TopicConsumer1 {

    @RabbitListener(queues = "topic.message1")
    public void receive(String message) {
        log.info("消费者1收到消息:{}", message);
    }
}

4、定义消费者2

/**
 * @author Gjing
 **/
@Component
@Slf4j
public class TopicConusmer2 {
    @RabbitListener(queues = "topic.message2")
    public void receive(String message) {
        log.info("消费者2收到消息:{}", message);
    }
}

5、创建controller进行调用

/**
 * @author Gjing
 **/
@RestController
public class TopicController {
    @Resource
    private TopicProducer topicProducer;

    @PostMapping("/message-topic")
    public void sendMessageTopic() {
        for (int i = 0; i < 10; i++) {
            this.topicProducer.send();
        }
    }
}

6、执行结果

1560475964_1_

四、fanout方式

1、定义配置类

/**
 * @author Gjing
 **/
@Configuration
public class RabbitMqConfiguration {
    
    /**
     * 声明一个名为fanout.1的队列
     */
    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanout.1");
    }
    /**
     * 声明一个名为fanout.2的队列
     */
    @Bean
    public Queue fanoutQueue2() {
        return new Queue("fanout.2");
    }
    /**
     * 声明一个名为fanout.3的队列
     */
    @Bean
    public Queue fanoutQueue3() {
        return new Queue("fanout.3");
    }

    /**
     * 声明一个名为fanoutExchange的转发器
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    /**
     * 将队列fanoutQueue1绑定到fanout转发器
     */
    @Bean
    public Binding bindFanout1() {
        return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
    }

    /**
     * 将队列fanoutQueue1绑定到fanout转发器
     */
    @Bean
    public Binding bindFanout2() {
        return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
    }

    /**
     * 将队列fanoutQueue1绑定到fanout转发器
     */
    @Bean
    public Binding bindFanout3() {
        return BindingBuilder.bind(fanoutQueue3()).to(fanoutExchange());
    }
}

2、定义生产者

/**
 * @author Gjing
 **/
@Component
public class FanoutProducer {
    @Resource
    private AmqpTemplate amqpTemplate;

    public void send() {
        String message = "hello, I am speaker";
        //这里的routingKey会被rabbitMQ忽略,如果不设置这个参数会导致发送消息失败,
        //所以这里随便写(我给他空字符串),rabbitMQ会默认发给所有绑定的
        this.amqpTemplate.convertAndSend("fanoutExchange","", message);
    }
}

3、定义消费者1

/**
 * @author Gjing
 **/
@Component
@Slf4j
public class FanoutConsumer1 {

    @RabbitListener(queues = "fanout.1")
    public void receive(String message) {
        log.info("消费者1收到消息:{}", message);
    }
}

4、定义消费者2

/**
 * @author Gjing
 **/
@Component
@Slf4j
public class FanoutConsumer2 {
    @RabbitListener(queues = "fanout.2")
    public void receive(String message) {
        log.info("消费者2收到消息:{}", message);
    }
}

5、定义消费者3

/**
 * @author Gjing
 **/
@Component
@Slf4j
public class FanoutConsumer3 {
    @RabbitListener(queues = "fanout.3")
    public void receive(String message) {
        log.info("消费者3收到消息:{}", message);
    }
}

6、创建controller调用

/**
 * @author Gjing
 **/
@RestController
public class FanoutController {
    @Resource
    private FanoutProducer fanoutProducer;

    @PostMapping("/message-fanout")
    public void sendFanout() {
        this.fanoutProducer.send();
    }
}

7、执行结果

1560476034_1_


五、RabbitMQ核心概念

server:又称Broker,接受客户端的连接实现AMQP实体服务;connection:与broker的连接;channel:网络通道,几乎所有的操作都是在channel中进行;message:服务器和应用程序之间传送的数据,由properties和body组成,properties可以对消息进行修饰,比如消息的优先级和高级特性, body为消息的内容;exchange:交换机,接收消息,根据路邮件转发消息到绑定的队列;binding:exchange和queue之间的虚拟连接,可以包含routing key;Routing key:一个路由规则,虚拟机用他确定如何路由一个特定信息;Queue:也称为message Queue,消息队列,保存信息并将它们转发给消费者。

Exchange类型:

  • Fanout:路由规则是把所有发送到该Exchange的消息路由到所有与她绑定的Queue中
    1

备注:生产者P生产消息1推送到Exchange,由于Exchange Type=fanout这时候会遵循fanout的规则将消息推送到所有与他绑定的Queue。

  • direct:把消息路由到那些binding key与routing key完全匹配的Queue中。
    2

备注:生产者P发送消息时Routing key = bloking时,这时候将消息传送到Exchange,Exchange获取到生产者发送过来的消息后,会根据自身的规则进行与匹配响应的Queue,这时候发现Queue1和Queue2都符合,就会将消息传送给这两个队列,如果我们以Routing key = create和routing key = confirm发送消息时,这时候消息只会被推送到Queue2队列中,其他的Routing key 的消息会被丢弃。

  • topic:模糊匹配,通过通配符满足一部分规则就可以传送,其中注意的是有两个字符 ‘星号’ 和#号,其中 星号 用于匹配一个单词,#号用于匹配多个单词(可以是0个)
    2

备注:当生产者发送消息Routing Key=F.C.E的时候,这时候只满足Queue1,所以会被路由到Queue中,如果Routing Key=A.C.E这时候会被同是路由到Queue1和Queue2中,如果Routing Key=A.F.B时,这里只会发送一条消息到Queue2中。

常见面试题

  • 什么是元数据?元数据分为哪些类型?包括哪些内容?与cluster相关的元数据有哪些?元数据是如何保存的?元数据在cluster中是如何分布的?

在非cluster模式中,元数据主要分为Queue元数据(Queue名字和属性等)、Exchange元数据(Exchange名字、类型、属性等)、binding元数据(存放路由关系的查找表)、vhost元数据(vhost范围内针对前三者的名字空间约束和安全属性设置)。在cluster模式下,包括cluster中node位置信息和node关系信息。元数据按照erlang node的类型确定是仅保存于RAM中,还是同时保存在RAM或者Disk上,元数据在cluster中是全node分布的

  • rabbitmq的一个Queue中存放的message是否有数量限制?

可以认为无限制,限制取决于机器的内存,但是消息过多会导致处理效率的下降。

  • rabbitmq如何实现延迟队列?

没有直接支持延迟队列功能,但是可以通过两个特性来实现延迟队列,①TTL:通过队列属性设置,队列中的所有消息都有相同的过期时间、对消息进行单独设置,每条消息TTL可以不同。如果同时使用,则消息的过期时间以两者之间TTL较小的那个数值为准,消息在队列的生存时间一旦超过设置的TTL值,就称为dead letter。②DLX:Queue可以配置X-dead-letter-exchange和x-dead-letter-routing-key(可选)两个参数,如果队列内出现了dead letter,则按照这两个参数重新路由转发到指定的队列。

  • X-dead-letter-exchange:出现dead letter之后将dead letter重新发送到指定exchange
  • 出现dead letter之后将dead letter重新按照指定的routing-key发送

    • 出现dead letter的情况有:
    1. 消息或者队列的TTL过期; 2. 队列达到最大长度; 3. 消息被消费者拒绝

前往第二章:SpringBoot使用RabbitMQ(二)

以上为个人见解,如有误欢迎各位指正

相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
8天前
|
消息中间件 Java 双11
RocketMQ:揭秘电商巨头背后的消息队列秘密
**RocketMQ概览:**高性能分布式消息队列,适用于有序消息、事务处理、流计算、消息推送、日志处理及Binlog分发。在双11等高流量场景下证明了其性能、稳定性和低延迟。Java开发,利于扩展,性能超RabbitMQ,支持死信队列,但可能有集成兼容性问题。适合Java开发者,为电商等场景优化,每秒处理大量消息。
28 3
RocketMQ:揭秘电商巨头背后的消息队列秘密
|
3天前
|
消息中间件 Java 程序员
Spring Boot整合RabbitMQ
Spring Boot整合RabbitMQ
|
6天前
|
消息中间件 Java Kafka
SpringBoot实用开发篇第六章(整合第三方技术,ActiveMQ,RabbitMQ,RocketMQ,Kafka)
SpringBoot实用开发篇第六章(整合第三方技术,ActiveMQ,RabbitMQ,RocketMQ,Kafka)
|
5天前
|
消息中间件 监控 Java
使用Spring Boot结合ActiveMQ和MQTT实现消息的发送和接收
使用Spring Boot结合ActiveMQ和MQTT实现消息的发送和接收
28 3
|
14天前
|
消息中间件 Java 测试技术
消息队列 MQ操作报错合集之设置了setKeepAliveInterval(1)但仍然出现客户端未连接,该怎么解决
在使用消息队列MQ时,可能会遇到各种报错情况。以下是一些常见的错误场景、可能的原因以及解决建议的汇总:1.连接错误、2.消息发送失败、3.消息消费报错、4.消息重试与死信处理、5.资源与权限问题、6.配置错误、7.系统资源限制、8.版本兼容性问题。
|
14天前
|
消息中间件 设计模式 网络安全
消息队列 MQ操作报错合集之broker启用controller配置时,遇到报错,是什么导致的
在使用消息队列MQ时,可能会遇到各种报错情况。以下是一些常见的错误场景、可能的原因以及解决建议的汇总:1.连接错误、2.消息发送失败、3.消息消费报错、4.消息重试与死信处理、5.资源与权限问题、6.配置错误、7.系统资源限制、8.版本兼容性问题。
|
14天前
|
消息中间件 Apache RocketMQ
消息队列 MQ操作报错合集之设置了controller后,有一主一从,但只显示一个,该怎么解决
在使用消息队列MQ时,可能会遇到各种报错情况。以下是一些常见的错误场景、可能的原因以及解决建议的汇总:1.连接错误、2.消息发送失败、3.消息消费报错、4.消息重试与死信处理、5.资源与权限问题、6.配置错误、7.系统资源限制、8.版本兼容性问题。
|
6天前
|
消息中间件
RabbitMQ是一个功能强大的开源消息代理软件,用于处理消息队列
RabbitMQ是一个功能强大的开源消息代理软件,用于处理消息队列
12 0
|
9天前
|
消息中间件 自然语言处理 负载均衡
RabbitMQ揭秘:轻量级消息队列的优缺点全解析
**RabbitMQ简介** RabbitMQ是源自电信行业的消息中间件,支持AMQP协议,提供轻量、快速且易于部署的解决方案。它拥有灵活的路由配置,广泛的语言支持,适用于异步处理、负载均衡、日志收集和微服务通信等场景。然而,当面临大量消息堆积或高吞吐量需求时,性能可能会下降,并且扩展和开发成本相对较高。
24 0
|
14天前
|
消息中间件 测试技术 开发工具
消息队列 MQ操作报错合集之收到"WARN RocketmqClient - consumeMessage Orderly return"警告,是什么原因
在使用消息队列MQ时,可能会遇到各种报错情况。以下是一些常见的错误场景、可能的原因以及解决建议的汇总:1.连接错误、2.消息发送失败、3.消息消费报错、4.消息重试与死信处理、5.资源与权限问题、6.配置错误、7.系统资源限制、8.版本兼容性问题。

热门文章

最新文章

相关产品

  • 云消息队列 MQ