AI 助理
备案控制台
开发者社区 Java技术栈 文章 正文

RabbitMQ 中的 7 种队列模式，写得太好了！

2021-12-20 89
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 七种模式介绍与应用场景简单模式（Hello World）

image.png

image.png

image.png

代码演示

代码中没有对后面两种模式演示，有兴趣可以自己研究

简单模式

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Sender {
    private final static String QUEUE_NAME = "simple_queue";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        // queue：队列名
        // durable：是否持久化
        // exclusive：是否排外 即只允许该channel访问该队列 一般等于true的话用于一个队列只能有一个消费者来消费的场景
        // autoDelete：是否自动删除 消费完删除
        // arguments：其他属性
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //消息内容
        String message = "simplest mode message";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println("[x]Sent '" + message + "'");
        //最后关闭通关和连接
        channel.close();
        connection.close();
    }
}
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receiver {
    private final static String QUEUE_NAME = "simplest_queue";
    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        // 获取连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" +
                    delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {
        });
    }
}

工作队列模式

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receiver1 {
    private final static String QUEUE_NAME = "queue_work";
    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 同一时刻服务器只会发送一条消息给消费者
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" +
                    delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {
        });
    }
}
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receiver2 {
    private final static String QUEUE_NAME = "queue_work";
    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 同一时刻服务器只会发送一条消息给消费者
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" +
                    delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {
        });
    }
}
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Sender {
    private final static String QUEUE_NAME = "queue_work";
    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        for (int i = 0; i < 100; i++) {
            String message = "work mode message" + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("[x] Sent '" + message + "'");
            Thread.sleep(i * 10);
        }
        channel.close();
        connection.close();
    }
}

发布订阅模式

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
public class Receive1 {
    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        // 订阅消息的回调函数
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
        };
        // 消费者，有消息时出发订阅回调函数
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
        });
    }
}
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
public class Receive2 {
    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        // 订阅消息的回调函数
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received2 '" + message + "'");
        };
        // 消费者，有消息时出发订阅回调函数
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
        });
    }
}
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Sender {
    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        String message = "publish subscribe message";
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + message + "'");
        channel.close();
        connection.close();
    }
}

路由模式

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receiver1 {
    private final static String QUEUE_NAME = "queue_routing";
    private final static String EXCHANGE_NAME = "exchange_direct";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 指定路由的key，接收key和key2
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key2");
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" +
                    delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {
        });
    }
}
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receiver2 {
    private final static String QUEUE_NAME = "queue_routing2";
    private final static String EXCHANGE_NAME = "exchange_direct";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 仅接收key2
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key2");
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" +
                    delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {
        });
    }
}
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Sender {
    private final static String EXCHANGE_NAME = "exchange_direct";
    private final static String EXCHANGE_TYPE = "direct";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 交换机声明
        channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);
        // 只有routingKey相同的才会消费
        String message = "routing mode message";
        channel.basicPublish(EXCHANGE_NAME, "key2", null, message.getBytes());
        System.out.println("[x] Sent '" + message + "'");
// channel.basicPublish(EXCHANGE_NAME, "key", null, message.getBytes());
// System.out.println("[x] Sent '" + message + "'");
        channel.close();
        connection.close();
    }
}

主题模式

public class Receiver1 {
    private final static String QUEUE_NAME = "queue_topic";
    private final static String EXCHANGE_NAME = "exchange_topic";
    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 可以接收key.1
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key.*");
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" +
                    delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {
        });
    }
}
public class Receiver1 {
    private final static String QUEUE_NAME = "queue_topic";
    private final static String EXCHANGE_NAME = "exchange_topic";
    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 可以接收key.1
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key.*");
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" +
                    delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {
        });
    }
}
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Sender {
    private final static String EXCHANGE_NAME = "exchange_topic";
    private final static String EXCHANGE_TYPE = "topic";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);
        String message = "topics model message with key.1";
        channel.basicPublish(EXCHANGE_NAME, "key.1", null, message.getBytes());
        System.out.println("[x] Sent '" + message + "'");
        String message2 = "topics model message with key.1.2";
        channel.basicPublish(EXCHANGE_NAME, "key.1.2", null, message2.getBytes());
        System.out.println("[x] Sent '" + message2 + "'");
        channel.close();
        connection.close();
    }
}

四种交换机介绍

1、直连交换机（Direct exchange）：


具有路由功能的交换机，绑定到此交换机的时候需要指定一个routing_key，交换机发送消息的时候需要routing_key，会将消息发送道对应的队列


2、扇形交换机（Fanout exchange）：


广播消息到所有队列，没有任何处理，速度最快


3、主题交换机（Topic exchange）：


在直连交换机基础上增加模式匹配，也就是对routing_key进行模式匹配，*代表一个单词，#代表多个单词


4、首部交换机（Headers exchange）：


忽略routing_key，使用Headers信息（一个Hash的数据结构）进行匹配，优势在于可以有更多更灵活的匹配规则


总结

这么多种队列模式中都有其应用场景，大家可以根据应用场景示例中进行选择。


原文链接：https://blog.csdn.net/qq_32828253/article/details/110450249


版权声明：本文为CSDN博主「我思知我在」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。

