SpringBoot之RabbitMQ的使用

一 、RabbitMQ的介绍

RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件,消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.具体过程如下:

从上图可看出,对于消息队列来说,生产者、消息队列、消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息队列,并且当消息队列收到消息之后,接收消息队列传来的消息,并且给予相应的处理。消息队列常用于分布式系统之间互相信息的传递。
对于RabbitMQ来说,除了这三个基本模块以外,还添加了一个模块,即交换机(Exchange)。它使得生产者和消息队列之间产生了隔离,生产者将消息发送给交换机,而交换机则根据调度策略把相应的消息转发给对应的消息队列。
交换机的主要作用是接收相应的消息并且绑定到指定的队列。交换机有四种类型,分别为Direct、topic、headers、Fanout。

Direct是RabbitMQ默认的交换机模式,也是最简单的模式。即创建消息队列的时候,指定一个BindingKey。当发送者发送消息的时候,指定对应的Key。当Key和消息队列的BindingKey一致的时候,消息将会被发送到该消息队列中。
topic转发信息主要是依据通配符,队列和交换机的绑定主要是依据一种模式(通配符+字符串),而当发送消息的时候,只有指定的Key和该模式相匹配的时候,消息才会被发送到该消息队列中。
headers也是根据一个规则进行匹配,在消息队列和交换机绑定的时候会指定一组键值对规则,而发送消息的时候也会指定一组键值对规则,当两组键值对规则相匹配的时候,消息会被发送到匹配的消息队列中。
Fanout是路由广播的形式,将会把消息发给绑定它的全部队列,即便设置了key,也会被忽略。

二 、SpringBoot整合RabbitMQ(Direct模式)

SpringBoot整合RabbitMQ非常简单,首先还是pom.xml引入依赖。

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

在application.properties中配置RabbitMQ相关的信息,并首先启动了RabbitMQ实例,并创建两个queue。

spring.application.name=spirng-boot-rabbitmq
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin

配置Queue(消息队列),由于采用的是Direct模式,需要在配置Queue的时候指定一个键,使其和交换机绑定。

@Configuration
public class RabbitConfig {

    @Bean
    public org.springframework.amqp.core.Queue Queue() {

        return new org.springframework.amqp.core.Queue("hello");

    }
}

接着就可以发送消息啦。在SpringBoot中,我们使用AmqpTemplate去发送消息。代码如下:

@Component
public class HelloSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send(int index) {

        String context = "hello Queue "+index + new Date();
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("hello", context);
    }
}

生产者发送消息之后就需要消费者接收消息。这里定义了两个消息消费者,用来模拟生产者与消费者一对多的关系。

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {

    @RabbitHandler
    public void process(String hello) {
        System.out.println("Receiver1  : " + hello);
    }
}
@Component
@RabbitListener(queues = "hello")
public class HelloReceiver2 {

    @RabbitHandler
    public void process(String hello) {
        System.out.println("Receiver2  : " + hello);
    }
}

在单元测试中模拟发送消息,批量发送10条消息,两个接收者分别接收了5条消息。

    @Autowired
    private HelloSender helloSender;
    @Test
    public void hello() throws Exception {
        for(int i=0;i<10;i++)
        {
            helloSender.send(i);
        }
    }

实际上RabbitMQ还可以支持发送对象,当然由于涉及到序列化和反序列化,该对象要实现Serilizable接口。这里定义了User对象,用来做发送消息内容。

import java.io.Serializable;
public class User implements Serializable{

    private String name;
    private String pwd;

    public String getPwd() {
        return pwd;
    }
    public void setPwd(String pwd) {
        this.pwd = pwd;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public User(String name, String pwd) {
        this.name = name;
        this.pwd = pwd;
    }
    @Override
public String toString() {
        return "User{" +"name=‘" + name + ‘\‘‘ +", pwd=‘" + pwd + ‘\‘‘ +‘}‘;
    }
}

在生产者中发送User对象。

@Component
public class ModelSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void sendModel(User user) {

        System.out.println("Sender object: " + user.toString());
        this.rabbitTemplate.convertAndSend("object", user);

    }
}

在消费者中接收User对象。

@Component
@RabbitListener(queues = "object")
public class ModelRecevicer {

    @RabbitHandler
    public void process(User user) {

        System.out.println("Receiver object : " + user);

    }
}

