20221130 RabbitMQ

MQ

MQ（Message Queue）消息队列，是基础数据结构中“先进先出”的一种数据结构。一般用来解决应用解耦，异步消息，流量削峰等问题，实现高性能，高可用，可伸缩和最终一致性架构。
主要的MQ产品包括：RabbitMQ、ActiveMQ、RocketMQ、ZeroMQ、Kafka、IBM WebSphere 等。

AMQP

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

RabbitMQ

RabbitMQ是AMQP服务器的一种。
它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制
RabbitMQ 是一个开源的消息代理和队列服务器，用来通过普通协议在完全不同的应用之间共享数据，RabbitMQ是使用 Erlang语言来编写的，并且RabbitMQ是基于AMQP协议的。
RabbitMQ是一个消息代理 - 一个消息系统的媒介。它可以为你的应用提供一个通用的消息发送和接收平台，并且保证消息在传输过程中的安全。

优点

1. 开源、性能优秀、稳定性保障
2. 提供可靠性消息投递模式(confirm)、返回模式(return)
3. 与SpringAMQP完美的整合、API丰富
4. 集群模式丰富，表达式配置，HA模式，镜像队列模型
5. 保证数据不丢失的前提下做到高可靠性、可用性

架构

RabbitMQ运行机制

组成部分说明如下:

1. Broker:消息队列服务进程，此进程包括两个部分:Exchange和Queue.
2. Exchange :消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
3. Queue :消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。
4. Producer:消息生产者，即生产方客户端，生产方客户端将消费发送到MQ…
5. Consumer:消息消费者，即消费方客户端，接收MQ转发的消息。

消息发布接收流程∶
-----发送消息-----
6. 生产者和Broker建立TCP连接。
7. 生产者和Broker建立通道。
8. 生产者通过通道消息发送Broker，由Exchange将消息进行转发。
9. Exchange将消息转发到指定的Queue(队列)

linux安装RabbitMQ

docker拉取RabbitMQ镜像

docker pull rabbitmq:3-management

查看RabbitMQ镜像

docker images

运行RabbitMQ

docker run -d -p 5672:5672 -p 15672:15672 --name myrabbitmq imageid

其中:
-d:代表后台运行
-p 5672:5672:代表将主机端口号5672映射到docker的5672端口
-p 15672:15672:代表将主机端口号15672映射到docker的15672端口
–name myrabbitmq:代表启动的rabbitmq的名字是myrabbitmq
imageid:代表代表docker镜像的镜像id的值.

查看docker进程

docker ps

使用本地浏览器访问

地址：localhost（ip）:15672 默认账号：guest 密码：guest

springboot整合RabbitMQ

官方文档：https://www.rabbitmq.com/getstarted.html

pom.xml文件

<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-amqpartifactId>
dependency>
<dependency>
    <groupId>org.springframework.amqpgroupId>
    <artifactId>spring-rabbit-testartifactId>
    <scope>testscope>
dependency>
1
2
3
4
5
6
7
8
9

application.properties

spring.rabbitmq.host=66.112.211.169
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
1
2
3
4
5

配置类

package com.etc.demomq1;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 创建一个Queue
 */
@Configuration
public class DemoQueue {

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

发送消息

package com.etc.demomq1;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

@Component
public class MessageSender {
    @Autowired
    private AmqpTemplate amqpTemplate;

    @Autowired
    private Queue queue;

//    @Autowired
 private RabbitTemplate template;


    /**
     * 发送消息的方法
     * @param username
     */
    public void sendMessage(String username){
        String msg = username + ": "+ LocalDateTime.now();
        amqpTemplate.convertAndSend(queue.getName(),msg);

    }


}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

接收消息

package com.etc.demomq1;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

/**
 * RabbitListener 监听的是hello的队列
 */
@Component
@RabbitListener(queues = "hello")
public class MessageReceiver {


    /**
     * 消息的处理
     * @param msg
     */
    @RabbitHandler
    public void getMessage(String msg){
        System.out.println("MessageReceiver :"+msg);
    }


}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

可以直接运行了