MQ消息队列的基础使用（以RabbitMQ为例）_Java

一、同步调用与异步调用

1、同步调用

在这里插入图片描述
同步调用优点：

时效性强，可以立即得到结果

同步调用存在的问题

（1）耦合度高：每次加入新的需求，都要修改原来的代码

（2）性能下降：调用者需要等待服务提供者响应，如果调用链过长则响应时间等于每次调用的时间之和

（3）资源浪费：调用链中的每个服务在等待响应过长中，不能释放请求占用的资源，高并发场景下会极度浪费系统资源

（4）级联失败：如果服务提供者出现问题，所有调用方都会跟着出问题，如同多米诺骨牌一样，迅速导致整个微服务群故障

2、异步调用

在这里插入图片描述
优点：

耦合度低
吞吐量提高
故障隔离
流量削峰

缺点：

依赖于broker的可靠性、安全性、吞吐能力
架构复杂，业务没有明显的流程线，不好追踪管理

二、初识mq

mq（messagequeue），中文是消息队列，字面意思就是存放消息的队列。也就是事件驱动架构中的broker。
在这里插入图片描述

三、rabbitmq

docker安装rabbitmq步骤

1、拉取镜像

docker pull rabbitmq

2、安装rabbitmq，执行命令

docker run -d -p 15672:15672 -p 5672:5672 \
        -e rabbitmq_default_user=admin \
        -e rabbitmq_default_pass=miss5016 \
        --hostname mq1 \
        --name my-rabbitmq\
        rabbitmq:latest

3、开放防火墙和安全组对应的端口

# 开放端口防火墙
firewall-cmd --permanent --zone=public --add-port=15672/tcp
firewall-cmd --permanent --zone=public --add-port=5672/tcp
# 重启防火墙
systemctl restart firewalld.service

4、启用rabbitmq的web端管理界面

# 进入容器内部
docker exec -it my-rabbitmq /bin/bash
# 允许使用web端管理
rabbitmq-plugins enable rabbitmq_management

5、通过浏览器访问，ip+端口号
在这里插入图片描述
6、输入docker run命令时的账号密码即可进入访问

7、问题

通道被禁用 stats in management ui are disabled on this node
在这里插入图片描述
解决方法：

（1）进入容器内部

docker exec -it my-rabbitmq /bin/bash

（2）切换到rabbitmq的配置目录下

cd /etc/rabbitmq/conf.d/

（3）修改配置

echo management_agent.disable_metrics_collector = false > management_agent.disable_metrics_collector.conf

（4）退出并重启docker容器

# 从容器内部退出
exit
# 重启容器
docker restart my-rabbitmq

（5）重新访问通道
在这里插入图片描述
rabbitmq的结构和概念

在这里插入图片描述

四、rabbitmq的消息模型

官网文档：https://www.rabbitmq.com/tutorials

1、基本消息队列（basicqueue）

在这里插入图片描述
总共分为三个角色：

在这里插入图片描述

publisher：消息发布者，将消息发送到队列queue
queue：消息队列，负责接受消息并缓存消息
consumer：订阅队列，处理队列中的消息

示例，通过springboot的测试类实现

（1）导入依赖

<dependency>
    <groupid>com.rabbitmq</groupid>
    <artifactid>amqp-client</artifactid>
    <version>5.20.0</version>
</dependency>

（2）生产者代码

@test
    void contextloads() {
        try {
            //1、建立连接
            connectionfactory factory = new connectionfactory();
            //2、设置连接参数
            factory.sethost("ip地址"); //主机名
            factory.setport(5672); //连接端口
            factory.setvirtualhost("/"); //虚拟主机
            factory.setusername("用户名"); //用户名
            factory.setpassword("密码"); //密码
            //3、建立连接
            connection connection = factory.newconnection();
            //4、创建通道
            channel channel = connection.createchannel();
            //5、创建队列并声明队列
            string queuename = "simple.queue"; //队列名称
            channel.queuedeclare(queuename,false, false, false, null);
            //6、发送消息
            string message = "hello rabbitmq";
            channel.basicpublish("", queuename, null, message.getbytes());
            system.out.println("消息发送完毕。。。");
            //关闭管道和连接
            channel.close();
            connection.close();
        }catch (exception e) {
            system.out.println("生产者发送消息失败 ----- > " + e.getmessage());
        }
    }

