我正在为一些发布/订阅(发布者/订阅者)代码使用 .Net RabbitMQ。一切正常,直到我开始关闭消费者。消费者正确处理发布的数据,直到我关闭最后一个消费者。毕竟消费者,我打开一个新的消费者,但没有任何反应。应用程序打开,但它没有收到来自发布者的任何数据。
我检查了发布者代码,发现当最后一个消费者关闭时,其频道的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));
}
}