任务分发
任务分发机制
使用任务队列的优点之一是可以轻易的进行一步工作。
如果我们现在积压了很多工作,可以通过增加消费者来解决这个问题,使得系统伸缩性更加容易
Round-robin(轮询分发)
发布者 RabbitMQ 发送几条消息1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
namespace WorkMQProduct
{
class NewTask
{
static string QueueName = "task_queue";
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
HostName = "localhost",
VirtualHost = "frexport",
UserName = "guest",
Password = "guest"
};
List<int> taskMessages = new List<int> { 2, 7, 2, 6, 5, 2, 2, 3 };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: QueueName,
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
var properties = channel.CreateBasicProperties();
properties.SetPersistent(true);
for (int i = 0; i < 8; i++)
{
string message = taskMessages[i] + "";
var body = Encoding.UTF8.GetBytes(message);
properties.CorrelationId = i + "";
channel.BasicPublish(exchange: "",
routingKey: QueueName,
basicProperties: properties,
body: body);
Console.WriteLine(" [x] {0} Sent {1}", properties.CorrelationId, message);
}
}
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
}
<!--more-->
消费者1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Threading;
namespace WorkMQConsumer
{
class Worker
{
static string QueueName = "task_queue";
static void Main(string[] args)
{
var factory = new ConnectionFactory() { };
factory.HostName = "localhost";
factory.VirtualHost = "frexport";
factory.UserName = "guest";
factory.Password = "guest";
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: QueueName,
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
Console.WriteLine(" [*] Waiting for message.");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (sender, ea) =>
{
var body = ea.Body;
var message = System.Text.Encoding.UTF8.GetString(body);
int x = int.Parse(message);
Console.WriteLine(" [x] Task {0} Receive {1} {2}", ea.BasicProperties.CorrelationId, message, DateTime.Now);
Thread.Sleep(1000*x);
Console.WriteLine(" [x] Done! at {0}", DateTime.Now);
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
channel.BasicConsume(queue: QueueName,
noAck: false,
consumer: consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
}
}
}
先启动两个消费者1
2$ MDConsumer/bin/Debug/MDConsumer.exe
$ MDConsumer/bin/Debug/MDConsumer.exe
再启动一个生产者1
$ MDProduct/bin/Debug/MDProduct.exe
效果
这个地方其实,所有消息会很快传给消费者,虽然没有消息应答
从上述的结果中,我们可以得知,在默认情况下,RabbitMQ不会顾虑消息者处理消息的能力,即使其中有的消费者闲置有的消费者高负荷。RabbitMQ会逐个发送消息到在序列中的下一个消费者(而不考虑每个任务的时长等等,且是提前一次性分配,并非一个一个分配)。平均每个消费者获得相同数量的消息,这种方式分发消息机制称为Round-Robin(轮询)。
Fair dispatch(公平分发)
您可能已经注意到,任务分发仍然没有完全按照我们想要的那样。比如:现在有2个消费者,所有的奇数的消息都是繁忙的,而偶数则是轻松的。按照轮询的方式,奇数的任务交给了第一个消费者,所以一直在忙个不停。偶数的任务交给另一个消费者,则立即完成任务,然后闲得不行。而RabbitMQ则是不了解这些的。这是因为当消息进入队列,RabbitMQ就会分派消息。它不看消费者为应答的数目,只是盲目的将第n条消息发给第n个消费者。
公平分发,则是根据消费者的处理能力来进行分发处理的。这里主要是通过设置prefetchCount 参数来实现的。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理规定的数量级个数的Message。换句话说,在接收到该Consumer的ack前,它不会将新的Message分发给它。 比如prefetchCount=1,则在同一时间下,每个Consumer在同一个时间点最多处理1个Message,同时在收到Consumer的ack前,它不会将新的Message分发给它。
graph LR; P([Product]); mq([Message Quene]) C1([Consumer1]); C2([Consumer2]); style P fill:#0ff,stroke:#333; style mq fill:#f00,stroke:#333; style C1 fill:#3cf,stroke:#333; style C2 fill:#3cf,stroke:#333; P-->mq; mq-->|prefetch=1|C1; mq-->|prefetch=1|C2;
修改工作线程1
channel.BasicQos(prefetchCount: 1, prefetchSize: 0, global: false);
注:如果所有的工作者都处于繁忙状态,你的队列有可能被填充满。你可能会观察队列的使用情况,然后增加工作者,或者使用别的什么策略。
还有一点需要注意,使用公平分发,必须关闭自动应答,改为手动应答。
效果
消息每次只会发送一条给消费者,只有消费者处理完成后,才会分发新的消息