（3）消费者代码

@test
    void contextloads() {
        try {
            //1、建立连接
            connectionfactory factory = new connectionfactory();
            //2、设置连接参数
            factory.sethost("ip地址"); //主机名
            factory.setport(5672); //连接端口
            factory.setvirtualhost("/"); //虚拟主机
            factory.setusername("用户名"); //用户名
            factory.setpassword("密码"); //密码
            //3、建立连接
            connection connection = factory.newconnection();
            //4、创建通道
            channel channel = connection.createchannel();
            //5、创建队列并声明队列
            string queuename = "simple.queue"; //队列名称
            channel.queuedeclare(queuename,false, false, false, null);
            //6、订阅消息
            channel.basicconsume(queuename, true, new defaultconsumer(channel){
                @override
                public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                    //将字节流内容转为消息本体
                    string message = new string(body);
                    system.out.println("消费者接收到消息 --------- > " + message);
                }
            });
            system.out.println("等待消息中。。。。");
            //关闭管道和连接
            channel.close();
            connection.close();
        }catch (exception e) {
            system.out.println("生产者发送消息失败 ----- > " + e.getmessage());
        }
    }

基本消息队列的消息发送流程：

建立connection
创建channel
利用channel声明队列
利用channel向队列发送消息

基本消息队列的消息接收流程

建立connection
创建channel
利用channel声明队列
定义consumer的消费行为handledelivery()
利用channel将消费者与队列绑定

2、工作消息队列（workqueue）

在这里插入图片描述

3、发布订阅（publish、subscribe）模式

根据交换机的类型，可分为不同的三种

（1）广播（fanout exchange）
在这里插入图片描述
（2）路由（direct exchange）

（3）主题（topic exchange）

五、springamqp

1、介绍

amqp：advanced message queueing protocol，是用于在应用程序或之间传递业务消息的开放标准，该协议与语言平台无关，更符合微服务中独立性的要求。

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

2、示例

简单消息队列

发送消息：

（1）在父工程中引入依赖，确保每个子工程都能使用

<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
<dependency>
    <groupid>org.springframework.boot</groupid>
    <artifactid>spring-boot-starter-amqp</artifactid>
</dependency>

（2）编写配置类

spring:
  rabbitmq:
    host: # 消息队列ip
    port: # 端口
    virtual-host: # 虚拟地址
    username: # 用户名
    password: # 密码

（3）测试类

@springboottest
class orderserviceapplicationtests {
    @autowired
    private rabbittemplate rabbittemplate;
    /**
     * 发送消息
     */
    @test
    void sendmessagemq() {
        //队列名称
        string queuename = "simple.queue";
        //消息
        string message = "你好，rabbitmq";
        //发送消息
        rabbittemplate.convertandsend(queuename, message);
    }

}

接收消息：

（1）引入依赖

<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
<dependency>
    <groupid>org.springframework.boot</groupid>
    <artifactid>spring-boot-starter-amqp</artifactid>
</dependency>

（2）编写配置类

spring:
  rabbitmq:
    host: # 消息队列ip
    port: # 端口
    virtual-host: # 虚拟地址
    username: # 用户名
    password: # 密码

（3）接收并处理消息

@component //监听消息的监听器
public class springrabbitlistener {
	//需要监听哪个消息队列中的消息
    @rabbitlistener(queues = "simple.queue")
    public void listenersimplequeue(string message) {
        //处理消息
        system.out.println("监听到 simple.queue 队列中的消息 --------- > " + message);
    }
}

