RabbitMQ，是否可以在保持FIFO的同时从一个通道消耗1条消息？

Question

我正在使用带有C#的RabbitMQ作为消息代理框架。在我的场景中，我有许多使用者可以从不同的队列中使用。从使用者的角度来看，几个队列形成一个逻辑组。

例如：队列1、队列2、队列3。

消费者1想要从[队列1，队列3]获取第一条消息，并且只有在不可用时才尝试从队列2获取

Consumer2将希望从[Queue2，Queue3]获取第一条消息，只有在不可用时，才尝试从Queue1获取

下面是一个应该返回消息的类：

public class MessagesProvider
{

    private IConnection _connection;
    private IModel _channel;
    private readonly IConnectionFactory _connectionFactory;
    private ConcurrentDictionary<string, string> _tagsQueueItems;
    private EventingBasicConsumer _consumer;
    private CancellationTokenSource _cts;
    private TaskCompletionSource<BasicGetResult> _tcs;

    public MessagesProvider()
    {
        _connectionFactory = new ConnectionFactory()
        {
            HostName = "localhost"
        };

    }

    public Task<BasicGetResult> GetMessage(int timeout, IEnumerable<string> queues)
    {
        _tagsQueueItems = new ConcurrentDictionary<string, string>();
        _cts = new CancellationTokenSource();
        _connection = _connectionFactory.CreateConnection();
        _channel = _connection.CreateModel();
        _channel.SingleMessagePerChannel();
        _consumer = new EventingBasicConsumer(_channel);
        _consumer.Received += OnReceive;

        foreach (var queue in queues)
        {
            var tag = _channel.BasicConsume(queue, false, _consumer);
            _tagsQueueItems.AddOrUpdate(queue, tag, (k, o) => queue);
        }

        return _tcs.Task;
    }

    private void SetResult(BasicGetResult result)
    {
        _tcs.SetResult(result);
        if (_channel.IsOpen)
            _channel.Close();
        if (_connection.IsOpen)
            _connection.Close();
    }

    private void OnReceive(object sender, BasicDeliverEventArgs e)
    {
        _consumer.Received -= OnReceive;
        _channel.BasicAck(e.DeliveryTag, false);
        var result = new BasicGetResult(e.DeliveryTag, e.Redelivered, e.Exchange, e.RoutingKey, 1,
            e.BasicProperties, e.Body);

        SetResult(result);

    }
}

通道定义：

 public static class  ChannelExtension
{
    public static void SingleMessagePerChannel(this IModel channel)
    {
        channel.BasicQos(0,1,true);
    }

    public static void SingleMessagePerConsumer(this IModel channel)
    {
        channel.BasicQos(0,1,false);
    }
}

问题是我一个接一个地注册一个队列(在“GetMessage”方法中)，我找不到一个“原子”操作来从一组队列中取出下一个FIFO消息。

我正在寻找这样做的方法：_consumer.getNextMessage（）；

另一种方法是使用来自第一组优先级的所有最前面的消息，在消费者端过滤它们以找到最旧的消息，并且不为其他消息发送ack（在同一优先级组中）。这种方法也有问题，因为它意味着在获取消息时，其他消费者将无法处理它们（直到没有ack）。

有什么建议吗？谢了。

Answer 1

我认为你并没有按照设计的方式使用队列和消费者。将队列视为要完成的工作的桶。每个工作桶可以有N个消费者来做这项工作。

每个消费者都应该设计为使用单个队列，而不知道或关心其他队列或其他消费者。

如果您试图实现“高优先级”队列，那么我建议您使用具有2个不同队列的单个交换：

< li >常规优先级队列 < li >高优先级队列

然后，默认情况下，消息被放入常规优先级队列。当需要高优先级的消息出现时，您可以设置消息的标头，以便将其推送到高优先级队列中。

我们今天使用同样的方法，使用直接队列，并在我们希望消息进入高优先级队列时添加一个头。