zoukankan      html  css  js  c++  java
  • RabbitMQ的五种工作方式详细

    在了解之前得先有个RabbitMQ客户端.官网: https://www.rabbitmq.com/getstarted.html

    • connections:无论生产者还是消费者,都需要与RabbitMQ建立连接后才可以完成消息的生产和消费,在这里可以查看连接情况

    • channels[ˈtʃænlz]:通道,建立连接后,会形成通道,消息的投递获取依赖通道。

    • Exchanges:交换机,用来实现消息的路由

    • Queues:队列,即消息队列,消息存放在队列中,等待消费,消费后被移除队列。

    一.基本消息模型

    官方文档说明:

    RabbitMQ是一个消息的代理者(Message Broker):它接收消息并且传递消息。

    你可以认为它是一个邮局:当你投递邮件到一个邮箱,你很肯定邮递员会终究会将邮件递交给你的收件人。与此类似,RabbitMQ 可以是一个邮箱、邮局、同时还有邮递员。

    不同之处在于:RabbitMQ不是传递纸质邮件,而是二进制的数据

    基本消息模型图:

    在上图的模型中,有以下概念:

    • P:生产者,也就是要发送消息的程序

    • C:消费者:消息的接受者,会一直等待消息到来。

    • queue:消息队列,图中红色部分。类似一个邮箱,可以缓存消息;生产者向其中投递消息,消费者从其中取出消息

    生产者

    连接工具类:

    public class ConnectionUtil {
        /**
         * 建立与RabbitMQ的连接
         * @return
         * @throws Exception
         */
        public static Connection getConnection() throws Exception {
            //定义连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            //设置服务地址
            factory.setHost("192.168.232.127");
            //端口
            factory.setPort(5672);
            //设置账号信息,用户名、密码、vhost
            factory.setVirtualHost("/leyou");
            factory.setUsername("leyou");
            factory.setPassword("leyou");
            // 通过工程获取连接
            Connection connection = factory.newConnection();
            return connection;
        }
    }
    

      

    application.yml

    spring:
      rabbitmq:
        host: 192.168.232.127
        username: leyou
        password: leyou
        virtual-host: /leyou
    

     

    生产者发送消息:

    public class Send {
    
        private final static String QUEUE_NAME = "simple_queue";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道,使用通道才能完成消息相关的操作
            Channel channel = connection.createChannel();
            // 声明(创建)队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 消息内容
            String message = "Hello World!";
            // 向指定的队列中发送消息
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            
            System.out.println(" [x] Sent '" + message + "'");
    
            //关闭通道和连接
            channel.close();
            connection.close();
        }
    }
    

      

    控制台:

    web控制台查看消息

    进入队列页面,可以看到新建了一个队列:simple_queue

     

    点击队列名称,进入详情页,可以查看消息:

    在控制台查看消息并不会将消息消费,所以消息还在。

    消费者获取消息

    public class Recv {
        private final static String QUEUE_NAME = "simple_queue";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 创建通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 定义队列的消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                        byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    System.out.println(" [x] received : " + msg + "!");
                }
            };
            // 监听队列,第二个参数:是否自动进行消息确认。
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }
    

      

    这个时候,队列中的消息就没了:

    消费者的消息确认机制

    通过刚才的案例可以看出,消息一旦被消费者接收,队列中的消息就会被删除。

    那么问题来了:RabbitMQ怎么知道消息被接收了呢?

    这就要通过消息确认机制(Acknowlege)来实现了。当消费者获取消息后,会向RabbitMQ发送回执ACK,告知消息已经被接收。不过这种回执ACK分两种情况:

    • 自动ACK:消息一旦被接收,消费者自动发送ACK

    • 手动ACK:消息接收后,不会发送ACK,需要手动调用

    大家觉得哪种更好呢?

    这需要看消息的重要性:

    • 如果消息不太重要,丢失也没有影响,那么自动ACK会比较方便

    • 如果消息非常重要,不容丢失。那么最好在消费完成后手动ACK,否则接收消息后就自动ACK,RabbitMQ就会把消息从队列中删除。如果此时消费者宕机,那么消息就丢失了。

    我们之前的测试都是自动ACK的,如果要手动ACK,需要改动我们的代码:

    public class Recv2 {
        private final static String QUEUE_NAME = "simple_queue";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 创建通道
            final Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 定义队列的消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                        byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    System.out.println(" [x] received : " + msg + "!");
                    // 手动进行ACK
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            // 监听队列,第二个参数false,手动进行ACK
            channel.basicConsume(QUEUE_NAME, false, consumer);
        }
    }
    

    注意到最后一行代码:

    channel.basicConsume(QUEUE_NAME, false, consumer);
    

    如果第二个参数为true,则会自动进行ACK;如果为false,则需要手动ACK。方法的声明:

     二. work消息模型

    说明

    在刚才的基本模型中,一个生产者,一个消费者,生产的消息直接被消费者消费。比较简单。

    Work queues,也被称为(Task queues),任务模型。

    当消息处理比较耗时的时候,可能生产消息的速度会远远大于消息的消费速度。长此以往,消息就会堆积越来越多,无法及时处理。此时就可以使用work 模型:让多个消费者绑定到一个队列,共同消费队列中的消息。队列中的消息一旦消费,就会消失,因此任务是不会被重复执行的。

     

    角色:

    • P:生产者:任务的发布者

    • C1:消费者,领取任务并且完成任务,假设完成速度较慢

    • C2:消费者2:领取任务并完成任务,假设完成速度快

    生产者

    生产者与案例1中的几乎一样:

    public class Send {
        private final static String QUEUE_NAME = "test_work_queue";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 获取通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 循环发布任务
            for (int i = 0; i < 50; i++) {
                // 消息内容
                String message = "task .. " + i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                System.out.println(" [x] Sent '" + message + "'");
    
                Thread.sleep(i * 2);
            }
            // 关闭通道和连接
            channel.close();
            connection.close();
        }
    }
    

    不过这里我们是循环发送50条消息。

    消费者1

    消费者2

    与消费者1基本类似,就是没有设置消费耗时时间。

    这里是模拟有些消费者快,有些比较慢。

    接下来,两个消费者一同启动,然后发送50条消息:

    能者多劳

    刚才的实现有问题吗?

    • 消费者1比消费者2的效率要低,一次任务的耗时较长

    • 然而两人最终消费的消息数量是一样的

    • 消费者2大量时间处于空闲状态,消费者1一直忙碌

    现在的状态属于是把任务平均分配,正确的做法应该是消费越快的人,消费的越多。

    怎么实现呢?

    我们可以修改设置,让消费者同一时间只接收一条消息,这样处理完成之前,就不会接收更多消息,就可以让处理快的人,接收更多消息 :

     

    再次测试:

     

     三. 订阅模型分类

     订阅模型示意图:

    前面2个案例中,只有3个角色:

    • P:生产者,也就是要发送消息的程序

    • C:消费者:消息的接受者,会一直等待消息到来。

    • queue:消息队列,图中红色部分。类似一个邮箱,可以缓存消息;生产者向其中投递消息,消费者从其中取出消息。

    而在订阅模型中,多了一个exchange角色,而且过程略有变化:

    • P:生产者,也就是要发送消息的程序,但是不再发送到队列中,而是发给X(交换机)

    • C:消费者,消息的接受者,会一直等待消息到来。

    • Queue:消息队列,接收消息、缓存消息。

    • Exchange:交换机,图中的X。一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。Exchange有以下3种类型:

      • Fanout:广播,将消息交给所有绑定到交换机的队列

      • Direct:定向,把消息交给符合指定routing key 的队列

      • Topic:通配符,把消息交给符合routing pattern(路由模式) 的队列

    Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!

    订阅模型共有三种:Fanout,Direct,Topic

    1.订阅模型-Fanout

    Fanout,也称为广播。

    流程说明

    流程图:

     

    在广播模式下,消息发送流程是这样的:

    • 1) 可以有多个消费者

    • 2) 每个消费者有自己的queue(队列)

    • 3) 每个队列都要绑定到Exchange(交换机)

    • 4) 生产者发送的消息,只能发送到交换机,交换机来决定要发给哪个队列,生产者无法决定。

    • 5) 交换机把消息发送给绑定过的所有队列

    • 6) 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

    生产者

    两个变化:

    • 1) 声明Exchange,不再声明Queue

    • 2) 发送消息到Exchange,不再发送到Queue

    public class Send {
    
        private final static String EXCHANGE_NAME = "fanout_exchange_test";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 获取通道
            Channel channel = connection.createChannel();
            
            // 声明exchange,指定类型为fanout
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            
            // 消息内容
            String message = "Hello everyone";
            // 发布消息到Exchange
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            System.out.println(" [生产者] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }
    

    消费者1

    public class Recv {
        private final static String QUEUE_NAME = "fanout_exchange_queue_1";
    
        private final static String EXCHANGE_NAME = "fanout_exchange_test";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 获取通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            // 定义队列的消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                        byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    System.out.println(" [消费者1] received : " + msg + "!");
                }
            };
            // 监听队列,自动返回完成
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }
    

    要注意代码中:队列需要和交换机绑定

    消费者2

    public class Recv2 {
        private final static String QUEUE_NAME = "fanout_exchange_queue_2";
    
        private final static String EXCHANGE_NAME = "fanout_exchange_test";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 获取通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
            
            // 定义队列的消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                        byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    System.out.println(" [消费者2] received : " + msg + "!");
                }
            };
            // 监听队列,手动返回完成
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }
    

      

    测试

    我们运行两个消费者,然后发送1条消息:

    2.订阅模型-Direct

    说明

    在Fanout模式中,一条消息,会被所有订阅的队列都消费。但是,在某些场景下,我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

    在Direct模型下:

    • 队列与交换机的绑定,不能是任意绑定了,而是要指定一个RoutingKey(路由key)

    • 消息的发送方在 向 Exchange发送消息时,也必须指定消息的 RoutingKey

    • Exchange不再把消息交给每一个绑定的队列,而是根据消息的Routing Key进行判断,只有队列的Routingkey与消息的 Routing key完全一致,才会接收到消息

    流程图:

    图解:

    • P:生产者,向Exchange发送消息,发送消息时,会指定一个routing key。

    • X:Exchange(交换机),接收生产者的消息,然后把消息递交给 与routing key完全匹配的队列

    • C1:消费者,其所在队列指定了需要routing key 为 error 的消息

    • C2:消费者,其所在队列指定了需要routing key 为 info、error、warning 的消息

    生产者

    此处我们模拟商品的增删改,发送消息的RoutingKey分别是:insert、update、delete

    public class Send {
        private final static String EXCHANGE_NAME = "direct_exchange_test";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 获取通道
            Channel channel = connection.createChannel();
            // 声明exchange,指定类型为direct
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
            // 消息内容
            String message = "商品新增了, id = 1001";
            // 发送消息,并且指定routing key 为:insert ,代表新增商品
            channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
            System.out.println(" [商品服务:] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }
    

    消费者1

    我们此处假设消费者1只接收两种类型的消息:更新商品和删除商品。

    public class Recv {
        private final static String QUEUE_NAME = "direct_exchange_queue_1";
        private final static String EXCHANGE_NAME = "direct_exchange_test";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 获取通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            
            // 绑定队列到交换机,同时指定需要订阅的routing key。假设此处需要update和delete消息
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
    
            // 定义队列的消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                        byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    System.out.println(" [消费者1] received : " + msg + "!");
                }
            };
            // 监听队列,自动ACK
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }
    

    消费者2

    我们此处假设消费者2接收所有类型的消息:新增商品,更新商品和删除商品。

    public class Recv2 {
        private final static String QUEUE_NAME = "direct_exchange_queue_2";
        private final static String EXCHANGE_NAME = "direct_exchange_test";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 获取通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            
            // 绑定队列到交换机,同时指定需要订阅的routing key。订阅 insert、update、delete
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
    
            // 定义队列的消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                        byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    System.out.println(" [消费者2] received : " + msg + "!");
                }
            };
            // 监听队列,自动ACK
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }
    

      

    测试

    我们分别发送增、删、改的RoutingKey,发现结果:

    3.订阅模型-Topic

    说明

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

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

    通配符规则:

    #:匹配一个或多个词

    *:匹配不多不少恰好1个词

    举例:

    item.#:能够匹配item.spu.insert 或者 item.spu

    item.*:只能匹配item.spu

    图示:

    解释:

    • 红色Queue:绑定的是usa.# ,因此凡是以 usa.开头的routing key 都会被匹配到

    • 黄色Queue:绑定的是#.news ,因此凡是以 .news结尾的 routing key 都会被匹配

    生产者

    使用topic类型的Exchange,发送消息的routing key有3种: item.isnertitem.updateitem.delete

    public class Send {
        private final static String EXCHANGE_NAME = "topic_exchange_test";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 获取通道
            Channel channel = connection.createChannel();
            // 声明exchange,指定类型为topic
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
            // 消息内容
            String message = "新增商品 : id = 1001";
            // 发送消息,并且指定routing key 为:insert ,代表新增商品
            channel.basicPublish(EXCHANGE_NAME, "item.insert", null, message.getBytes());
            System.out.println(" [商品服务:] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }
    

    消费者1

    我们此处假设消费者1只接收两种类型的消息:更新商品和删除商品

    public class Recv {
        private final static String QUEUE_NAME = "topic_exchange_queue_1";
        private final static String EXCHANGE_NAME = "topic_exchange_test";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 获取通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            
            // 绑定队列到交换机,同时指定需要订阅的routing key。需要 update、delete
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");
    
            // 定义队列的消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                        byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    System.out.println(" [消费者1] received : " + msg + "!");
                }
            };
            // 监听队列,自动ACK
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }
    

    消费者2

    我们此处假设消费者2接收所有类型的消息:新增商品,更新商品和删除商品。

    /**
     * 消费者2
     */
    public class Recv2 {
        private final static String QUEUE_NAME = "topic_exchange_queue_2";
        private final static String EXCHANGE_NAME = "topic_exchange_test";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 获取通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            
            // 绑定队列到交换机,同时指定需要订阅的routing key。订阅 insert、update、delete
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*");
    
            // 定义队列的消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                        byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    System.out.println(" [消费者2] received : " + msg + "!");
                }
            };
            // 监听队列,自动ACK
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }
    

      

     

    四.持久化

    如何避免消息丢失?

    1) 消费者的ACK机制。可以防止消费者丢失消息。

    2) 但是,如果在消费者消费之前,MQ就宕机了,消息就没了。

    所以我们需要将消息持久化到硬盘,以防服务宕机。

    要将消息持久化,前提是:队列、Exchange都持久化

    交换机持久化

    队列持久化

    消息持久化

     

    五.面试常见题目

    总结:

    面试题1:如何解决消息丢失?

    • ack(消费者确认)

    • 持久化

    • 生产者确认(publisher confirm):生产者发送消息后,等待mq的ACK,如果没有收到或者收到失败信息,则重试。如果收到成功消息则业务结束。

    • 可靠消息服务(可选):对于部分不支持生产者确认的消息队列,可以发送消息前,将消息持久化到数据库,并记录消息状态,后续消息发送、消费等过程都依赖于数据库中消息状态的判断和修改。

    面试题2:如何避免消息堆积

    • 通过同一个队列多消费者监听,实现消息的争抢,加快消息消费速度。

    面试题3:如何保证消息的有序性?

    答:大部分业务对消息的有序性要求不高,如果遇到对时序要求较高的业务,分两种情况来处理:

    • 业务同时对并发要求不高:

      • 保证消息发送时有序同步发送

      • 保证消息发送被同一个队列接收

      • 保证一个队列只有一个消费者,可以有从机(待机状态),实现高可用。

      • 实现主从(Zookeeper集群选主)

    • 业务同时对并发要求较高:

      • 满足上述第一个场景的条件

      • 可以有多个队列

      • 有时序要求的一组消息,通过hash方式分派到一个固定队列.

    面试题4:如何避免消息重复消费?

    • 保证接口幂等即可,那么如何保证接口幂等呢?

      • 某些接口天生幂等,例如查询请求

      • 某些接口天生不幂等,比如新增,还有某些接口的修改功能

        • 能根据具体的业务或状态来确定的,在消费端通过业务判断是否执行过

    面试题五:RabbitMQ用途:

      1.数据同步,比如说缓存,索引库,页面静态化它们都依赖于数据库,所以数据库一旦改变,都需要跟着变,可以利用MQ机制来做同步。

      2.异步调用解耦合,同步指的是这个代码直接利用工具类来调用方法,耦合高,如果没执行完则需一直等待阻塞。异步调用指的是MQ,比如你需要某个服务的某个代码,这个服务只要发MQ,然后你从MQ接受它就行,解耦合,它们并发执行效率高。

      3.流量削峰,比如说有许多人来发送复杂业务,执行耗时间长,这样服务调用并发差,我们可以MQ发送消息让复杂业务先执行,我先来接受大量并发请求。

  • 相关阅读:
    数组初始化 和 vector初始化
    剑指offer42 左旋转字符串
    k sum(lintcode)
    背包问题2 (lintcode)
    92.背包问题(lintcode)
    72. Edit Distance
    79 最长公共子串 (lintcode)
    77 最长公共子序列 (lintcode)
    132. Palindrome Partitioning II
    Mysql经常使用函数汇总
  • 原文地址:https://www.cnblogs.com/jike1219/p/11182480.html
Copyright © 2011-2022 走看看