MassTransit RabbitMQ 오류 대기열에서 소비 된 메시지의 절반을 Error_Skipped 대기열로 이동합니다.

Question

오류 대기열을 원래 대기열로 다시 이동하려고합니다. 이 작업을 수행하기 위해 Error Queue에 Consumer를 생성 한 다음 Required Queue에 게시했습니다. 이것을 사용하면 소비 된 메시지의 절반이 게시되지만 나머지 절반은 Error_Skipped 대기열로 전송됩니다.

나는 성공하지 못한 채 많은 것을 시도해 보았습니다. 그래서 그것은 제가 누락 된 것일 수도 있습니다.

여기 내 코드의 샘플은 다음과 같습니다

public class ClaimsMessage 
{ 
    public string Description { get; set; } 

    public DateTime Date { get; set; } 

    public bool Handled { get; set; } 
} 

public class ClaimsMessageErrorConsumer : IConsumer<Fault<ClaimsMessage>> 
{ 
    public async Task Consume(ConsumeContext<Fault<ClaimsMessage>> context) 
    { 
     try 
     { 
      await context.Publish<ClaimsMessage>(context.Message.Message); 

     } 
     catch (Exception e) 
     { 
      string error = e.Message; 
     } 
    } 
} 

public static IBusControl CreateClaimsErrorConsumerBus(string endPoint) 
{ 
    var busControl = Bus.Factory.CreateUsingRabbitMq(cfg => 
    { 
     var host = cfg.Host(new Uri("rabbitmq://localhost/"), h => 
     { 
      h.Username("guest"); 
      h.Password("guest"); 
     }); 

     cfg.ReceiveEndpoint(host, endPoint, e => 
     { 
      e.Consumer(() => new ClaimsMessageErrorConsumer()); 
     }); 
    }); 
    return busControl; 
} 

Answer 1

다시 처리 큐에 오류 큐에서 메시지를 이동하는 경우, 당신은 Publish를 호출해서는 안 -이 모든 가입자에게 메시지를 다시 보냅니다. 큐 이름을 이미 알고 있으므로 메시지를 다시 큐에 보냅니다. 여러분이보고있는 것은 오류 대기열에 소비자를 생성했기 때문에 해당 메시지에 대한 교환 바인딩이 만들어 졌다는 것입니다.

그래서, 대신 이렇게 : 메시지의 충실도가 유지되도록

추가 신용 들어

sbc.ReceiveEndpoint("input_error", x => 
{ 
    // this prevents extra message bindings from being created 
    x.BindMessageExchanges = false; 

    x.Consumer<MyMover>(() => new MyMover(inputQueueAddress); 
}); 

public class MyMover : 
    IConsumer<ClaimsMessage> 
{ 
    public async Task Consume(ConsumeContext<ClaimsMessage> context) 
    { 
     try 
     { 
      var endpoint = await context.GetSendEndpoint(_inputQueueAddress); 
      await endpoint.Send<ClaimsMessage>(context.Message); 
     } 
     catch (Exception e) 
     { 
      string error = e.Message; 
     } 
    } 
} 

원래 메시지 헤더를 통해 복사합니다.