SSM+shiro等javaweb收藏

rabbitMQ 消息模型和ACK消息确认机制

2018-08-12  本文已影响197人  Scorw

rabbitMQ官网：http://www.rabbitmq.com/getstarted.html

  1. 简单队列


    helloworld.png

    官方介绍：
    RabbitMQ是一个消息代理：它接受和转发消息。 你可以把它想象成一个邮局：当你把邮件放在邮箱里时，你可以确定邮差先生最终会把邮件发送给你的收件人。 在这个比喻中，RabbitMQ是邮政信箱，邮局和邮递员。
    RabbitMQ与邮局的主要区别是它不处理纸张，而是接受，存储和转发数据消息的二进制数据块。
    P（producer/ publisher）：生产者，一个发送消息的用户应用程序。
    C（consumer）：消费者，消费和接收有类似的意思，消费者是一个主要用来等待接收消息的用户应用程序
    队列（红色区域）：rabbitmq内部类似于邮箱的一个概念。虽然消息流经rabbitmq和你的应用程序，但是它们只能存储在队列中。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。
    总之：
    生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。
    我们将用Java编写两个程序;发送单个消息的生产者，以及接收消息并将其打印出来的消费者。我们将详细介绍Java API中的一些细节，专注于这个非常简单的事情，以便开始使用。这是一个消息传递的“Hello World”。
    我们将调用我们的消息发布者（发送者）Send和我们的消息消费者（接收者）Recv。发布者将连接到RabbitMQ，发送一条消息，然后退出。
    生产者发送消息到队列：

public class Send {
private final static String QUEUE_NAME = "test_queue";
public static void main(String[] argv) throws Exception {
    // 获取到连接以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    // 从连接中创建通道，这是完成大部分API的地方。
    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();
  }
}

执行控制台打印：


image.png

管理工具中查看消息：


image.png
消费者从管理工具中获取消息： 
     public class Recv {
private final static String QUEUE_NAME = "test_queue";
public static void main(String[] argv) throws Exception {
    // 获取到连接以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 声明队列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    // 定义队列的消费者
    Consumer consumer = new DefaultConsumer(channel) {
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                byte[] body) throws IOException {
            String message = new String(body, "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
        }
    };
    // 监听队列
    channel.basicConsume(QUEUE_NAME, true, consumer);
}
}

  1. Work queues
    工作队列或者竞争消费者模式


    worke queues.png

    在第一篇教程中，我们编写了一个程序，从一个命名队列中发送并接受消息。在这里，我们将创建一个工作队列，在多个工作者之间分配耗时任务。
    工作队列，又称任务队列。主要思想就是避免执行资源密集型任务时，必须等待它执行完成。相反我们稍后完成任务，我们将任务封装为消息并将其发送到队列。 在后台运行的工作进程将获取任务并最终执行作业。当你运行许多工人时，任务将在他们之间共享，但是一个消息只能被一个消费者获取。
    这个概念在Web应用程序中特别有用，因为在短的HTTP请求窗口中无法处理复杂的任务。
    接下来我们来模拟这个流程：
    P：生产者：任务的发布者
    C1：消费者，领取任务并且完成任务，假设完成速度较快
    C2：消费者2：领取任务并完成任务，假设完成速度慢
    默认消费机制没个领取相同的消息数量，一个一个消费，如果有些消费者消费速度快，有些消费者消费速度慢，消费慢的是否会造成消息堆积？
    Work模式的“能者多劳”：
    我们可以使用basicQos方法和prefetchCount = 1设置。 这告诉RabbitMQ一次不要向工作人员发送多于一条消息。 或者换句话说，不要向工作人员发送新消息，直到它处理并确认了前一个消息。 相反，它会将其分派给不是仍然忙碌的下一个工作人员。


    image.png

  2. 订阅模式：


    PublishSubscribe.png

    解读：

Exchange：交换机在另一篇 文章中有详细讲解：https://www.jianshu.com/p/a8668b85b914

  1. 路由模式


    Routing.png

    在订阅模式中，生产者发布消息，所有消费者都可以获取所有消息。
    在路由模式中，我们将添加一个功能 - 我们将只能订阅一部分消息。 例如，我们只能将重要的错误消息引导到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。
    但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
    在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
    消息的发送方在 向 Exchange发送消息时，也必须指定消息的 routing key。
    P:生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
    X:Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
    C1：消费者，其所在队列指定了需要routing key 为 error 的消息
    C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

5.通配符模式：

Topics.png
消息必须以一定的格式发送到路由：
1） 它必须是由点分隔的单词列表。单词可以是任何东西，但通常它们指定了与该消息相关的一些功能。只要您愿意，路由键中可以有多少个字，最多255个字节。但是队列绑定时，想获取对应的数据，则绑定键也必须是相同的形式。交换机背后的逻辑类似于：一个消息根据特殊的routing key发送到匹配该key的绑定队列中。
2） 绑定的routing key 也可以使用通配符：
：匹配不多不少一个词
#：匹配一个或多个词
在这个例子中，我们将发送所有描述动物的消息。消息将使用由三个字（两个点）组成的routing key发送。路由关键字中的第一个单词将描述速度，第二个颜色和第三个种类：“<speed>.<color>.<species>”。
我们创建了三个绑定：Q1绑定了绑定键“ .orange.”，Q2绑定了“.*.rabbit”和“lazy.＃”。
Q1匹配所有的橙色动物。
Q2匹配关于兔子以及懒惰动物的消息。

  1. RPC:不属于mq这里不做讲解。


    RPC.png

ACK机制 ：消息确认机制

在MQ中，队列中的任何消息，只能被消费1次，一旦消费，立刻删除！

为什么需要消息确认？
RabbitMQ的消息机制中，如果一个队列的消息被成功消费，那么消息就会被删除。
那么问题来了：RabbitMQ如何判断消息是否被消费了呢？仅仅是有消费者领取消息就可以了吗？
如果消费者领取消息后，还没执行操作就挂掉了呢？或者抛出了异常？消息消费失败，但是RabbitMQ无从得知，这样消息就丢失了！
因此，RabbitMQ有一个ACK机制
在RabbitMQ中，消息确认有两种模式：
i. 自动模式，我们无需任何操作，在消息被消费者领取后，就会自动确认，消息也会被从队列删除。
ii. 手动模式，消息被消费后，我们需要调用RabbitMQ提供的API来实现消息确认。
我们在调用：channel.basicConsume()方法的时候，通过指定第二个参数来设置是自动还是手动：


image.png image.png

自动ACK存在的问题：消费者，程序抛出异常。但是消息依然被消费。

防止消息丢失可以手动：演示如下：

当消费者发生异常时:


image.png

等待排除异常以后还可以正常消费：


image.png

演示demo案例可以区码云下载：https://gitee.com/currey/rabbitMQ

上一篇下一篇

猜你喜欢

热点阅读