rabbitmq消费者获取消息慢

rabbitmq消费者获取消息慢

简单队列

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 = Connection();

// 从连接中创建通道，这是完成大部分API的地方。

Channel channel = ateChannel();

// 声明(创建)队列，必须声明队列才能够发送消息，我们可以把消息发送到队列中。

// 声明一个队列是幂等的 - 只有当它不存在时才会被创建

channel.queueDeclare(QUEUE_NAME, false, false, false, null);

// 消息内容

String message = "Hello World!";

channel.basicPublish("", QUEUE_NAME, null, Bytes());

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 = Connection();

Channel channel = ateChannel();

// 声明队列

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);

}

}

Work queues

工作队列或者竞争消费者模式

worke queues.png

在第一篇教程中，我们编写了一个程序，从一个命名队列中发送并接受消息。在这里，我们将创建一个工作队列，在多个工作者之间分配耗时任务。

工作队列，又称任务队列。主要思想就是避免执行资源密集型任务时，必须等待它执行完成。相反我们稍后完成任务，我们将任务封装为消息并将其发送到队列。 在后台运行的工作进程将获取任务并最终执行作业。当你运行许多工人时，任务将在他们之间共享，但是一个消息只能被一个消费者获取。

这个概念在Web应用程序中特别有用，因为在短的HTTP请求窗口中无法处理复杂的任务。

接下来我们来模拟这个流程：

P：生产者：任务的发布者

C1：消费者，领取任务并且完成任务，假设完成速度较快

C2：消费者2：领取任务并完成任务，假设完成速度慢

默认消费机制没个领取相同的消息数量，一个一个消费，如果有些消费者消费速度快，有些消费者消费速度慢，消费慢的是否会造成消息堆积？

Work模式的“能者多劳”：

我们可以使用basicQos方法和prefetchCount = 1设置。 这告诉RabbitMQ一次不要向工作人员发送多于一条消息。 或者换句话说，不要向工作人员发送新消息，直到它处理并确认了前一个消息。 相反，它会将其分派给不是仍然忙碌的下一个工作人员。

image.png

订阅模式：

PublishSubscribe.png

解读：

个生产者，多个消费者

每一个消费者都有自己的一个队列

生产者没有将消息直接发送到队列，而是发送到了交换机

每个队列都要绑定到交换机

生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的

X(Exchanges)：交换机一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

Exchange类型有以下几种：

Fanout：广播，将消息交给所有绑定到交换机的队列

Direct：定向，把消息交给符合指定routing key 的队列

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

路由模式

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发送。路由关键字中的第一个单词将描述速度，第二个颜色和第三个种类：“..”。

我们创建了三个绑定：Q1绑定了绑定键“ .orange.”，Q2绑定了“.*.rabbit”和“lazy.＃”。

Q1匹配所有的橙色动物。

Q2匹配关于兔子以及懒惰动物的消息。

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