注意：消息一旦消费就会从队列中移除，rabbitmq没有消息回溯功能

work queue 工作队列

提高消息处理的速度，避免消息堆积

示例：

生产者，生产100条消息

@test
    void sendmessageworkqueue() throws interruptedexception {
        string queuename = "simple.queue";
        string message = "你好，rabbitmq__";
        for (int i = 1; i <= 100; i++) {
            rabbittemplate.convertandsend(queuename, message+i);
            thread.sleep(20);
        }
    }

两个消费者，共同消费，设置不同的处理速度

@component
public class springrabbitlistener {

    @rabbitlistener(queues = "simple.queue")
    public void listenerwork1queue(string message) throws interruptedexception {
        system.out.println("消费者1消费了消息 --------- > " + message + " " + localdatetime.now());
        thread.sleep(20);
    }

    @rabbitlistener(queues = "simple.queue")
    public void listenerwork2queue(string message) throws interruptedexception {
        system.err.println("消费者2......消费了消息 --------- > " + message + " " + localdatetime.now());
        thread.sleep(200);
    }
}

结果：即使设置了不同的处理速度，但是两个消费者仍然处理了相同的消息数量，并没有因为处理得快的消费者会处理更多消息

原因：因为rabbitmq中的一个消息预取机制造成的

消息预取：当有大量消息到达消息队列时，队列会进行消息投递，消费者的通道（channel）会预先把消息取出，就会平均分配队列中的消息，并不会因为消费者的处理能力进行分配

控制方法：

修改配置文件，设置值，可以控制预取消息的上限

spring:
  rabbitmq:
    host: # 消息队列ip
    port: # 端口
    virtual-host: # 虚拟地址
    username: # 用户名
    password: # 密码
    listener:
      simple:
        prefetch: 1 # 每次只能取1条消息，处理完后再取

发布（publish）、订阅（subscribe）

发布订阅模式允许将同一消息发送给多个消费者。实现方式是加入了exchange（交换机）。

在这里插入图片描述

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

发布订阅-fanout exchange（广播）

fanout exchange会将接收到的消息路由到每一个跟其绑定的queue（队列）

在这里插入图片描述

示例：

（1）在消费者中声明，将交换机与队列进行绑定

声明两个队列和一个交换机，进行绑定

@configuration
public class fanoutconfig {
    /**
     * 声明交换机
     * @return
     */
    @bean
    public fanoutexchange fanoutexchange() {
        //返回一个声明的交换机
        return new fanoutexchange("fanout.exchange");    
    }

    /**
     * 声明第一个队列
     * @return
     */
    @bean
    public queue fanoutqueue1() {
        return new queue("fanout.queue1");
    }

    /**
     * 声明第二个队列
     * @return
     */
    @bean
    public queue fanoutqueue2() {
        return new queue("fanout.queue2");
    }

    /**
     * 绑定队列到交换机
     * @param fanoutqueue1
     * @param fanoutexchange
     * @return
     */
    @bean
    public binding bindingfanoutqueue1(queue fanoutqueue1, fanoutexchange fanoutexchange) {
        //将队列1绑定到交换机
        return bindingbuilder.bind(fanoutqueue1).to(fanoutexchange);
    }
    /**
     * 绑定队列到交换机
     * @param fanoutqueue2
     * @param fanoutexchange
     * @return
     */
    @bean
    public binding bindingfanoutqueue2(queue fanoutqueue2, fanoutexchange fanoutexchange) {
        //将队列2绑定到交换机
        return bindingbuilder.bind(fanoutqueue2).to(fanoutexchange);
    }
}

（2）处理消息

@component
public class springrabbitlistener {
    @rabbitlistener(queues = "fanout.queue1")
    public void listenerfanoutqueue1(string message) {
        system.out.println("监听到 fanout.queue1 队列中的消息 --------- > " + message);
    }

    @rabbitlistener(queues = "fanout.queue2")
    public void listenerfanoutqueue2(string message) {
        system.out.println("监听到 fanout.queue2 队列中的消息 --------- > " + message);
    }
}

