【发布时间】:2017-07-29 19:45:05
【问题描述】:
我有一段代码使用 rabbitMQ 来管理一段时间内的作业列表。 因此,我有一个连接和一个对 RabbitMQ 服务器开放的通道来执行这些作业的操作。我使用以下内容排队作业:
public override void QueueJob(string qid, string jobId) {
this.VerifyReadyToGo();
this.CreateQueue(qid);
byte[] messageBody = Encoding.UTF8.GetBytes(jobId);
this.channel.BasicPublish(
exchange: Exchange,
routingKey: qid,
body: messageBody,
basicProperties: null
);
OLog.Debug($"Queued job {jobId} on {qid}");
}
public override string RetrieveJobID(string qid) {
this.VerifyReadyToGo();
this.CreateQueue(qid);
BasicGetResult data = this.channel.BasicGet(qid, false);
string jobData = Encoding.UTF8.GetString(data.Body);
int addCount = 0;
while (!this.jobWaitingAck.TryAdd(jobData, data.DeliveryTag)) {
// try again.
Thread.Sleep(10);
if (addCount++ > 2) {
throw new JobReceptionException("Failed to add job to waiting ack list.");
}
}
OLog.Debug($"Found job {jobData} on queue {qid} with ackId {data.DeliveryTag}");
return jobData;
}
问题在于,在这样的任何方法调用(Publish、Get 或 Acknowledge)之后,都会创建某种后台线程,当通道和连接关闭时,该线程不会关闭。 这意味着测试通过并且操作成功完成,但是当应用程序尝试关闭时它会挂起并且永远不会完成。
这里是连接方法供参考
public override void Connect() {
if (this.Connected) {
return;
}
this.factory = new ConnectionFactory {
HostName = this.config.Hostname,
Password = this.config.Password,
UserName = this.config.Username,
Port = this.config.Port,
VirtualHost = VirtualHost
};
this.connection = this.factory.CreateConnection();
this.channel = this.connection.CreateModel();
this.channel.ExchangeDeclare(
exchange: Exchange,
type: "direct",
durable: true
);
}
我可以做些什么来纠正这个问题(rabbitmq 客户端阻止应用程序退出)?
【问题讨论】: