AI 助理
备案控制台
开发者社区 数据库 文章 正文

rabbitmq系列（二）几种常见模式的应用场景及实现

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

一、简单模式

原理：生产者将消息交给默认的交换机，交换机获取消息后交给绑定这个生产者的队列（投递规则为队列名称和routing key 相同的队列），监听当前队列的消费者获取信息并执行消费逻辑。

场景：有一个oa系统，用户通过接收手机验证码进行注册，页面上点击获取验证码后，将验证码放到消息队列，然后短信服务从队列中获取到验证码，并发送给用户。

实现：

生产者：

public class Producter {
 
    public static void main(String[] args) throws Exception {
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
        // 4. 通过channel发布消息
        /**
         * 四个参数：
         * 第一个参数是交换机的名称
         * 第二个参数是路由键
         * 第三个参数标识消息的一些额外的属性
         * 第四个是消息的具体的内容
         */
        String message = "字节";
        for(int i = 0;i < 5;i ++){
 
            channel.basicPublish("","byte001",null,message.getBytes());
        }
        // 5. 释放资源，释放channel 和 链接对象
 
        channel.close();
        connection.close();
    }
}

消费者：

public class Consumer {
 
    public static void main(String[] args) throws Exception {
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
 
        // 4. 创建出消息队列
        /**
         * 第一个参数是消息队列的名称
         * 第二个参数表示消息是否持久化
         * 第三个参数标识消息队列是否被channel独占
         * 第四个参数标识是否自动删除消息队列，当消息队列没有绑定交换机后是否自动删除
         * 第五个是消息队列扩展参数
         */
        String queueName = "byte001";
        channel.queueDeclare(queueName, true, false, false, null);
 
        // 5. 创建消费者,对消息进行处理
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                String s = new String(body, "UTF-8");
                System.out.println("获取到的消息："+s);
            }
        };
        // 6. 通过channel消费者和消息队列关联
        /**
         * 第一个参数是消息队列的名字
         * 第二个参数是否自动签收（即消费消息后告知服务器已被消费）
         * 第三个参数是消费者
         */
        channel.basicConsume(queueName, true, consumer);
    }
}

二、工作模式

原理：生产者将消息交给交换机，交换机交给绑定的队列，队列有多个消费者监听，一条消息只能由一个消费者消费，这样就形成了资源竞争，谁的资源空闲大，争抢到的可能性就大。

场景：有一个电商平台，有两个订单服务，用户下单的时候，任意一个订单服务消费用户的下单请求生成订单即可。不用两个订单服务同时消费用户的下单请求。

实现：

生产者：

public class Producter {
 
    public static final String QUEUE_NAME = "byte002";
    public static void main(String[] args) throws Exception {
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
 
        // 申明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        // 4. 通过channel发布消息
        /**
         * 四个参数：
         * 第一个参数是交换机的名称
         * 第二个参数是路由键
         * 第三个参数标识消息的一些额外的属性
         * 第四个是消息的具体的内容
         */
        String message = "字节";
        for(int i = 0;i < 100;i ++){
 
            channel.basicPublish("",QUEUE_NAME,null,(message+i).getBytes());
        }
        // 5. 释放资源，释放channel 和 链接对象
 
        channel.close();
        connection.close();
    }
}

消费者1：

public class Consumer {
 