文章标签：
云消息队列 MQ
消息中间件
关键词：
云消息队列 MQ模式
云消息队列 MQ队列
消息队列RabbitMQ版队列
云消息队列 MQ队列模式
消息队列RabbitMQ版模式
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
技术君
目录
相关文章
小王老师呀
|
4月前
|
消息中间件
RabbitMQ的 RPC 消息模式你会了吗？
【9月更文挑战第11天】RabbitMQ 的 RPC（远程过程调用）消息模式允许客户端向服务器发送请求并接收响应。其基本原理包括：1) 客户端发送请求，创建回调队列并设置关联标识符；2) 服务器接收请求并发送响应至回调队列；3) 客户端根据关联标识符接收并匹配响应。实现步骤涵盖客户端和服务器的连接、信道创建及请求处理。注意事项包括关联标识符唯一性、回调队列管理、错误处理及性能考虑。RPC 模式适用于构建可靠的分布式应用程序，但需根据需求调整优化。
小王老师呀
71 2
遇见明天
|
8天前
|
消息中间件 网络协议 RocketMQ
RocketMQ Controller 模式 始终更新成本机ip
ontrollerAddr=192.168.24.241:8878 但是日志输出Update controller leader address to 127.0.0.1:8878。导致访问失败
遇见明天
40 3
何雨晨
|
3月前
|
消息中间件 存储 监控
RabbitMQ 队列之战：Classic 和 Quorum 的性能洞察
RabbitMQ 是一个功能强大的消息代理，用于分布式应用程序间的通信。它通过队列临时存储消息，支持异步通信和解耦。经典队列适合高吞吐量和低延迟场景，而仲裁队列则提供高可用性和容错能力，适用于关键任务系统。选择哪种队列取决于性能、持久性和容错性的需求。
何雨晨
231 6
程序员小海绵
|
4月前
|
消息中间件 JSON Java
谷粒商城笔记+踩坑（22）——库存自动解锁。RabbitMQ延迟队列
库存自动解锁。RabbitMQ延迟队列
程序员小海绵
107 0
谷粒商城笔记+踩坑（22）——库存自动解锁。RabbitMQ延迟队列
游客j4mujezz7vm7y
|
4月前
|
消息中间件
rabbitmq,&队列
rabbitmq,&队列
游客j4mujezz7vm7y
46 0
土木林森
|
5月前
|
消息中间件 存储 NoSQL
MQ的顺序性保证：顺序队列、消息编号、分布式锁，一文全掌握！
【8月更文挑战第24天】消息队列（MQ）是分布式系统的关键组件，用于实现系统解耦、提升可扩展性和可用性。保证消息顺序性是其重要挑战之一。本文介绍三种常用策略：顺序队列、消息编号与分布式锁，通过示例展示如何确保消息按需排序。这些方法各有优势，可根据实际场景灵活选用。提供的Java示例有助于加深理解与实践应用。
土木林森
151 2
土木林森
|
5月前
|
消息中间件 开发者
【RabbitMQ深度解析】Topic交换器与模式匹配：掌握消息路由的艺术！
【8月更文挑战第24天】在消息队列（MQ）体系中，交换器作为核心组件之一负责消息路由。特别是`topic`类型的交换器，它通过模式匹配实现消息的精准分发，适用于发布-订阅模式。不同于直接交换器和扇形交换器，`topic`交换器支持更复杂的路由策略，通过带有通配符（如 * 和 #）的模式字符串来定义队列与交换器间的绑定关系。
土木林森
87 2
小宋1021
|
4月前
|
消息中间件 JSON Java
玩转RabbitMQ声明队列交换机、消息转换器
玩转RabbitMQ声明队列交换机、消息转换器
小宋1021
107 0
秋夜雨巷
|
5月前
|
消息中间件
RabbitMQ广播模式
RabbitMQ广播模式
秋夜雨巷
88 1
1288912195458132
|
5月前
|
消息中间件 应用服务中间件 网络安全
rabbitMQ镜像模式搭建
rabbitMQ镜像模式搭建
1288912195458132
148 0

Java技术栈

热门文章

最新文章

  • 1
    大厂都在用的分布式事务方案，Seata+RocketMQ带你打破10万QPS瓶颈
  • 2
    C5GAME 游戏饰品交易平台借助 RocketMQ Serverless 保障千万级玩家流畅体验
  • 3
    朗新科技集团如何用云消息队列 RocketMQ 版“快、准、狠”破解业务难题？
  • 4
    2024最全RabbitMQ集群方案汇总
  • 5
    说说MQ在你项目中的应用(一)
  • 6
    高弹性、低成本的云消息队列RabbitMQ 版
  • 7
    2024最全RocketMQ集群方案汇总
  • 8
    如何将Spring Boot + RabbitMQ应用程序部署到Pivotal Cloud Foundry (PCF)
  • 9
    多协议网关BL110钡铼6路RS485转MQTT协议云网关
  • 10
    MQ四兄弟：如何实现延时消息
  • 1
    【java常见的面试题】RabbitMQ如何设置消息过期 ?
    250
  • 2
    【java常见的面试题】使用RabbitMQ如何保证消息不丢失 ?
    158
  • 3
    【java常见的面试题】为什么会选择使用RabbitMQ ? 有什么好处 ?
    250
  • 4
    【java常见的面试题】项目中哪里用到了RabbitMQ ?
    121
  • 5
    RabbitMQ延时队列插件rabbitmq_delayed_message_exchange-3.8.0.ez，有需要的朋友可自取
    194
  • 6
    RabbitMQ入门指南：初学者也能读懂的教程（五）
    151
  • 7
    RabbitMQ入门指南：初学者也能读懂的教程（四）
    111
  • 8
    RabbitMQ入门指南：初学者也能读懂的教程（三）
    123
  • 9
    RabbitMQ入门指南：初学者也能读懂的教程（二）
    157
  • 10
    RabbitMQ入门指南：初学者也能读懂的教程（一）
    1550

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台

    • 相关实验场景

    更多
  • 部署RabbitMQ
  • RocketMQ监控/告警一站式搭建应用
  • RocketMQ运维自我实践
  • RocketMQ的常规运维实践应用
  • RocketMQ中使用Java客户端发送消息和消费的应用
  • RocketMQ一站式入门使用
    • 下一篇
    阿里云无影云电脑免费试用，最长可试用3个月