RabbitMQ只处理50条消息，然后阻止

我在.net中使用RabbitMQ，当我在队列中放置100条消息时，看到一个奇怪的问题。它处理约50条消息，然后Dequeue()方法只是挂起。如果我重新启动服务，它会处理剩余的项目。RabbitMQ只处理50条消息，然后阻止

编辑：它正在处理完全50％的队列。当我添加1000条消息时，它只处理500条。即使当单线程时

我在这里丢失了什么？

private void InitializeAgent() { 
     var agentFactory = new ConnectionFactory() { HostName = "localhost" }; 
     agentConnection = agentFactory.CreateConnection(); 
     agentChannel = agentConnection.CreateModel(); 
     var ok = agentChannel.QueueDeclare(GetType().Name, true, false, false, null); 
     consumer = new QueueingBasicConsumer(agentChannel); 
     agentChannel.BasicConsume(GetType().Name, false, consumer); 
    } 

    public void DequeueMessages() { 
     ThreadPool.SetMaxThreads(200, 200); 
     ThreadPool.SetMinThreads(200, 200); 
     var ea = consumer.Queue.Dequeue(); 
     ThreadPool.QueueUserWorkItem(ProcessWorkInThread, ea); 
    } 

    public void AgentTask() { 
     var instance = factory.GetInstance(threadItem); 

     while (true) 
      DequeueMessages(); 
    } 

    private void ProcessWorkInThread(object state) { 
     var ea = state as BasicDeliverEventArgs; 

     var message = Encoding.UTF8.GetString(ea.Body); 

     var settings = new JsonSerializerSettings(); 
     settings.ContractResolver = new DefaultContractResolver() { DefaultMembersSearchFlags = BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public }; 
     var item = JsonConvert.DeserializeObject<TEntity>(message, settings); 

     Thread.Sleep(10000) //simulate work 
     lock (agentChannel)    
      agentChannel.BasicAck(ea.DeliveryTag, false);    
    }

来源

2014-10-31 Chris Kooken

相当确定IModel不是线程安全的。 .net用户指南明确指出，不应该在线程之间共享IModel。 – user1450877 2014-10-31 14:50:31

@ user1450877我可以在dequeue＆ack周围添加锁吗？ – 2014-10-31 15:01:35

必须为线程使用一个IModel。 – Gabriele 2014-10-31 15:10:15

这必须是一个.NET客户端版本问题，使用3.4.0，下面的代码工作正常。

static readonly ConnectionFactory Factory = new ConnectionFactory { HostName = "localhost" }; 
static readonly IConnection Connection = Factory.CreateConnection(); 
static QueueingBasicConsumer consumer; 
static IModel agentChannel; 

static CancellationTokenSource _tokenSource; 

static void Main(string[] args) 
{ 
    _tokenSource = new CancellationTokenSource(); 

    const string queueName = "testQueue"; 
    agentChannel = Connection.CreateModel(); 
    agentChannel.QueueDeclare(queueName, true, false, false, null); 
    agentChannel.QueueBind(queueName, "testExchange", ""); 

    consumer = new QueueingBasicConsumer(agentChannel); 
    agentChannel.BasicConsume(queueName, false, consumer); 

    while (!_tokenSource.Token.IsCancellationRequested) 
    { 
     DequeueMessages(); 
    } 
    Console.ReadLine(); 
    _tokenSource.Cancel(); 
} 

static void DequeueMessages() 
{ 
    ThreadPool.SetMaxThreads(200, 200); 
    ThreadPool.SetMinThreads(200, 200); 
    var ea = consumer.Queue.Dequeue(); 
    ThreadPool.QueueUserWorkItem(ProcessWorkInThread, ea); 
} 

static void ProcessWorkInThread(object state) 
{ 
    var ea = state as BasicDeliverEventArgs; 

    var message = Encoding.UTF8.GetString(ea.Body); 

    var settings = new JsonSerializerSettings(); 
    settings.ContractResolver = new DefaultContractResolver() 
    { 
     DefaultMembersSearchFlags = BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public 
    }; 
    var item = JsonConvert.DeserializeObject<string>(message, settings); 
    Console.WriteLine(item); 
    Thread.Sleep(10000); //simulate work 
    lock (agentChannel) 
     agentChannel.BasicAck(ea.DeliveryTag, false); 
}

来源

2015-01-14 08:53:11 Bensmind

RabbitMQ只处理50条消息，然后阻止

回答

相关问题