0
0

RabbitMQ

RabbitMQ

什么是消息队列

消息（Message）是指在应用间传送的数据。消息可以非常简单，比如只包含文本字符串，也可以更复杂，可能包含嵌入对象。

消息队列（Message Queue）是一种应用间的通信方式，消息发送后可以立即返回，由消息系统来确保消息的可靠传递。消息发布者只管把消息发布到 MQ 中而不用管谁来取，消息使用者只管从 MQ 中取消息而不管是谁发布的

消息队列常见实现与技术对比

MQ，中文是消息队列（MessageQueue），字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。

比较常见的MQ实现：

  • ActiveMQ
  • RabbitMQ
  • RocketMQ
  • Kafka

几种常见MQ的对比：
RabbitMQActiveMQRocketMQ****Kafka公司/社区RabbitApache阿里Apache开发语言ErlangJavaJavaScala&Java协议支持AMQP，XMPP，SMTP，STOMPOpenWire,STOMP，REST,XMPP,AMQP自定义协议自定义协议可用性高一般高高单机吞吐量一般差高非常高消息延迟微秒级毫秒级毫秒级毫秒以内消息可靠性高一般高一般
追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

RabbitMQ

RabbitMQ是基于Erlang语言开发的开源消息通信中间件，官网地址：https://www.rabbitmq.com/

具有：大并发、消息可靠、响应快速、等优点

安装RabbitMQ

docker安装

拉取镜像

docker pull rabbitmq:版本

运行容器

docker run \
 -e RABBITMQ_DEFAULT_USER=root \
 -e RABBITMQ_DEFAULT_PASS=123456 \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3-management

注意：外网访问需要开放15672和5672端口

访问http://ip:15672即可进入web管理界面

RabbitMQ中的几个概念

  • channel:操作MQ的工具
  • exchange:路由消息到队列中
  • queue:缓存消息
  • virtual host:虚拟主机，是对queue，exchange等资源的逻辑分组

常见消息模型

  • 基本消息队列（BasicQueue）
  • 工作消息队列（WorkQueue）
  • 发布订阅（Publish、Subscribe）： - Fanout Exchange:广播- Direct Exchange:路由- Topic Exchange:主题

SpringAMQP

SpringAmqp的官方地址：https://spring.io/projects/spring-amqp

AMQP

Advanced Message Queuing Protocol，是用于在应用程序之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求

Spring AMQP

Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。包含两部分，其中spring-amqp是基础抽象，spring-rabbit是底层的默认实现。

Spring AMQP实现简单消息发送与接收

依赖

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

基础配置

spring:rabbitmq:host: ip # rabbitMQ的ip地址port:5672 # 端口username: 用户名
    password: 密码
    virtual-host: /
    listener:simple:prefetch:1

消息发送

@Autowiredprivate RabbitTemplate rabbitTemplate;@TestvoidtestSimpleQueue(){
        String queueName ="simple.queue";
        String message ="hello amqp";
        rabbitTemplate.convertAndSend (queueName,message);}

消息接收

@ComponentpublicclassSpringRabbitListener{@RabbitListener(queues ="simple.queue")publicvoidlistenSimpleQueue(String msg){
        System.out.println ("接收到消息"+msg);}}

Work Queue工作队列

Work queue，工作队列可以提高消息处理速度，避免队列消息堆积

消息发送

模拟WorkQueue，实现一个队列绑定多个消费者

@TestvoidtestSimpleQueue()throws InterruptedException {
        String queueName ="simple.queue";
        String message ="hello amqp__";for(int i =0; i <50; i++){
            rabbitTemplate.convertAndSend (queueName,message + i);
            Thread.sleep (20);}}

@RabbitListener(queues ="simple.queue")publicvoidlistenWorkQueue1(String msg)throws InterruptedException {
        System.out.println ("接收到消息消费者1："+msg);
        Thread.sleep (20);}@RabbitListener(queues ="simple.queue")publicvoidlistenWorkQueue2(String msg)throws InterruptedException {
        System.out.println ("接收到消息消费者2："+msg);
        Thread.sleep (200);}

注意RabbitMq默认两个消费者消费一样多，需要进行配置

spring:rabbitmq:listener:simple:prefetch:1 #一次只拿一条消息

注意：

  • 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
  • 通过设置prefetch来控制消费者预取的消息数量

发布（Pubmish）、订阅（Subscribe）

发布订阅模式与简单队列与工作队列的区别就是允许将同一条消息发送给多个发送者。实现方式就是加入了exchange（交换机）

常见的exchange类型包括：

  • Fanout:广播
  • Direct:路由
  • Topic:话题

注意：exchange负责消息路由，而不是存储，路由失败则消息丢失

发布订阅-Fanout Exchange

Fanout Exchange会将接收到的消息广播到每一个跟其绑定的queue

在消费者服务中声明Exchange、Queue、Binding

