任务分发
任务分发机制
使用任务队列的优点之一是可以轻易的进行一步工作。
如果我们现在积压了很多工作，可以通过增加消费者来解决这个问题，使得系统伸缩性更加容易
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);
注：如果所有的工作者都处于繁忙状态，你的队列有可能被填充满。你可能会观察队列的使用情况，然后增加工作者，或者使用别的什么策略。
还有一点需要注意，使用公平分发，必须关闭自动应答，改为手动应答。
效果
消息每次只会发送一条给消费者，只有消费者处理完成后，才会分发新的消息