RabbitMQ Pub/Sub：关闭最后一个消费者会关闭发布者的通道/模型。为什么？

Question

我将 .Net RabbitMQ 用于一些 pub/sub（发布者/订阅者）代码。一切正常，直到我开始关闭消费者。消费者正确处理发布的数据，直到我关闭最后一个消费者。毕竟消费者，我打开一个新的消费者，但没有任何反应。应用程序打开，但它没有收到来自发布者的任何数据。

我检查了发布者代码，发现当最后一个消费者关闭时，其频道的 IsOpen 属性变为 false。我不知道是否有一些设置可以让频道即使在其消费者关闭后也保持打开状态。

这是我的发布商代码： 编辑我最初粘贴了错误的代码。

这是我的消费者代码：

public MyConsumer
{
private readonly ConnectionFactory _factory;
private readonly IConnection _connection;
private readonly IModel _channel;
private readonly Timer _timer;

private SubscriptionConsumerType(string ipAddress, string exchangeName, TimeSpan tsPullCycle)
{
    //set up connection
    this._factory = new ConnectionFactory();
    this._factory.HostName = ipAddress;
    this._connection = this._factory.CreateConnection();
    this._channel = this._connection.CreateModel();

    //set up and bind the exchange
    this._channel.ExchangeDeclare(exchangeName, "fanout", false, true, new Dictionary<string, object>());
    string queueName = this._channel.QueueDeclare().QueueName;
    this._channel.QueueBind(queueName, exchangeName, "");

    //start consuming
    QueueingBasicConsumer consumer = new QueueingBasicConsumer(this._channel);
    this._channel.BasicConsume(queueName, true, consumer);

    //periodically check for new messages from the publisher
    this._timer = new Timer(new TimerCallback(this.TimerStep), consumer, tsPullCycle, tsPullCycle);
}

public void Dispose()
{
    if (this._timer != null)
        this._timer.Dispose();

    if (this._channel != null)
    {
        this._channel.Close();
        this._channel.Dispose();
    }

    if (this._connection != null)
    {
        this._connection.Close();
        this._connection.Dispose();
    }
}
}

目前，我的解决方法是始终在某处打开一个消费者窗口。不过，理想情况下，我希望我的发布者无论打开多少个消费者窗口都可以运行。谢谢。

编辑糟糕，我粘贴了错误的生产者代码。就是这样：

private SubscriptionBroadcastType(string ipAddress, string exchangeName)
{
    this._factory = new ConnectionFactory();
    this._factory.HostName = ipAddress;
    this._connection = this._factory.CreateConnection();
    this._channel = this._connection.CreateModel();

    this._exchangeName = exchangeName;
    this._channel.ExchangeDeclare(exchangeName, SubscriptionBroadcastType.BROADCAST, SubscriptionBroadcastType.DURABLE, SubscriptionBroadcastType.AUTO_DELETE, new Dictionary<string, object>());
}

public void BroadcastMessage(string message)
{
    lock (this._syncroot) //protect _channel
    {
        if (this._channel.IsOpen)
            this._channel.BasicPublish(this._exchangeName, "", null, System.Text.Encoding.UTF8.GetBytes(message));
    }
}

Answer 1

我认为你在这里可能有一些根本性的错误。请检查您是否发布了正确的代码。当我读到它时，您有一个生产者创建一个特定的命名队列并直接发布到队列。您有一个消费者创建一个特定的命名交换，然后创建一个动态命名的新队列并将其绑定到交换。然后，您正在从此队列中读取，但它不能是您最初发布到的队列。

我会先修复您的代码，在您的发布者代码中创建一个交换，使用您可以在您的消费者代码中访问的特定名称。此行将出现在生产者线程中，而不是队列声明行中：

this._channel.ExchangeDeclare(exchangeName, "fanout", false, true, new Dictionary<string, object>());

然后您将需要发布到该交换所以而不是发布到队列的行更改为：

this._channel.BasicPublish(exchangeName, "", this._basicProperties, System.Text.Encoding.UTF8.GetBytes(message));

您的消费者应该被正确设置为从队列中接收这些消息。看看这是否有助于解决您的问题。