（3）发送消息

/**
 * 发送消息到交换机
 */
@test
void testsendfanoutexchange() {
    string exchangename = "fanout.exchange";
    string message = "交换机转发的消息";
    rabbittemplate.convertandsend(exchangename, "", message);
}

这样就实现了每个队列都能消费消息，而不是消息一旦被其中一个消费者消费后，就会移除的问题

发布订阅-directexchange（路由模式）

direct exchange会将接收到的消息根据规则路由到指定的queue，因此称为路由模式（routes）。

每一个queue都会与exchange设置一个bindingkey
发布者发送消息时，指定消息的routingkey
exchange将消息路由到bindingkey与消息routingkey一致的队列

示例：

（1）声明队列，交换机，通过注解的方式

@component
public class springdirectlistener {
    /**
     * 声明第一个消费者
     * @param message
     */
    @rabbitlistener(bindings = @queuebinding(
            value = @queue(name = "direct.queue1"),
            exchange = @exchange(name = "direct.exchange", type = exchangetypes.direct),
            key = {"red", "blue"}
    ))
    public void listenerdirect1queue(string message) {
        system.out.println("direct消费者1监听到的消息 ---------- > " + message);
    }

    /**
     * 声明第二个消费者
     * @param message
     */
    @rabbitlistener(bindings = @queuebinding(
            value = @queue(name = "direct.queue2"),
            exchange = @exchange(name = "direct.exchange", type = exchangetypes.direct),
            key = {"red", "yellow"}
    ))
    public void listenerdirect2queue(string message) {
        system.out.println("direct消费者2监听到的消息 ---------- > " + message);
    }
}

（2）发消息

/**
 * 发送消息到交换机
 */
@test
void testsenddirectexchange() {
    string exchangename = "direct.exchange";
    string message = "direct交换机转发的消息";
    string routingkey = "red"; //发送给指定routingkey的队列
    rabbittemplate.convertandsend(exchangename, routingkey, message);
}

发布订阅-topicexchange（主题模式）

topicexchange与directexchange类似，区别在于routingkey必须是多个单词的列表，并且用 . 分割
在这里插入图片描述

示例：

（1）声明交换机，队列

@component
public class springtopiclistener {

    @rabbitlistener(bindings = @queuebinding(
            value = @queue("topic.queue1"),
            exchange = @exchange(value = "topic.exchange", type = exchangetypes.topic),
            key = {"china.#"}
    ))
    public void listenertopic1queue(string message) {
        system.out.println("topic消费者1监听到的消息 ---------- > " + message);
    }

    @rabbitlistener(bindings = @queuebinding(
            value = @queue("topic.queue2"),
            exchange = @exchange(value = "topic.exchange", type = exchangetypes.topic),
            key = {"#.news"}
    ))
    public void listenertopic2queue(string message) {
        system.out.println("topic消费者2监听到的消息 ---------- > " + message);
    }
}

（2）测试消息发送

/**
 * 发送消息到交换机
 */
@test
void testsendtopicexchange() {
    string exchangename = "topic.exchange";
    string message = "topic交换机转发的消息";
    string routingkey = "11.news";
    rabbittemplate.convertandsend(exchangename, routingkey, message);
}

六、消息转换器

测试：

（1）新增一个队列（queue）

/**
 * 声明一个队列
 * @return
 */
@bean
public queue objectqueue() {
    return new queue("object.queue");
}

（2）测试发送对象类型数据

/**
 * 发送消息
 */
@test
void sendobjectmessagemq() {
    string queuename = "object.queue";
    map<string, object> map = new hashmap<>();
    map.put("username", "张三");
    map.put("age", 12);
    rabbittemplate.convertandsend(queuename, map);
}

结果：
在这里插入图片描述

原因：