在单元测试中注入ModelSender 对象,实例化User对象,然后发送。

    @Autowired
    private ModelSender modelSender;
    @Test
    public void model() throws Exception {

        User user=new User("abc","123");
        modelSender.sendModel(user);
    }

三 、SpringBoot整合RabbitMQ(Topic转发模式)

首先需要在RabbitMQ服务端创建交换机topicExchange,并绑定两个queue:topic.message、topic.messages。

新建TopicRabbitConfig,设置对应的queue与binding。

@Configuration
public class TopicRabbitConfig {

    final static String message = "topic.message";
    final static String messages = "topic.messages";

    @Bean
    public Queue queueMessage() {
        return new Queue(TopicRabbitConfig.message);
    }  

    @Bean
    public Queue queueMessages() {
        return new Queue(TopicRabbitConfig.messages);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("topicExchange");
    }

    @Bean
    Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
    }

    @Bean
    Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
    }
}

创建消息生产者,在TopicSender中发送3个消息。

@Component
public class TopicSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String context = "hi, i am message all";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("topicExchange", "topic.1", context);
    }

    public void send1() {
        String context = "hi, i am message 1";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("topicExchange", "topic.message", context);
    }

    public void send2() {
        String context = "hi, i am messages 2";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("topicExchange", "topic.messages", context);
    }
}

生产者发送消息,这里创建了两个接收消息的消费者。

@Component
@RabbitListener(queues = "topic.message")
public class TopicReceiver {

    @RabbitHandler
    public void process(String message) {
        System.out.println("Topic Receiver1  : " + message);
    }
}

@Component
@RabbitListener(queues = "topic.messages")
public class TopicReceiver2 {

    @RabbitHandler
    public void process(String message) {

        System.out.println("Topic Receiver2  : " + message);

    }
}

在单元测试中注入TopicSender,利用topicSender 发送消息。

    @Autowired
    private TopicSender topicSender;
    @Test
    public void topicSender() throws Exception {
        topicSender.send();
        topicSender.send1();
        topicSender.send2();
    }

从上面的输出结果可以看到,Topic Receiver2 匹配到了所有消息,Topic Receiver1只匹配到了1个消息。

四 、SpringBoot整合RabbitMQ(Fanout Exchange形式)

Fanout Exchange形式又叫广播形式,因此我们发送到路由器的消息会使得绑定到该路由器的每一个Queue接收到消息。首先需要在RabbitMQ服务端创建交换机fanoutExchange,并绑定三个queue:fanout.A、fanout.B、fanout.C。

与Topic类似,新建FanoutRabbitConfig,绑定交换机和队列。

@Configuration
public class FanoutRabbitConfig {

    @Bean
    public Queue AMessage() {
        return new Queue("fanout.A");
    }

    @Bean
    public Queue BMessage() {
        return new Queue("fanout.B");
    }

    @Bean
    public Queue CMessage() {
        return new Queue("fanout.C");
    }

    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    @Bean
    Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(AMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(BMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(CMessage).to(fanoutExchange);
    }
}

创建消息生产者,在FanoutSender中发送消息。

@Component
public class FanoutSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {

        String context = "hi, fanout msg ";
        System.out.println("FanoutSender : " + context);
        this.rabbitTemplate.convertAndSend("fanoutExchange","", context);

    }
}

然后创建了3个接收者FanoutReceiverA、FanoutReceiverB、FanoutReceiverC。

@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {

    @RabbitHandler
    public void process(String message) {
        System.out.println("fanout Receiver A  : " + message);
    }
}
@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB {

    @RabbitHandler
    public void process(String message) {
        System.out.println("fanout Receiver B: " + message);
    }
}
@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {

    @RabbitHandler
    public void process(String message) {
        System.out.println("fanout Receiver C: " + message);
    }
}

在单元测试中注入消息发送者,发送消息。

    @Autowired
    private FanoutSender fanoutSender;
    @Test
    public void fanoutSender() throws Exception {
        fanoutSender.send();
    }

从下图可以看到3个队列都接收到了消息。

本章节创建的类比较多,下图为本章节的结构,也可以直接查看demo源码了解。

原文地址:https://www.cnblogs.com/5ishare/p/10163318.html

时间: 2024-11-04 04:38:39

SpringBoot之RabbitMQ的使用的相关文章

springboot学习笔记-6 springboot整合RabbitMQ

