RabbitMQ(一) - 基本结构、SpringBoot整合RabbitMQ、工作队列、发布订阅、直接、主题交换机模式

RabbitMQ结构

  • Publisher : 生产者

  • Queue: 存储消息的容器队列;

  • Consumer:消费者

  • Connection:消费者与消息服务的TCP连接

  • Channel:信道,是TCP里面的虚拟连接。例如:电缆相当于TCP,信道是一条独立光纤束,一条TCP连接上创建多少条信道是没有限制的。TCP一旦打开,就会出AMQP信道。无论是发布消息,接收消息,订阅队列,这些动作都是通过信道完成的。
    Broker: 一台消息服务就是一个Broker;

  • Exchange:交换机、负责接收生产者的消息,转发到队列中、交换机和队列通过路由键绑定、可以理解为每个队列都有自己的名称;
    在这里插入图片描述

SpringBoot整合RabbitMQ

  • Queue
    • 消息存放于队列中, 若是RabbitMQ挂了,则消息会丢失,因此要开启持久化, 将durable设置为true,
    • 若是没有消费者消费该队列,则该队列会自动删除, 因此需要将autoDelete参数设置为false;
    public Queue(String name) {//  队列名称, 是否持久化,是否独占, 是否自动删除this(name, true, false, false);}
  • @RabbitListener
@RabbitListener(bindings=@QueueBinding(value= @Queue(value="${mq.config.queue.info}",autoDelete="true"),exchange=@Exchange(value="${mq.config.exchange}",type=ExchangeTypes.DIRECT),key="${mq.config.queue.info.routing.key}"))

用来标记消费者;exchange表示交换器信息、类型;bindings表示监听器要绑定的队列、以及队列信息;
key:代表交换机和队列通过key绑定的;

  • AmqpTemplate / RabbitTempldate:
    生产者通过依赖此工具类发送消息;

先安装RabbitMQ,创建SpringBoot项目,修改配置

# 应用名称
spring.application.name=boolfilter# 应用服务 WEB 访问端口
server.port=8080spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

入门级别程序

发送hello world程序;
在这里插入图片描述
生产者:

public class Tut1Sender {@Autowiredprivate RabbitTemplate template;@Autowiredprivate Queue queue;@Scheduled(fixedDelay = 1000, initialDelay = 500)public void send() {String message = "Hello World!";this.template.convertAndSend(queue.getName(), message);System.out.println(" [x] Sent '" + message + "'");}
}

消费者:

@RabbitListener(queues = "hello")
public class Tut1Receiver {@RabbitHandlerpublic void receive(String in) {System.out.println(" [x] Received '" + in + "'");}
}

将生产者、消费者注入容器;

@Configuration
@EnableScheduling
public class Tut1Config {@Beanpublic Queue hello() {return new Queue("hello");}@Beanpublic Tut1Receiver receiver() {return new Tut1Receiver();}@Beanpublic Tut1Sender sender() {return new Tut1Sender();}
}

运行结果:

[x] Sent ‘Hello World!’
[x] Received ‘Hello World!’
[x] Sent ‘Hello World!’
[x] Received ‘Hello World!’
[x] Sent ‘Hello World!’

工作队列

主要思想是避免 立即执行资源密集型任务,必须等待它要完成。相反,我们将任务安排在以后完成。我们将任务封装为消息并将其发送到队列。正在运行的工作进程 在后台将弹出任务并最终执行工作
在这里插入图片描述
生产者:

public class Tut2Sender {@Autowiredprivate RabbitTemplate template;@Autowiredprivate Queue queue;AtomicInteger dots = new AtomicInteger(0);AtomicInteger count = new AtomicInteger(0);@Scheduled(fixedDelay = 1000, initialDelay = 500)public void send() {`在这里插入代码片`StringBuilder builder = new StringBuilder("Hello");if (dots.incrementAndGet() == 4) {dots.set(1);}for (int i = 0; i < dots.get(); i++) {builder.append('.');}builder.append(count.incrementAndGet());String message = builder.toString();template.convertAndSend(queue.getName(), message);System.out.println(" [x] Sent '" + message + "'");}}

消费者:

@RabbitListener(queues = "hello")
public class Tut2Receiver {private final int instance;public Tut2Receiver(int i) {this.instance = i;}@RabbitHandlerpublic void receive(String in) throws InterruptedException {StopWatch watch = new StopWatch();watch.start();System.out.println("instance " + this.instance +" [x] Received '" + in + "'");doWork(in);watch.stop();System.out.println("instance " + this.instance +" [x] Done in " + watch.getTotalTimeSeconds() + "s");}private void doWork(String in) throws InterruptedException {for (char ch : in.toCharArray()) {if (ch == '.') {Thread.sleep(1000);}}}
}

队列、生产者、消费者注入容器:

@Configuration
public class Tut2Config {@Beanpublic Queue hello() {return new Queue("hello");}private static class ReceiverConfig {@Beanpublic Tut2Receiver receiver1() {return new Tut2Receiver(1);}@Beanpublic Tut2Receiver receiver2() {return new Tut2Receiver(2);}}@Beanpublic Tut2Sender sender() {return new Tut2Sender();}
}

运行结果:

[x] Sent ‘Hello.1’
instance 1 [x] Received ‘Hello.1’
[x] Sent ‘Hello…2’
instance 2 [x] Received ‘Hello…2’
instance 1 [x] Done in 1.0062309s
[x] Sent ‘Hello…3’
instance 1 [x] Received ‘Hello…3’
instance 2 [x] Done in 2.0085791s
[x] Sent ‘Hello.4’
instance 2 [x] Received ‘Hello.4’

消息确认

  • SpringBoot整合RabbitMQ代码中,若消费者消费出现异常,则会重新进入队列, 一般生产环境中,是要有重试机制的;
  • 若是要关闭重试机制、则设置defaultRequeueRejected=false, 或者抛出AmqpRejectAndDontRequeueException异常,这样框架会帮我们自动提交确认channel.basicAck()
  • 重试机制也会存在问题、若是消费者服务关闭、则消息会不断重新入队、导致RabbitMQ内存最终爆满宕机;
  • 消息的ACK确认机制默认是打开的;如果忘记了ACK,后果很严重,当Consumer退出时,消息会一直重新分发,然后RabbitMq会占用越来越多的内存,由于RabbitMq会长时间运行,出现“内存泄露”是致命的

异常处理方案:

  • 使用try-catch捕捉
  • 使用重试机制、超过一定次数、则丢弃消息或放入死信队列;

spring.rabbitmq.listener.retry.max-attempts=5 //重试超过5次,消息丢弃;

公平调度与循环调度

  • 默认情况下,RabbitMQ 会将每条消息发送给下一个消费者。平均而言,每个消费者将获得相同数量的 消息。这种分发消息的方式称为轮询。 在这种模式下,调度不一定完全按照我们想要的方式工作。 若是存在两台机器,一台性能好、一台性能差, 而RabbitMQ对此一无所知,仍然会调度 消息均匀。发生这种情况是因为 RabbitMQ 只是在消息时调度消息 进入队列。它不看未确认的数量 面向消费者的消息。它只是盲目地发送每 n 条消息 给第 n 个消费者,这就导致了一台机器特别忙碌、一台机器空闲;

  • “公平调度”是Spring AMQP的默认配置。Consumer可以向服务器声明一个prefetchCount, 表示轮到自己时、自己可处理多少消息;这样RabbitMQ转发消息给消费者时、会先看Consumer正在处理的消息数量是否达到了prefetchCount, 若已达到该值,则发给其他的Consumer;

发布/订阅

在这里插入图片描述
特点:一条消息同时会被所有消费者消息;X是交换机(Exchange);交换机和队列进行绑定(Binding)
交换机负责接收生产者发送的消息,再转发消息到队列中;实现了生产者与队列的解耦;

RabbitMQ 中消息传递模型的核心思想是生产者 从不将任何消息直接发送到队列

示例1 : 广播匿名队列

发送者:

public class Tut3Sender {@Autowiredprivate RabbitTemplate template;@Autowiredprivate FanoutExchange fanout;AtomicInteger dots = new AtomicInteger(0);AtomicInteger count = new AtomicInteger(0);@Scheduled(fixedDelay = 1000, initialDelay = 500)public void send() {StringBuilder builder = new StringBuilder("Hello");if (dots.getAndIncrement() == 3) {dots.set(1);}for (int i = 0; i < dots.get(); i++) {builder.append('.');}builder.append(count.incrementAndGet());String message = builder.toString();template.convertAndSend(fanout.getName(), "", message);System.out.println(" [x] Sent '" + message + "'");}}

消费者:

public class Tut3Receiver {@RabbitListener(queues = "#{autoDeleteQueue1.name}")public void receive1(String in) throws InterruptedException {receive(in, 1);}@RabbitListener(queues = "#{autoDeleteQueue2.name}")public void receive2(String in) throws InterruptedException {receive(in, 2);}public void receive(String in, int receiver) throws InterruptedException {StopWatch watch = new StopWatch();watch.start();System.out.println("instance " + receiver + " [x] Received '" + in + "'");doWork(in);watch.stop();System.out.println("instance " + receiver + " [x] Done in "+ watch.getTotalTimeSeconds() + "s");}private void doWork(String in) throws InterruptedException {for (char ch : in.toCharArray()) {if (ch == '.') {Thread.sleep(1000);}}}}

交换机、匿名队列、绑定,生产者、消费者注入容器;

public class Tut3Config {@Beanpublic FanoutExchange fanout() {return new FanoutExchange("tut.fanout");}private static class ReceiverConfig {@Beanpublic Queue autoDeleteQueue1() {return new AnonymousQueue();}@Beanpublic Queue autoDeleteQueue2() {return new AnonymousQueue();}@Beanpublic Binding binding1(FanoutExchange fanout,Queue autoDeleteQueue1) {return BindingBuilder.bind(autoDeleteQueue1).to(fanout);}@Beanpublic Binding binding2(FanoutExchange fanout,Queue autoDeleteQueue2) {return BindingBuilder.bind(autoDeleteQueue2).to(fanout);}@Beanpublic Tut3Receiver receiver() {return new Tut3Receiver();}}@Beanpublic Tut3Sender sender() {return new Tut3Sender();}
}

运行结果:

instance 1 [x] Received 'Hello.1'
instance 2 [x] Received 'Hello.1'
instance 2 [x] Done in 1.0057994s
instance 1 [x] Done in 1.0058073s
....

模拟Spring容器发布ContextRefreshedEvent事件

通常情况下,业务开发中,经常会监听该事件做扩展,例如初始化数据, 打印日志等等;
生产者:

public class AppContextSender {@AutowiredRabbitTemplate rabbitTemplate;@Scheduled(fixedDelay = 1000, initialDelay = 500)public void publishContextRefreshEvent() {rabbitTemplate.convertAndSend("contextRefreshedExchange", "", "publish refreshed event");}
}

消费者:

@RabbitListener(queues = {"initQueue"})
public class InitContextRefreshedConsumer {@RabbitHandlerpublic void consum(String in) {System.out.println("init :"+in);}
}@RabbitListener(queues = "logQueue")
public class LogContextRefreshedConsumer {@RabbitHandlerpublic void consum(String in) {System.out.println("log : "+in);}
}

交换机、队列、绑定、生产者、消费者注入容器:

@Configuration
public class ContextRefreshedConfig {@Beanpublic FanoutExchange contextRefreshedExchange(){return new FanoutExchange("contextRefreshedExchange");}@Beanpublic AppContextSender appContextSender() {return new AppContextSender();}public static class ConsumerConfig {@Beanpublic Queue initQueue() {return new Queue("initQueue");}@Beanpublic Queue logQueue() {return new Queue("logQueue");}@Beanpublic Binding initBinding(Queue initQueue, FanoutExchange contextRefreshedExchange) {return BindingBuilder.bind(initQueue).to(contextRefreshedExchange);}@Beanpublic Binding logBinding(Queue logQueue, FanoutExchange contextRefreshedExchange) {return BindingBuilder.bind(logQueue).to(contextRefreshedExchange);}@Beanpublic InitContextRefreshedConsumer initContextRefreshedConsumer() {return new InitContextRefreshedConsumer();}@Beanpublic LogContextRefreshedConsumer logContextRefreshedConsumer() {return new LogContextRefreshedConsumer();}}}

log : publish refreshed event
init :publish refreshed event
log : publish refreshed event
init :publish refreshed event

Direct直接模式

  • 交换器绑定多个队列,每个绑定关系有自己的路由键;
  • 之前业务开发中、有一个交换机、绑定了两个队列,一个队列用来发送邮件,一个队列用来发送短信, 像广播模式下,如果只想发邮件,则没法t做到,使用direct模式和工作模式则可以做到, 最后使用了direct

在这里插入图片描述
生产者:

public class BaseServiceSender {@Autowiredprivate RabbitTemplate template;@Autowiredprivate DirectExchange messageExchange;AtomicInteger index = new AtomicInteger(0);AtomicInteger count = new AtomicInteger(0);private final String[] keys = {"sms", "mail"};@Scheduled(fixedDelay = 1000, initialDelay = 500)public void send() {//短信String sms = "{userName: xxx; phone:xxx}";template.convertAndSend(messageExchange.getName(), "sms", sms);//邮件String mail = "{userName: xxx; mail:xxx}";template.convertAndSend(messageExchange.getName(), "mail", mail);}
}

消费者:

@RabbitListener(queues = "mailQueue")
public class MailConsumer {@RabbitHandlerpublic void consum(String in) {System.out.println("send mail : " + in);}
}@RabbitListener(queues = "smsQueue")
public class SmsConsumer {@RabbitHandlerpublic void consum(String in) {System.out.println("send sms : " + in);}
}

交换机、队列,绑定、消费者,生产者注入容器:

@Configuration
public class DirectConfig {@Beanpublic DirectExchange messageExchange() {return new DirectExchange("messageExchange");}@Beanpublic BaseServiceSender baseServiceSender() {return new BaseServiceSender();}public static class ConsumerGroup {@Beanpublic MailConsumer mailConsumer() {return new MailConsumer();}@Beanpublic SmsConsumer smsConsumer() {return new SmsConsumer();}@Beanpublic Queue mailQueue() {return new Queue("mailQueue");}@Beanpublic Queue smsQueue() {return new Queue("smsQueue");}@Beanpublic Binding smsBinding(DirectExchange messageExchange, Queue smsQueue){return BindingBuilder.bind(smsQueue).to(messageExchange).with("sms");}@Beanpublic Binding mailBinding(DirectExchange messageExchange, Queue mailQueue){return BindingBuilder.bind(mailQueue).to(messageExchange).with("mail");}}
}

运行结果

send mail : {userName: xxx; mail:xxx}
send sms : {userName: xxx; phone:xxx}
send sms : {userName: xxx; phone:xxx}
send mail : {userName: xxx; mail:xxx}

Topic主题模式

  • 发送到主题交换的消息不能有任意routing_key
    • 它必须是单词列表,由点分隔。这 单词可以是任何东西,一些有效的路由密钥示例: “stock.usd.nyse”, “nyse.vmw”, “quick.orange.rabbit”。可以有 路由密钥中随心所欲地包含多个单词,最多可达 255 个 字节。
  • 绑定密钥也必须采用相同的形式。主题交换背后的逻辑类似于直接交换 - 发送的消息带有 特定的路由键将被传递到所有队列 绑定匹配的绑定键
  • *(星号)可以代替一个词。
  • #(哈希)可以替换零个或多个单词。

在这里插入图片描述
若是消息指定的路由键为"xxx.orange.xxx", 则会匹配到Q1, 若是"lazy.xxx.xx"则是Q2;

生产者:

public class Tut5Sender {@Autowiredprivate RabbitTemplate template;@Autowiredprivate TopicExchange topic;AtomicInteger index = new AtomicInteger(0);AtomicInteger count = new AtomicInteger(0);private final String[] keys = {"quick.orange.rabbit", "lazy.orange.elephant", "quick.orange.fox","lazy.brown.fox", "lazy.pink.rabbit", "quick.brown.fox"};@Scheduled(fixedDelay = 1000, initialDelay = 500)public void send() {StringBuilder builder = new StringBuilder("Hello to ");if (this.index.incrementAndGet() == keys.length) {this.index.set(0);}String key = keys[this.index.get()];builder.append(key).append(' ');builder.append(this.count.incrementAndGet());String message = builder.toString();template.convertAndSend(topic.getName(), key, message);System.out.println(" [x] Sent '" + message + "'");}}

消费者:

public class Tut5Receiver {@RabbitListener(queues = "#{autoDeleteQueue1.name}")public void receive1(String in) throws InterruptedException {receive(in, 1);}@RabbitListener(queues = "#{autoDeleteQueue2.name}")public void receive2(String in) throws InterruptedException {receive(in, 2);}public void receive(String in, int receiver) throwsInterruptedException {StopWatch watch = new StopWatch();watch.start();System.out.println("instance " + receiver + " [x] Received '"+ in + "'");doWork(in);watch.stop();System.out.println("instance " + receiver + " [x] Done in "+ watch.getTotalTimeSeconds() + "s");}private void doWork(String in) throws InterruptedException {for (char ch : in.toCharArray()) {if (ch == '.') {Thread.sleep(1000);}}}
}

交换器,队列,绑定、生产者,消费者注入容器:

@Configuration
public class Tut5Config {@Beanpublic TopicExchange topic() {return new TopicExchange("tut.topic");}private static class ReceiverConfig {@Beanpublic Tut5Receiver receiver() {return new Tut5Receiver();}@Beanpublic Queue autoDeleteQueue1() {return new AnonymousQueue();}@Beanpublic Queue autoDeleteQueue2() {return new AnonymousQueue();}@Beanpublic Binding binding1a(TopicExchange topic,Queue autoDeleteQueue1) {return BindingBuilder.bind(autoDeleteQueue1).to(topic).with("*.orange.*");}@Beanpublic Binding binding1b(TopicExchange topic,Queue autoDeleteQueue1) {return BindingBuilder.bind(autoDeleteQueue1).to(topic).with("*.*.rabbit");}@Beanpublic Binding binding2a(TopicExchange topic,Queue autoDeleteQueue2) {return BindingBuilder.bind(autoDeleteQueue2).to(topic).with("lazy.#");}}@Beanpublic Tut5Sender sender() {return new Tut5Sender();}}

运行结果:

[x] Sent ‘Hello to lazy.orange.elephant 1’
instance 2 [x] Received ‘Hello to lazy.orange.elephant 1’
instance 1 [x] Received ‘Hello to lazy.orange.elephant 1’
[x] Sent ‘Hello to quick.orange.fox 2’
[x] Sent ‘Hello to lazy.brown.fox 3’
instance 1 [x] Done in 2.0110456s

RPC远程过程调用

RabbitMQ也实现了RPC的功能,但是业务开发中,根本没有使用场景,RPC要么使用Dubbo, 要么使用OpenFeign, 使用RabbitMQ做RPC的信息,目前都没有看到;

总结

  • 就目前来说、工作队列、发布订阅两个模式,业务开发中会使用到,其他的消息场景很少见。
  • 底层是基于RabbitMQ-client做的封装出RabbitTempldate使用;除非远古项目,否则不推荐使用RabbitMQ-Client原生API写,太费时间了。我写了一会就放弃了

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/54084.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

如何隐藏开源流媒体EasyPlayer.js视频H.265播放器的实时录像按钮?

目前我们TSINGSEE青犀视频所有的视频监控平台&#xff0c;集成的都是EasyPlayer.js版播放器&#xff0c;它属于一款高效、精炼、稳定且免费的流媒体播放器&#xff0c;可支持多种流媒体协议播放&#xff0c;包括WebSocket-FLV、HTTP-FLV&#xff0c;HLS&#xff08;m3u8&#x…

通讯协议032——全网独有的OPC HDA知识一之聚合(一)

本文简单介绍OPC HDA规范的基本概念&#xff0c;更多通信资源请登录网信智汇(wangxinzhihui)。 本节旨在详细说明HDA聚合的要求和性能。其目的是使HDA聚合标准化&#xff0c;以便HDA客户端能够可靠地预测聚合计算的结果并理解其含义。如果用户需要聚合中的自定义功能&#xff…

【C++】数据结构与算法:常用查找算法

&#x1f60f;★,:.☆(&#xffe3;▽&#xffe3;)/$:.★ &#x1f60f; 这篇文章主要介绍常用查找算法。 学其所用&#xff0c;用其所学。——梁启超 欢迎来到我的博客&#xff0c;一起学习&#xff0c;共同进步。 喜欢的朋友可以关注一下&#xff0c;下次更新不迷路&#x1…

【网络】DNS、ICMP、NAT

目录 一、DNS(Domain Name System) 1、DNS背景 2、域名简介 二、ICMP协议 1、ICMP功能 2、ICMP的报文格式 3、ping命令 4、traceroute命令 三、NAT技术 1、NAT技术背景 2、NAT IP转换过程 3、NAPT 4、NAT技术的缺陷 5、NAT和代理服务器 一、DNS(Domain Name Syste…

Visual Studio在Debug模式下,MFC工程中包含Eigen库时的定义冲突的问题

Visual Studio在Debug模式下&#xff0c;MFC工程中包含Eigen库时的定义冲突的问题 报错信息 Eigen\src\Core\PlainObjectBase.h(143,5): error C2061: 语法错误: 标识符“THIS_FILE” Eigen\src\Core\PlainObjectBase.h(143,1): error C2333: “Eigen::PlainObjectBase::opera…

python+django+mysql项目实践二(前端及数据库)

python项目实践 环境说明&#xff1a; Pycharm 开发环境 Django 前端 MySQL 数据库 Navicat 数据库管理 前端模板 添加模板 在templates下创建 views文件中添加 创建数据库 连接数据库 在setting文件中进行配置 创建表

MyBatis查询数据库(4)

前言&#x1f36d; ❤️❤️❤️SSM专栏更新中&#xff0c;各位大佬觉得写得不错&#xff0c;支持一下&#xff0c;感谢了&#xff01;❤️❤️❤️ Spring Spring MVC MyBatis_冷兮雪的博客-CSDN博客 终于到了MyBatis最后一篇&#xff0c;这篇讲的是动态SQL的使用。 复杂情…

无涯教程-Perl - 环境配置

在开始编写Perl程序之前&#xff0c;让我们了解如何设置我们的Perl环境。 您的系统更有可能安装了perl。只需尝试在$提示符下给出以下命令- $perl -v 如果您的计算机上安装了perl&#xff0c;那么您将收到以下消息: This is perl 5, version 16, subversion 2 (v5.16.2) b…

ES6 数组的用法

1. forEach() 用来循环遍历的 for 数组名.forEach(function (item,index,arr) {})item:数组每一项 , index : 数组索引 , arr:原数组作用: 用来遍历数组 let arr [1, 2, 3, 4]; console.log(arr); let arr1 arr.forEach((item, index, arr) > {console.log(item, index…

推荐几款主流的Css Reset

CSS Reset CSS Reset&#xff08;CSS重置&#xff09;是一种技术&#xff0c;用于消除不同浏览器之间默认样式的差异&#xff0c;以确保网页在各个浏览器中的显示一致性。由于不同浏览器对元素的默认样式有所不同&#xff0c;使用CSS Reset可以将这些默认样式归零或统一&#x…

阿里云安全组设置

简介​ 云主机安全组必须打开如下端口&#xff1a; ssh&#xff1a;22http&#xff1a;80https&#xff1a;443ftp&#xff1a;21、20000&#xff5e;30000 阿里云安全组端口开放教程​ 腾讯云安全组端口开放教程​ 华为云安全组端口开放教程​

第四次作业

1. 简述静态网页和动态网页的区别。 静态页面&#xff1a;请求响应信息&#xff0c;发给客户端进行处理&#xff0c;由浏览器进行解析&#xff0c;显示的页面。在网站设计中&#xff0c;纯粹HTML格式的网页&#xff08;可以包含图片、视频JS (前端功能实现&#xff09;、CSS (…