rabbitmq只支持字节，而spring允许发object对象，说明spring会将对象做序列化，用的序列化方式是jdk的序列化方式x-java-serialized-object，这种序列化缺点是性能差，安全性有问题，可能存在注入风险，数据长度太长了

spring的对象处理事由org.springframework.amqp.support.converter.messageconverter来处理的，而默认实现的simplemessageconverter，基于jdk的objectoutputstream完成序列化。如果要修改需要定义一个messageconverter类型的bean即可。

步骤：

（1）导入依赖

<!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.dataformat/jackson-dataformat-xml -->
<dependency>
    <groupid>com.fasterxml.jackson.dataformat</groupid>
    <artifactid>jackson-dataformat-xml</artifactid>
    <version>2.17.1</version>
</dependency>

（2）声明一个bean，将消息转换的bean注入spring容器中

/**
 * 声明消息转换类型
 * @return
 */
@bean
public messageconverter messageconverter() {
    return new jackson2jsonmessageconverter();
}

（3）将消息清除，重新测试
在这里插入图片描述
消息接收：

（1）引入依赖

<!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.dataformat/jackson-dataformat-xml -->
<dependency>
    <groupid>com.fasterxml.jackson.dataformat</groupid>
    <artifactid>jackson-dataformat-xml</artifactid>
    <version>2.17.1</version>
</dependency>

（2）注入bean

/**
 * 声明消息转换类型
 * @return
 */
@bean
public messageconverter messageconverter() {
    return new jackson2jsonmessageconverter();
}

（3）消息接收处理

@rabbitlistener(queues = "object.queue")
public void listenerobjectqueue(map<string, object> map) {
    system.out.println("监听到 object.queue 队列中的消息 --------- > " + map);
}

MQ消息队列的基础使用（以RabbitMQ为例）

2024年07月31日 • Java •我要评论

一、同步调用与异步调用

1、同步调用

2、异步调用

二、初识mq

三、rabbitmq

四、rabbitmq的消息模型

1、基本消息队列（basicqueue）

2、工作消息队列（workqueue）

3、发布订阅（publish、subscribe）模式

五、springamqp

1、介绍

2、示例

简单消息队列

work queue 工作队列

发布（publish）、订阅（subscribe）

发布订阅-fanout exchange（广播）

发布订阅-directexchange（路由模式）

发布订阅-topicexchange（主题模式）

六、消息转换器

相关文章:

计算机毕业设计hadoop+spark+hive知识图谱酒店推荐系统酒店数据分析可视化大屏酒店爬虫高德地图API 酒店预测系统大数据毕业设计

Hadoop 安装及目录结构

什么是Flink CDC，以及如何使用_flinkcdc(1)，一文全懂

HDFS读写流程详细过程

大数据基础知识

发表评论


验证码：

MQ消息队列的基础使用（以RabbitMQ为例）

2024年07月31日 • Java •我要评论

一、同步调用与异步调用

1、同步调用

2、异步调用

二、初识mq

三、rabbitmq

四、rabbitmq的消息模型

1、基本消息队列（basicqueue）

2、工作消息队列（workqueue）

3、发布订阅（publish、subscribe）模式

五、springamqp

1、介绍

2、示例

简单消息队列

work queue 工作队列

发布（publish）、订阅（subscribe）

发布订阅-fanout exchange（广播）

发布订阅-directexchange（路由模式）

发布订阅-topicexchange（主题模式）

六、消息转换器

相关文章:

计算机毕业设计hadoop+spark+hive知识图谱酒店推荐系统 酒店数据分析可视化大屏 酒店爬虫 高德地图API 酒店预测系统 大数据毕业设计

Hadoop 安装及目录结构

什么是Flink CDC，以及如何使用_flinkcdc(1)，一文全懂

HDFS读写流程详细过程

大数据基础知识

发表评论

计算机毕业设计hadoop+spark+hive知识图谱酒店推荐系统酒店数据分析可视化大屏酒店爬虫高德地图API 酒店预测系统大数据毕业设计