SpringBoot集成RabbitMQ

写程序的小王叔叔 2023-03-28 原文

主页:小王叔叔的博客欢迎来访
支持:点赞收藏关注
RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。

1.RabbitMQ介绍

RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

2.相关概念

通常我们谈到队列服务, 会有三个概念：发消息者、队列、收消息者，RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和队列之间, 加入了交换器 (Exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列

左侧 P 代表生产者，也就是往 RabbitMQ 发消息的程序。
中间即是 RabbitMQ，其中包括了交换机和队列。
右侧 C 代表消费者，也就是往 RabbitMQ 拿消息的程序。

其中比较重要概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。

虚拟主机：一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单，RabbitMQ当中，用户只能在虚拟主机的粒度进行权限控制。因此，如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创建一个虚拟主机。每一个RabbitMQ服务器都有一个默认的虚拟主机“/”。
交换机：Exchange 用于转发消息，但是它不会做存储，如果没有 Queue bind 到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息。

这里有一个比较重要的概念：路由键。消息到交换机的时候，交互机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。

绑定：也就是交换机需要和队列相绑定，这其中如上图所示，是多对多的关系。

SpringBoot集成RabbitMQ非常简单，如果只是简单的使用配置非常少，springboot提供了spring-boot-starter-amqp项目对消息各种支持。

3.简单使用

1.配置pom包

主要是添加spring-boot-starter-amqp的支持

org.springframework.boot
        spring-boot-starter-amqp

2.配置文件

配置rabbitmq的安装地址、端口以及账户信息.

spring.application.name=spirng-boot-rabbitmq

spring.rabbitmq.host=192.168.0.86
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

3.队列配置

@Configuration
public class RabbitConfig {

    @Bean
    public Queue Queue() {
        return new Queue("hello");
    }

}

4.发送者

rabbitTemplate是springboot 提供的默认实现
public class HelloSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String context = "hello " + new Date();
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("hello", context);
    }

}

5.接收者

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

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

}

6.测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqHelloTest {

    @Autowired
    private HelloSender helloSender;

    @Test
    public void hello() throws Exception {
        helloSender.send();
    }

}

注意，发送者和接收者的queue name必须一致，不然不能接收

多对多使用

一个发送者，N个接收者或者N个发送者和N个接收者会出现什么情况呢？

一对多发送

对上面的代码进行了小改造,接收端注册了两个Receiver,Receiver1和Receiver2，发送端加入参数计数，接收端打印接收到的参数，下面是测试代码，发送一百条消息，来观察两个接收端的执行效果.

@Test
public void oneToMany() throws Exception {
    for (int i=0;i<100;i++){
        neoSender.send(i);
    }
}
结果如下：
Receiver 1: spirng boot neo queue ****** 11
Receiver 2: spirng boot neo queue ****** 12
Receiver 2: spirng boot neo queue ****** 14
Receiver 1: spirng boot neo queue ****** 13
Receiver 2: spirng boot neo queue ****** 15
Receiver 1: spirng boot neo queue ****** 16
Receiver 1: spirng boot neo queue ****** 18
Receiver 2: spirng boot neo queue ****** 17
Receiver 2: spirng boot neo queue ****** 19
Receiver 1: spirng boot neo queue ****** 20

根据返回结果得到以下结论

一个发送者，N个接受者,经过测试会均匀的将消息发送到N个接收者中

多对多发送

复制了一份发送者，加入标记，在一百个循环中相互交替发送

@Test
    public void manyToMany() throws Exception {
        for (int i=0;i<100;i++){
            neoSender.send(i);
            neoSender2.send(i);
        }
}
结果如下：
Receiver 1: spirng boot neo queue ****** 20
Receiver 2: spirng boot neo queue ****** 20
Receiver 1: spirng boot neo queue ****** 21
Receiver 2: spirng boot neo queue ****** 21
Receiver 1: spirng boot neo queue ****** 22
Receiver 2: spirng boot neo queue ****** 22
Receiver 1: spirng boot neo queue ****** 23
Receiver 2: spirng boot neo queue ****** 23
Receiver 1: spirng boot neo queue ****** 24
Receiver 2: spirng boot neo queue ****** 24
Receiver 1: spirng boot neo queue ****** 25
Receiver 2: spirng boot neo queue ****** 25

