使用线程 RabbitMQ 使用者处理事件

Question

我在 C# 中使用 RabbitMQ dot net 库在不同的队列上有两个消费者。

我想要什么：

由于一些业务逻辑，我必须在一个消费者中等待一段时间 所以我为此目的使用了Thread.Sleep()

问题

如果我在一个事件中使用Thread.Sleep，第二个线程也不会暂停

我的代码：

consumer.Received += (model, ea) =>
{
try
{
DRModel drModel = JsonConvert.DeserializeObject<DRModel>(Encoding.UTF8.GetString(ea.Body));
RMQReturnType type = ProcessSubmitSMS(drModel);
if (type == RMQReturnType.ACK)
channel.BasicAck(ea.DeliveryTag, false);
else
{ 
channel.BasicNack(ea.DeliveryTag, false, true);
Thread.Sleep(300000); // <=== SLEEP
}
}
catch (Exception ex)
{
channel.BasicNack(ea.DeliveryTag, false, true);
WriteLog(ControlChoice.ListError, "Exception: " + ex.Message + " | Stack Trace: " + ex.StackTrace.ToString() + " | [Consumer Event]");
}
};

Answer 1

这似乎是互斥类的一个很好的例子，你需要的是多线程中的条件睡眠。不知道你需要的逻辑，但你的东西类似于下面的代码：

public class Consumer
{
public event EventHandler Received;
public virtual void OnReceived()
{
Received?.Invoke(this, EventArgs.Empty);
}
}
class Program
{
static void Main(string[] args)
{
var mutex = new Mutex();
var consumer =  new Consumer();
consumer.Received += (model, ea) =>
{
try
{
mutex.WaitOne();
var id = Guid.NewGuid().ToString();
Console.WriteLine($"Start mutex {id}");
Console.WriteLine($"Mutex finished {id}");
Console.WriteLine($"Start sleep {id}");
if ( new Random().Next(10000)  % 2 == 0) // randomly sleep, that your condition
{
Thread.Sleep(3000); // <=== SLEEP
}
Console.WriteLine($"Sleep finished {id}");
}
catch (Exception ex)
{
mutex.ReleaseMutex(); // this is where you release, if something goes wrong
}
finally
{
mutex.ReleaseMutex();// always release it
}
};

Parallel.For(0, 10, t =>   //running 10 threads in parallel and stops all if the condition is true
{
consumer.OnReceived();
});
Console.ReadLine();
}
}

}

Answer 2

我的代码中存在一些逻辑错误，我理解。 在兔子 我在两个不同的频道上创建了两个消费者事件，所以我认为它不会在这里共享，我错了，连接是共享黑白频道的，所以我为此明确定义了两个连接。 据我了解 消费者块通道和通道块连接和连接在两个事件中是相同的。