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.发布订阅模型----主题


一:什么是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);
    }

相关推荐

  1. RabbitMQ入门

    2024-06-19 09:40:03       38 阅读

最近更新

  1. docker php8.1+nginx base 镜像 dockerfile 配置

    2024-06-19 09:40:03       3 阅读
  2. Could not load dynamic library ‘cudart64_100.dll‘

    2024-06-19 09:40:03       3 阅读
  3. 在Django里面运行非项目文件

    2024-06-19 09:40:03       2 阅读
  4. Python语言-面向对象

    2024-06-19 09:40:03       2 阅读

热门阅读

  1. python 数据清洗基础教程

    2024-06-19 09:40:03       16 阅读
  2. MongoDB基础知识

    2024-06-19 09:40:03       15 阅读
  3. 十三、数论基础

    2024-06-19 09:40:03       14 阅读
  4. Ruby 数据库访问 - DBI 教程

    2024-06-19 09:40:03       18 阅读
  5. 安卓交叉编译——ndk

    2024-06-19 09:40:03       18 阅读
  6. Swarm 集群管理

    2024-06-19 09:40:03       16 阅读
  7. PostgreSQL源码分析——创建用户

    2024-06-19 09:40:03       16 阅读
  8. Linux 上的 TTY 是什么?

    2024-06-19 09:40:03       16 阅读