结论：和一对多一样，接收端仍然会均匀接收到消息.

4.高级使用

对象的支持
springboot以及完美的支持对象的发送和接收，不需要格外的配置。
//发送者
public void send(User user) {
    System.out.println("Sender object: " + user.toString());
    this.rabbitTemplate.convertAndSend("object", user);
}

...

//接受者
@RabbitHandler
public void process(User user) {
    System.out.println("Receiver object : " + user);
}
结果如下：
Sender object: User{name='neo', pass='123456'}
Receiver object : User{name='neo', pass='123456'}

Topic Exchange

topic 是RabbitMQ中最灵活的一种方式，可以根据routing_key自由的绑定不同的队列

首先对topic规则配置，这里使用两个队列来测试

@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("exchange");
    }

    @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.#");
    }
}

使用queueMessages同时匹配两个队列，queueMessage只匹配"topic.message"队列

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

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

发送send1会匹配到topic.#和topic.message 两个Receiver都可以收到消息，发送send2只有topic.#可以匹配所有只有Receiver2监听到消息

Fanout Exchange

Fanout 就是我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。

Fanout 相关配置

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

}
这里使用了A、B、C三个队列绑定到Fanout交换机上面，发送端的routing_key写任何字符都会被忽略：
public void send() {
        String context = "hi, fanout msg ";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}
结果如下：
Sender : hi, fanout msg 
...
fanout Receiver B: hi, fanout msg 
fanout Receiver A  : hi, fanout msg 
fanout Receiver C: hi, fanout msg

结果说明，绑定到fanout交换机上面的队列都收到了消息.

⚠️注意 ~

?本期内容就结束了，如果内容有误，麻烦大家评论区指出！
如有疑问❓可以在评论区?或私信?，尽我最大能力?‍♀️帮大家解决?‍?！

如果我的文章有帮助到您，欢迎点赞+关注✔️鼓励博主?，您的鼓励是我分享的动力???~

有关SpringBoot集成RabbitMQ的更多相关文章

