RabbitMQ 入门-阿里云开发者社区

一：什么是MQ

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

ActiveMQ
RabbitMQ
RocketMQ
Kafka

几种常见MQ的对比：

	RABBITMQ	ACTIVEMQ	ROCKETMQ	KAFKA
公司/社区	Rabbit	Apache	阿里	Apache
开发语言	Erlang(并发能力强，性能及其好)	Java	Java	Scala&Java

协议支持	AMQP，XMPP，SMTP，STOMP	OpenWire,STOMP，REST,XMPP,AMQP	自定义协议	自定义协议
可用性	高	一般	高	高
单机吞吐量(性能承载能力)	一般	差	高	非常高

消息延迟	微秒级	毫秒级	毫秒级	毫秒以内
消息可靠性	高	一般	高	一般

追求可用性(当需要处理数据时，资源处于可用状态的程度)：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力(十万级别的)：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

二：安装RabbitMQ

我们将RabbitMQ安装在linux上使用以下两种方式可以安装（安装过程保证自己的docker容器一直是开着的）

第一种方法使用docker镜像，直接拉取安装即可

命令：docker pull rabbitmq:3.8-management

第二种方法使用本地下载

RabbitMQ的镜像包我已上传在我的个人博客资源中，将镜像包拖到tmp目录下

使用命令

//在线解压即可

docker load -i mq.tar

如下：

解压好后，接着输入以下命令

docker run \

# 设置环境变量用户名是 dlwlrma

-e RABBITMQ_DEFAULT_USER=itcast \

# 设置环境变量密码是 zhien0516

-e RABBITMQ_DEFAULT_PASS=123321 \

# 挂载数据卷，后面高级会用到下面的插件

-v mq-plugins:/plugins \

# mq的名字

--name mq \

# 主机名这里不配置也可以，后期如果是集群必须配置

--hostname mq \

# web可视化终端监控端口；mq的ui界面管理平台端口

-p 15672:15672 \

# 程序与mq交互的访问端口；发消息和收消息的端口

-p 5672:5672 \

# 后端运行

-d \

# 镜像名称

rabbitmq:3.8-management

精简命令如下：

docker run \

-e RABBITMQ_DEFAULT_USER=dlwlrma\

-e RABBITMQ_DEFAULT_PASS=zhien0516 \

-v mq-plugins:/plugins \

--name mq \

--hostname mq \

-p 15672:15672 \

-p 5672:5672 \

-d \

rabbitmq:3.8-management

安装好后，可以在浏览器中输入自己的linux的ip地址访问即可出现以下画面，输入自己刚刚设置的用户名和密码即可成功登录

三：在Java中如何实现MQ的使用

RabbitMQ的五种消息模型

1.基本消息队列（BasicQueue）

P（producer/ publisher）：生产者，一个发送消息的用户应用程序。我们自己书写代码发送。

C（consumer）：消费者，消费和接收有类似的意思，消费者是一个主要用来等待接收消息的用户应用程序。我们自己书写代码接收。

队列（红色区域）：存在于rabbitmq内部。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。

总之：生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。

2.工作消息队列（WorkQueue）

工作消息队列是基本消息队列的增强版，具有多个消费者消费队列的消息。假设消息队列中积压了多个消息，那么此时可以使用多个消费者来消费队列中的消息。效率要比基本消息队列模型高。

3. 发布订阅（Publish、Subscribe）

又根据交换机类型不同分为三种：

但是三种类型都有一个共性：

1、1个生产者，多个消费者

2、每一个消费者都有自己的一个队列

3、生产者没有将消息直接发送到队列，而是发送到了交换机

4、每个队列都要绑定到交换机

5、生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的

X（Exchanges）：交换机一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失

广播类型：

将消息交给所有绑定到交换机的队列,生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。交换机把消息发送给绑定过的所有队列.队列的消费者都能拿到消息。实现一条消息被多个消费者消费.

4.Direct Exchange：路由