基于Bean实现

/**
 * @Author: 生如夏花
 * @Date: 2022/3/15 17:07
 */@ConfigurationpublicclassFanoutConfig{/**
     * 声明一个交换机
     * @return
     */@Beanpublic FanoutExchange fanoutExchange(){returnnewFanoutExchange("test.fanout");}/**
     * 声明一个队列
     * @return
     */@Beanpublic Queue queue1(){returnnewQueue("fanout.queue1");}/**
     * 绑定交换机与队列
     * @param queue1
     * @param fanoutExchange
     * @return
     */@Beanpublic Binding fanoutBinding1(Queue queue1,FanoutExchange fanoutExchange){return BindingBuilder
                .bind (queue1).to (fanoutExchange);}@Beanpublic Queue queue2(){returnnewQueue("fanout.queue2");}@Beanpublic Binding fanoutBinding2(Queue queue2,FanoutExchange fanoutExchange){return BindingBuilder
                .bind (queue2).to (fanoutExchange);}}

发送消息

/**
     * 广播模式发送消息
     * @throws InterruptedException
     */@TestvoidtestSendFanoutExchange()throws InterruptedException {
        String exchange ="test.fanout";
        String message ="hello amqp__";
        rabbitTemplate.convertAndSend (exchange,"",message);}

接收消息

@RabbitListener(queues ="fanout.queue1")publicvoidlistenWorkQueue1(String msg){
    System.out.println ("接收到消息消费者1："+msg);}@RabbitListener(queues ="fanout.queue2")publicvoidlistenWorkQueue2(String msg){
    System.out.println ("接收到消息消费者2："+msg);}
发布订阅-DirectExchange

Direct Exchange会将接收到的消息根据路由到指定Queue,因此称为路由模式（routes）

  • 每一个Queue都与Exchange设置一个Bindingkey
  • 发布者发送消息时，指定消息的RoutingKey
  • Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

发送消息

/**
     * 路由模式发送消息
     */@TestvoidtestSendDirectExchange(){
        String exchange ="test.direct";
        String colorKey ="yellow";
        String message ="hello direct---";
        rabbitTemplate.convertAndSend (exchange,colorKey,message+colorKey);}

接收消息

基于@RabbitListener注解实现

@RabbitListener(bindings =@QueueBinding(
        value =@Queue(name ="direct.queue1"),
        exchange =@Exchange(name ="test.direct",type = ExchangeTypes.DIRECT),
        key ={"red","blue"}))publicvoidlistenDirectQueue1(String msg){
    System.out.println ("接收到消息:"+msg);}@RabbitListener(bindings =@QueueBinding(
        value =@Queue(name ="direct.queue2"),
        exchange =@Exchange(name ="test.direct"),
        key ={"red","yellow"}))publicvoidlistDirectQueue2(String msg){
    System.out.println ("接收到消息："+msg);}
发布订阅-TopicExchange

TopicExchange与DirectExchange类似，区别在于routeingKey必须是多个单词的列表，并且以.分割

Queue与Exchange指定BindingKey时可以使用通配符

  • #：代指0个或者多个单词
  • *：代指一个单词

发送消息

/**
 * 话题模式发送消息
 */@TestvoidtestSendTopicExchange(){
    String exchange ="test.topic";
    String key ="315.news";
    String message ="央视综合频道正在进行315晚会直播";
    rabbitTemplate.convertAndSend (exchange,key,message);}

接收消息

@RabbitListener(bindings =@QueueBinding(
        value =@Queue("topic.queue1"),
        exchange =@Exchange(value ="test.topic",type = ExchangeTypes.TOPIC),
        key ={"#.news"}))//会接收所有.news结尾的单词publicvoidlistTopicQueue1(String msg){
    System.out.println ("接收到消息："+msg);}

消息转换器

在SpringAMQP的发送方法中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息。SpringAMQP会帮我们序列化为字节后发送

注意：Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的，基于jdk的ObjectOutputStream完成序列化。

如果要修改只需要定义一个MessageConverter类型的Bean即可。推荐使用json方式序列化，步骤如下

引入依赖

<dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId></dependency>

创建bean

@Beanpublic MessageConverter messageConverter(){returnnewJackson2JsonMessageConverter();}

然后正常收发消息即可，就可以实现通过json来序列化发送接收消息的对象

标签： rabbitmq java
本文转载自: https://blog.csdn.net/xu2414506319/article/details/123514016
版权归原作者 许喜朝 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

机器人技术在安全领域的重要作用

安全学习记录——网络篇（一）

图腾柱PFC：HP1010为您的电动两轮车之旅提供绿色，高效，安全的动力

单元测试数据库回滚问题

这才是单元测试，之前我们都错了

【C语言基础】：字符函数和字符串函数

ChatGPT-Next-Web SSRF漏洞+XSS漏洞复现(CVE-2023-49785)

文章导航