ruby-on-rails - 如何使辅助方法在 Rails 集成测试中可用？ - 2
我在app/helpers/sessions_helper.rb中有一个帮助程序文件，其中包含一个方法my_preference，它返回当前登录用户的首选项。我想在集成测试中访问该方法。例如，这样我就可以在测试中使用getuser_path(my_preference)。在其他帖子中，我读到这可以通过在测试文件中包含requiresessions_helper来实现，但我仍然收到错误NameError:undefinedlocalvariableormethod'my_preference'.我做错了什么？require'test_helper'require'sessions_hel
ruby-on-rails - 我如何将 Hoptoad 与 DelayedJob 和 DaemonSpawn 集成？ - 2
我一直很高兴地使用DelayedJob习惯用法:foo.send_later(:bar)这会调用DelayedJob进程中对象foo的方法bar。我一直在使用DaemonSpawn在我的服务器上启动DelayedJob进程。但是...如果foo抛出异常，Hoptoad不会捕获它。这是任何这些包中的错误...还是我需要更改某些配置...或者我是否需要在DS或DJ中插入一些异常处理来调用Hoptoad通知程序？回应下面的第一条评论。classDelayedJobWorker 最佳答案尝试monkeypatchingDelayed::W
jenkins部署1--jenkins+gitee持续集成 - 2
前置步骤我们都操作完了，这篇开始介绍jenkins的集成。话不多说，看操作1、登录进入jenkins后会让你选择安装插件，选择第一个默认的就行。安装完成后设置账号密码，重新登录。2、配置JDK和Git都需要执行路径，所以需要先把执行路径找到，先进入服务器的docker容器，2.1JDK的路径root@69eef9ee86cf:/usr/bin#echo$JAVA_HOME/usr/local/openjdk-82.2Git的路径root@69eef9ee86cf:/#whichgit/usr/bin/git3、先配置JDK和Git。点击：ManageJenkins>>GlobalToolCon
三分钟集成 TapTap 防沉迷 SDK（Unity 版） - 2
三分钟集成Tap防沉迷SDK（Unity版）一、SDK介绍基于国家对上线所有游戏必须增加防沉迷功能的政策下，TapTap推出防沉迷SDK，供游戏开发者进行接入；允许未成年用户在周五、六、日以及法定节假日晚上8:00-9:00进行游戏，防沉谜时间段进入游戏会弹窗进行提示！开发环境要求：Unity2019.4或更高版本iOS10或更高版本Android5.0（APIlevel21）或更高版本🔗Unity集成Demo参考链接🔗UnityTapSDK功能体验APK下载链接二、集成前准备1.创建应用进入开发者后台,按照提示开始创建应用；2.开通服务在使用TDS实名认证和防沉迷服务之前，需要在上面创建的应
ruby-on-rails - RailsTutorial - 第 8.4.3 章 - 在集成测试中添加用户后未清除测试数据库 - 2
我被这个难住了。到目前为止教程中的一切都进行得很顺利，但是当我将这段代码添加到我的/spec/requests/users_spec.rb文件中时，事情开始变得糟糕:describe"success"doit"shouldmakeanewuser"dolambdadovisitsignup_pathfill_in"Name",:with=>"ExampleUser"fill_in"Email",:with=>"ryan@example.com"fill_in"Password",:with=>"foobar"fill_in"Confirmation",:with=>"foobar"cl
ruby-on-rails - 将 Angular JS 与 Rails 集成 - 2
我需要一些指导来了解如何将Angular整合到rails中。选择Rails的原因:我喜欢他们偏执的做事方式。还有迁移，gem真的很酷。使用angular的原因:我正在研究和寻找最适合SPA的框架。Backbone似乎太抽象了。我不得不在Angular和Ember之间做出选择。我首先开始阅读Angular，它对我来说很有意义。所以我从来没有去读过关于ember的文章。使用Angular和Rails的原因:我研究并尝试使用小型框架，例如grape、slim(是的，我也使用php)。但我觉得需要坚持项目的长期范围。我个人喜欢用Rails的方式做事。这就是我需要帮助的地方，我在Rails4中有
ruby - 在 Maven 集成中运行 Ruby 单元测试 - 2
有没有人有在Maven中运行用Ruby编写的单元测试的经验。任何输入，如要使用的库/maven插件，将不胜感激!我们已经在使用Maven+hudson+Junit。但是我们正在引入Ruby单元测试，找不到任何同样好的组合。最佳答案我建议让Maven使用ExecMavenPlugin启动rake测试(exec:exec目标)并使用ci_reportergem生成单元测试结果的XML文件，Hudson、Bamboo等可以读取该文件，以与JUnit测试相同的格式显示测试结果。如果您不需要使用mvntest运行Ruby测试，您也可以只使
ruby - 使用 Gatling 作为集成测试工具 - 2
目前我有一小套针对我的网络服务器运行的集成测试，它发出请求并断言一些关于响应应该是什么的假设。这些是用Ruby编写的，生成http请求。我一直在看Gatling作为压力测试工具，但我想知道它是否也可以用于集成测试。这样，所有端点请求都可以在压力测试和集成测试中重复使用。我可能在这里失去了一些东西，因为没有RSpec的BDD，但不必两次创建相同的测试。有没有人有这样使用gatling的经验？最佳答案您可以使用AssertionAPI并设置验收标准。但是，Gatling不是浏览器，不会运行/测试您的Javascript，因此这种方法
Spring Boot集成ElasticSearach - 2
文章目录前言一、Elasticsearch版本介绍二、客户端种类三、客户端与版本兼容性四、引入Elasticsearch依赖包五、客户端配置六、Elasticsearch使用前言ElasticSearch是Elastic公司出品的一款功能强大的搜索引擎，被广泛的应用于各大IT公司，它的代码位于https://github.com/elastic/elasticsearch，目前是一个开源项目。ElasticSearch公司的另外两个开源产品Logstash、Kibana与ElasticSearch构成了著名的ELK技术栈。。他们三个共同形成了一个强大的生态圈。简单地说，Logstash负责数据
springboot定时任务 - 2
如果您希望在Spring中启用定时任务功能，则需要在主类上添加 @EnableScheduling 注解。这样Spring才会扫描 @Scheduled 注解并执行定时任务。在大多数情况下，只需要在主类上添加 @EnableScheduling 注解即可，不需要在Service层或其他类中再次添加。以下是一个示例，演示如何在SpringBoot中启用定时任务功能：@SpringBootApplication@EnableSchedulingpublicclassApplication{publicstaticvoidmain(String[]args){SpringApplication.ru