1.在广播模式中，生产者发布消息，所有消费者都可以获取所有消息。

2.在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）.消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。

3.P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。

4.X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列

5.Topic Exchange：主题

1.Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

2.Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

3.通配符规则：

#：匹配一个或多个词

*：匹配恰好1个词

在Java中实现五种模型

1.基本消息队列

消费者：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
 
//将当前类放到SpringIOC容器中
@Component
public class SpringRabbitListener {
    /**
     *
     * @author dlwlrma
     * @date 2024/6/17 20:18
     * @param msg
     * TODO: @RabbitListener 为定义监听哪一个队列
     */
    @RabbitListener(queues = {"simple.queue"})
    //Spring自动将接收的消息给方法参数msg
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("spring 消费者接收到消息：【" + msg + "】");
    }
}

发布者：

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
 
 
//TODO:是一个测试启动器，可以加载SpringBoot测试注解,让测试方法在Spring容器环境下执行
@RunWith(SpringRunner.class)
//TODO:目的是加载ApplicationContext，启动spring容器
@SpringBootTest
public class SpringAmqpTest {
    //自动装配RabbitTemplate模板对象
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    /**
     *
     * @author dlwlrma
     * @date 2024/6/17 20:23
     * TODO： Basic Queue 简单队列模型
     */
    @Test
    public void testSimpleQueue() {
        //1.定义变量保存队列的名字
        String queueName = "simple.queue";
        //2.定义变量保存要发送的消息
        String message = "hello啊，我是mq,我发送的是简单队列模型";
        //3.发送消息
        rabbitTemplate.convertAndSend(queueName,message);
    }
}

2.工作消息队列

消费者：

  @RabbitListener(queues = {"simple.queue"})
    public  void listenWorkQueue1Message(String msg) throws InterruptedException {
        System.out.println("用户1已收到了消息："+msg);
        //休眠20毫秒
        Thread.sleep(20);
    }
 
    @RabbitListener(queues = {"simple.queue"})
    public  void listenWorkQueue2Message(String msg) throws InterruptedException {
        System.out.println("用户2已收到了消息："+msg);
        //休眠100毫秒
        Thread.sleep(100);
    }

发布者:

   /**
    *
    * @author dlwlrma
    * @date 2024/6/17 20:24
    * TODO：workQueue 向队列中不停发送消息，模拟消息堆积。
    */
    @Test
    public void testWorkQueue() throws InterruptedException {
        //1.定义变量保存队列的名字
        String queueName = "simple.queue";
        //2.定义变量保存要发送的消息
        String message = "我是mq,我发送的是hello_message_";
        //3.发送消息
        for (int i = 1; i <=50 ; i++) {
            rabbitTemplate.convertAndSend(queueName,message+i);
            //4.模拟每隔20毫秒发送一次
            Thread.sleep(20);
        }
    }

3.发布订阅模型--广播

定义绑定类：将交换机与多个队列绑定起来

 
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
 
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class FanoutConfig {
 
    //1.定义交换机的名字
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("dlwlrma.fanout");
    }
 
    //2.定义队列1
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
 
    //3.将交换机与队列绑定起来
    @Bean
    public Binding bindingExchangeToFanoutQueue1(FanoutExchange fanoutExchange,Queue fanoutQueue1){
        //将队列1与交换机绑定起来
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }
 
    //4.定义队列2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
 
    //3.将交换机与队列绑定起来
    @Bean
    public Binding bindingExchangeToFanoutQueue2(FanoutExchange fanoutExchange,Queue fanoutQueue2){
        //将队列2与交换机绑定起来
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
 
}

消费者：

    /**
     *
     * @author dlwlrma
     * @date 2024/6/17 22:14
     * @param msg
     * TODO:模拟订阅发布：广播类型
     */
    @RabbitListener(queues = {"fanout.queue1"})
    public  void listenFAnoutQueue1Message(String msg) {
        System.out.println("用户1已收到了消息："+msg);
    }
 
    @RabbitListener(queues = {"fanout.queue2"})
    public  void listenFAnoutQueue2Message(String msg) {
        System.out.println("用户2已收到了消息："+msg);
    }