一 RabbitMQ的介绍 RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件.这些软件有很多,包括ActiveMQ(apache公司的),RocketMQ(阿里巴巴公司的,现已经转让给apache). 消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.具体过程如下: 从上图可看出,对于消息队列来说,生产者,消息队列,消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息

springboot系列-springboot整合RabbitMQ

一 RabbitMQ的介绍 RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件.这些软件有很多,包括ActiveMQ(apache公司的),RocketMQ(阿里巴巴公司的,现已经转让给apache). 消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.具体过程如下: 从上图可看出,对于消息队列来说,生产者,消息队列,消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息

SpringBoot整合RabbitMQ之发送接收消息实战

实战前言 前几篇文章中,我们介绍了SpringBoot整合RabbitMQ的配置以及实战了Spring的事件驱动模型,这两篇文章对于我们后续实战RabbitMQ其他知识要点将起到奠基的作用的.特别是Spring的事件驱动模型,当我们全篇实战完毕RabbitMQ并大概了解一下RabbitMQ相关组件的源码时,会发现其中的ApplicationEvent.ApplicationListener.ApplicationEventPublisher跟RabbitMQ的Message.Listener.R

SpringBoot整合RabbitMQ之典型应用场景实战一

实战前言RabbitMQ 作为目前应用相当广泛的消息中间件,在企业级应用.微服务应用中充当着重要的角色.特别是在一些典型的应用场景以及业务模块中具有重要的作用,比如业务服务模块解耦.异步通信.高并发限流.超时业务.数据延迟处理等. RabbitMQ 官网拜读首先,让我们先拜读 RabbitMQ 官网的技术开发手册以及相关的 Features,感兴趣的朋友可以耐心的阅读其中的相关介绍,相信会有一定的收获,地址可见:http://www.rabbitmq.com/getstarted.html 在阅

SpringBoot整合RabbitMQ入门~~

SpringBoot整合RabbitMQ 入门2020-01-12 创建生产者类,并且在yml配置文件中配置5要素连接MQ yml配置文件 spring: rabbitmq: host: xx.xx.xx.xx port: 5672 virtual-host: / username: 默认guest password: 默认guest 编写生产者代码 使用@Configuration 表名它是配置类 再类中声明三要素 //交换机名称 @Bean("itemTopicExchange")

Springboot 整合RabbitMq ,用心看完这一篇就够了

Springboot 整合RabbitMq ,用心看完这一篇就够了 https://blog.csdn.net/qq_35387940/article/details/100514134包括有rabbitMq相关的一些简单理论介绍,provider消息推送实例,consumer消息消费实例,Direct.Topic.Fanout的使用,消息回调.手动确认等. (但是关于rabbitMq的安装,就不介绍了)——————————————————————————————————————————————

springboot使用RabbitMQ教程

1.安装rabbitmqdocker安装,拉取镜像docker pull rabbitmq:management创建容器并启动docker run -d --name rabbitmq --publish 5671:5671 --publish 5672:5672 --publish 4369:4369 --publish 25672:25672 --publish 15671:15671 --publish 15672:15672 rabbitmq:management管理地址:http://

SpringBoot整合RabbitMQ之整合配置篇

实战背景:RabbitMQ实战第一阶段-RabbitMQ的官网拜读已经结束了,相信诸位童鞋或多或少都能入了个门,如果还是觉得迷迷糊糊似懂非懂的,那我建议诸位可以亲自去拜读拜读官网的技术手册或者看多几篇我的视频跟源码!因为接下来我们将进入第二阶段,即应用实战阶段(备注:第一阶段的内容主要以视频的形式分享,感兴趣的童鞋可以加QQ群:583522159 自行获取) 实战分析:应用实战,当然是指真正的在企业级项目中的应用.在这一阶段中,我将以目前流行的微服务架构为奠基,整合RabbitMQ实现项目中常见

Java SpringBoot集成RabbitMq实战和总结

目录 交换器.队列.绑定的声明 关于消息序列化 同一个队列多消费类型 注解将消息和消息头注入消费者方法 关于消费者确认 关于发送者确认模式 消费消息.死信队列和RetryTemplate RPC模式的消息(不常用) 关于消费模型 关于RabbitMq客户端的线程模型 在公司里一直在用RabbitMQ,由于api已经封装的很简单,关于RabbitMQ本身还有封装的实现没有了解,最近在看RabbitMQ实战这本书,结合网上的一些例子和spring文档,实现了RabbitMQ和spring的集成,对着