    public static final String QUEUE_NAME = "byte002";
    public static void main(String[] args) throws Exception {
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
 
        // 4. 创建出消息队列
        /**
         * 第一个参数是消息队列的名称
         * 第二个参数表示消息是否持久化
         * 第三个参数标识消息队列是否被channel独占
         * 第四个参数标识是否自动删除消息队列，当消息队列没有绑定交换机后是否自动删除
         * 第五个是消息队列扩展参数
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicQos(1); // 告诉服务器，在我们没有确认当前消息时不要给我们发送新的消息
        // 5. 创建消费者,对消息进行处理
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                String s = new String(body, "UTF-8");
                System.out.println("消费者1收到的内容："+s);
                try {
                    Thread.sleep(10); // 模拟消费耗时
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(),false); // 参数2false为确认收到消息，true为拒绝收到消息
            }
        };
        // 6. 通过channel消费者和消息队列关联
        /**
         * 第一个参数是消息队列的名字
         * 第二个参数是否自动签收（即消费消息后告知服务器已被消费）
         * 第三个参数是消费者
         */
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

消费者2：

public class Consumer2 {
 
    public static final String QUEUE_NAME = "byte002";
    public static void main(String[] args) throws Exception {
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
 
        // 4. 创建出消息队列
        /**
         * 第一个参数是消息队列的名称
         * 第二个参数表示消息是否持久化
         * 第三个参数标识消息队列是否被channel独占
         * 第四个参数标识是否自动删除消息队列，当消息队列没有绑定交换机后是否自动删除
         * 第五个是消息队列扩展参数
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicQos(1); // 告诉服务器，在我们没有确认当前消息时不要给我们发送新的消息
        // 5. 创建消费者,对消息进行处理
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                String s = new String(body, "UTF-8");
                System.out.println("消费者2收到的内容："+s);
                try {
                    Thread.sleep(500); // 模拟消费耗时
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(),false); // 参数2false为确认收到消息，true为拒绝收到消息
            }
        };
        // 6. 通过channel消费者和消息队列关联
        /**
         * 第一个参数是消息队列的名字
         * 第二个参数是否自动签收（即消费消息后告知服务器已被消费）
         * 第三个参数是消费者
         */
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

保证资源竞争的代码就是这一行channel.basicQos(1);如果不加这一行，我们会发现两个消费者是轮询消费消息的。

三、发布订阅模式

原理：生产者将消息扔给交换机，交换机类型是fanout，不同的队列注册到交换机上，不同的消费注册在不同的队列上。所有消费者都会收到消息。

场景：有一个商城，我们新添加一个商品后，可能同时需要去更新缓存和数据库。

实现：

生产者：

public class Producter {
 
    // 定义交换机的名字
    public static final String EXCHANGE_NAME="byte003";
    public static void main(String[] args) throws Exception {
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
        // 定义一个交换机，类型是fanout
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
 
        // 因为消息先发到交换机，交换机没有保存功能，所以如果没有消费者，消息会丢失
        channel.basicPublish(EXCHANGE_NAME,"",null,"发布订阅模式的消息".getBytes());
 
        channel.close();
        connection.close();
    }
}

消费者1：

public class Consumer1 {
    public static final String EXCHANGE_NAME="byte003";
 
    public static void main(String[] args) throws Exception {
 
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
 
        String queueName = "queue003";
        channel.queueDeclare(queueName,false,false,false,null);
 
        // 绑定队列到交换机
        channel.queueBind(queueName,EXCHANGE_NAME,"");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                String s = new String(body, "UTF-8");
                System.out.println("消费者1："+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

消费者2：

public class Consumer2 {
    public static final String EXCHANGE_NAME="byte003";
 
    public static void main(String[] args) throws Exception {
 
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
 
        String queueName = "queue004";
        channel.queueDeclare(queueName,false,false,false,null);
 
        // 绑定队列到交换机
        channel.queueBind(queueName,EXCHANGE_NAME,"");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                String s = new String(body, "UTF-8");
                System.out.println("消费者2："+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

需要注意的一点就是交换机没有保存功能，如果没有消费者，则消息会丢失。

四、路由模式

原理：生产者将消息发送给交换机，消息携带具体的routingkey。交换机类型是direct，接收到消息中的routingkey，比对与之绑定的队列的routingkey，分发到不同的队列上。

场景：还是一样，有一个商城，新添加了一个商品，实时性不是很高，只需要添加到数据库即可，不用刷新缓存。

实现：

生产者：

public class Producter {
 
    // 定义交换机的名字
    public static final String EXCHANGE_NAME="byte004";
    public static void main(String[] args) throws Exception {
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
        // 定义一个交换机，类型是direct
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");
 
        // 因为消息先发到交换机，交换机没有保存功能，所以如果没有消费者，消息会丢失
        channel.basicPublish(EXCHANGE_NAME,"key1",null,"发布路由模式的消息".getBytes());
 
        channel.close();
        connection.close();
    }
}

消费者1：

public class Consumer1 {
    public static final String EXCHANGE_NAME="byte004";
 
    public static void main(String[] args) throws Exception {
 
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
 
        String queueName = "queue005";
        channel.queueDeclare(queueName,false,false,false,null);
 
        // 绑定队列到交换机
        /**
         * 参数3是routingkey，只有和它一样的routingkey的消息才会被当前消费者收到
         */
        channel.queueBind(queueName,EXCHANGE_NAME,"key1");
        // 如果要接收多个routingkey的消息，在执行一次上面的代码即可，如下
        channel.queueBind(queueName,EXCHANGE_NAME,"key2");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                String s = new String(body, "UTF-8");
                System.out.println("消费者1："+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

消费者2：

public class Consumer2 {
    public static final String EXCHANGE_NAME="byte004";
 
    public static void main(String[] args) throws Exception {
 
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
 
        String queueName = "queue006";
        channel.queueDeclare(queueName,false,false,false,null);
 
        // 绑定队列到交换机
        channel.queueBind(queueName,EXCHANGE_NAME,"key2");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                String s = new String(body, "UTF-8");
                System.out.println("消费者2："+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

五、主题模式

原理：路由模式的一种，路由功能添加了模糊匹配。星号（*）代表1个单词，#号（#）代表一个或多个单词。具体可参考路由模式。

场景：还是一样，有一个商城，新添加了一个商品，实时性不是很高，只需要添加到数据库即可，数据库包含了主数据库mysql1和从数据库mysql2的内容，不用刷新缓存。

实现：

生产者：

public class Producter {
 
    // 定义交换机的名字
    public static final String EXCHANGE_NAME="byte004";
    public static void main(String[] args) throws Exception {
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
        // 定义一个交换机，类型是topic
        channel.exchangeDeclare(EXCHANGE_NAME,"topic");
 
        // 因为消息先发到交换机，交换机没有保存功能，所以如果没有消费者，消息会丢失
        channel.basicPublish(EXCHANGE_NAME,"key.1.2",null,"发布路由模式的消息".getBytes());
 
        channel.close();
        connection.close();
    }
}

消费者1：

public class Consumer1 {
    public static final String EXCHANGE_NAME="byte004";
 
    public static void main(String[] args) throws Exception {
 
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
 
        String queueName = "queue005";
        channel.queueDeclare(queueName,false,false,false,null);
 
        // 绑定队列到交换机
        /**
         * 参数3是routingkey，只有和它一样的routingkey的消息才会被当前消费者收到
         */
        channel.queueBind(queueName,EXCHANGE_NAME,"key.*");
        // 如果要接收多个routingkey的消息，在执行一次上面的代码即可，如下
        channel.queueBind(queueName,EXCHANGE_NAME,"abc.#");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                String s = new String(body, "UTF-8");
                System.out.println("消费者1："+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

消费者2：

public class Consumer2 {
    public static final String EXCHANGE_NAME="byte004";
 
    public static void main(String[] args) throws Exception {
 
 
        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
 
        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();
 
        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
 
        String queueName = "queue006";
        channel.queueDeclare(queueName,false,false,false,null);
 
        // 绑定队列到交换机
        channel.queueBind(queueName,EXCHANGE_NAME,"key.#");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                String s = new String(body, "UTF-8");
                System.out.println("消费者2："+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

代码已上传：

github地址： https://github.com/binzh303/zhixie-code-example

gitee地址： https://gitee.com/javaXiaoCaiJi/zhixie-code-example

文章标签：
云消息队列 MQ
消息中间件
数据库
缓存
关键词：
云消息队列 MQ模式
云消息队列 MQ应用
云消息队列 MQ模式应用场景
云消息队列 MQ应用场景
相关实践学习
消息队列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
weixin_836869520
目录
相关文章
三分钟热度的鱼
|
2月前
|
消息中间件 Java RocketMQ
消息队列 MQ产品使用合集之当SpringBoot应用因网络不通而启动失败时，该如何解决
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
79 0
秋夜雨巷
|
4天前
|
消息中间件
RabbitMQ广播模式
RabbitMQ广播模式
秋夜雨巷
13 1
三分钟热度的鱼
|
21天前
|
消息中间件 传感器 负载均衡
消息队列 MQ使用问题之如何配置一主一从的同步复制模式
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
21 0
消息队列 MQ使用问题之如何配置一主一从的同步复制模式
萝卜丝丸子
|
26天前
|
消息中间件 存储 Kafka
MetaQ/RocketMQ 原理问题之RocketMQ DLedger融合模式的问题如何解决
MetaQ/RocketMQ 原理问题之RocketMQ DLedger融合模式的问题如何解决
萝卜丝丸子
32 2
三分钟热度的鱼
|
21天前
|
消息中间件 Java Apache
消息队列 MQ使用问题之如何在内外网环境下使用单组节点单副本模式
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
31 0
萝卜丝丸子
|
26天前
|
消息中间件 负载均衡 RocketMQ
MetaQ/RocketMQ 原理问题之在广播模式下，RebalanceService工作的问题如何解决
MetaQ/RocketMQ 原理问题之在广播模式下，RebalanceService工作的问题如何解决
萝卜丝丸子
14 0
三分钟热度的鱼
|
2月前
|
消息中间件 Arthas 监控
消息队列 MQ产品使用合集之每次重置reconsumeTimes就无法达到死信阈值，重试次数是否就要应用方控制
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
28 0
消息队列 MQ产品使用合集之每次重置reconsumeTimes就无法达到死信阈值，重试次数是否就要应用方控制
Jack20
|
2月前
|
数据采集 监控 物联网
MQTT协议在智能制造中的应用案例与效益分析
【6月更文挑战第8天】MQTT协议在智能制造中的应用案例与效益分析
Jack20
80 1
泥腿子架构师
|
2月前
|
消息中间件 Java Spring
Spring Boot与RabbitMQ的集成应用
Spring Boot与RabbitMQ的集成应用
泥腿子架构师
27 0
米诺斯
|
2月前
|
消息中间件
RabbitMQ配置单活模式队列
RabbitMQ配置单活模式队列
米诺斯
36 0

热门文章

最新文章

  • 1
    信用算力基于 RocketMQ 实现金融级数据服务的实践
  • 2
    springboot+redis+rabbitmq实现模拟秒杀系统(附带docker安装mysql，rabbitmq，redis教程)
  • 3
    WebAPi的可视化输出模式(RabbitMQ、消息补偿相关)所有webapi似乎都缺失的一个功能
  • 4
    RabbitMQ 消费端限流、TTL、死信队列
  • 5
    python rabbitmq send messages
  • 6
    MQ是如何保证消息不丢失的,你这样回答面试官一定说I Want U
  • 7
    RocketMQ 5.0: 存储计算分离新思路
  • 8
    RocketMQ在面试中那些常见问题及答案+汇总
  • 9
    RabbitMQ入门指南：初学者也能读懂的教程（一）
  • 10
    rabbit.bat server启动闪退，rabbitmq (ArgumentError) argument error，rabbitmq页面无法开启解决方案
  • 1
    Spring Cloud RocketMQ：构建可靠消息驱动的微服务架构
    139
  • 2
    RabbitMQ组件介绍
    53
  • 3
    SpringBoot基于RabbitMQ实现消息可靠性
    69
  • 4
    SpringBoot实现RabbitMQ的通配符交换机(SpringAMQP 实现Topic交换机)
    28
  • 5
    MQTT：通信协议的三位主角
    184
  • 6
    SpringBoot实现RabbitMQ的定向交换机(SpringAMQP 实现Direct定向交换机)
    44
  • 7
    SpringBoot实现RabbitMQ的广播交换机(SpringAMQP 实现Fanout广播交换机)
    37
  • 8
    SpringBoot实现RabbitMQ的WorkQueue(SpringAMQP 实现WorkQueue)
    42
  • 9
    Docker安装部署RabbitMQ(单机版)
    197
  • 10
    RabbitMQ使用教程
    74

    • 相关课程

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

    • 相关电子书

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

    • 相关实验场景

    更多
  • 快速学会使用云消息队列RabbitMQ版
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 部署RabbitMQ
  • RocketMQ一站式入门使用
    • 下一篇
    通义千问API入门教程