发布者：

 
    /**
     *
     * @author dlwlrma
     * @date 2024/6/17 20:24
     * TODO： Fanout Exchange：广播
     */
    @Test
    public void testFanoutExchange() {
        //1.定义变量保存交换机名字
        String exchangeName = "dlwlrma.fanout";
        //2.定义变量保存要发送的信息
        String message = "hello，我是订阅发布";
        //3.发送消息
        //第二个参数为广播类型
        rabbitTemplate.convertAndSend(exchangeName,"",message);
    }

4.发布订阅模型---路由

用基于注解的方式定义交换机和队列在consumer的SpringRabbitListener中添加两个消费者

 /**
    *
    * @author dlwlrma
    * @date 2024/6/18 23:42
    * @param msg 
    * TODO:  1. value = @Queue(name = "direct.queue1") 表示绑定的第一个队列
    *             2.exchange = @Exchange(name = "dlwlrma.direct", type = ExchangeTypes.DIRECT) 表示交换机名和类型
    *             3.key = {"red", "blue"} 表示路由key
    */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "dlwlrma.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "blue"}
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");
    }
 
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "dlwlrma.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg){
        System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");
    }

发布者：

 
    /**
     *
     * @author dlwlrma
     * @date 2024/6/17 20:24
     * TODO：Direct Exchange：路由
     */
    @Test
    public void testSendDirectExchange() {
        // 交换机名称
        String exchangeName = "dlwlrma.direct";
        // 消息
        String message = "红色警报！惊险大怪兽！";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "red", message);
    }

5.发布订阅模型----主题

消费者：

  /**
     *
     * @author dlwlrma
     * @date 2024/6/18 23:42
     * @param msg
     * TODO:  1.value = @Queue(name = "topic.queue1") 表示绑定的第一个队列
     *             2.exchange = @Exchange(name = "dlwlrma.topic", type = ExchangeTypes.TOPIC) 表示交换机名和类型
     *             3. key = "china.#" 表示路由key只要以china开始都会接收
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "dlwlrma.topic", type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void listenTopicQueue1(String msg) {
        System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");
    }
 
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "dlwlrma.topic", type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void listenTopicQueue2(String msg) {
        System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");
    }

发布者：

  /**
     *
     * @author dlwlrma
     * @date 2024/6/17 20:25
     * Topic Exchange：主题
     */
    @Test
    public void testSendTopicExchange() {
        // 交换机名称
        String exchangeName = "dlwlrma.topic";
        // 消息
        String message = "喜报！孙悟空大战哥斯拉，胜!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
    }

RabbitMQ 入门

一：什么是MQ

二：安装RabbitMQ

三：在Java中如何实现MQ的使用

RabbitMQ的五种消息模型

1.基本消息队列（BasicQueue）

2.工作消息队列（WorkQueue）

3. 发布订阅（Publish、Subscribe）

4.Direct Exchange：路由

5.Topic Exchange：主题

在Java中实现五种模型

1.基本消息队列

2.工作消息队列

3.发布订阅模型--广播

4.发布订阅模型---路由

5.发布订阅模型----主题

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

RabbitMQ 入门

一：什么是MQ

二：安装RabbitMQ

三：在Java中如何实现MQ的使用

RabbitMQ的五种消息模型

1.基本消息队列（BasicQueue）

2.工作消息队列（WorkQueue）

3. 发布订阅（Publish、Subscribe）

4.Direct Exchange：路由

5.Topic Exchange：主题

在Java中实现五种模型

1.基本消息队列

2.工作消息队列

3.发布订阅模型--广播

4.发布订阅模型---路由

5.发布订阅模型----主题

热门文章

最新文章

相关课程

相关电子书

相关实验场景