c# - EasyNetQ-如何重试失败的消息并在邮件正文/标题中保留RetryCount？

Question

我正在使用EasyNetQ，需要重试原始队列中的失败消息。问题是:即使我成功地增加了TriedCount变量(在每个msg的正文中)，当EasyNetQ在异常发生后将消息发布到默认错误队列时，更新的TriedCount也不在msg中!大概是因为它只是将原始消息转储到错误队列，而没有使用方的更改。

更新的TriedCount适用于进程内重新发布，但不适用于通过EasyNetQ Hosepipe或EasyNetQ Management Client重新发布的情况。 Hosepipe生成的文本文件没有更新TriedCount。

public interface IMsgHandler<T> where T: class, IMessageType
{
    Task InvokeMsgCallbackFunc(T msg);
    Func<T, Task> MsgCallbackFunc { get; set; }
    bool IsTryValid(T msg, string refSubscriptionId); // Calls callback only 
                                                      // if Retry is valid
}

public interface IMessageType
{
    int MsgTypeId { get; }

    Dictionary<string, TryInfo> MsgTryInfo {get; set;}

}

public class TryInfo
{   
    public int TriedCount { get; set; }

    /*Other information regarding msg attempt*/
}

public bool SubscribeAsync<T>(Func<T, Task> eventHandler, string subscriptionId)
{
    IMsgHandler<T> currMsgHandler = new MsgHandler<T>(eventHandler, subscriptionId);
    // Using the msgHandler allows to add a mediator between EasyNetQ and the actual callback function
    // The mediator can transmit the retried msg or choose to ignore it
    return _defaultBus.SubscribeAsync<T>(subscriptionId, currMsgHandler.InvokeMsgCallbackFunc).Queue != null;
}

我还尝试通过 Management API(粗糙代码)重新发布自己:

var client = new ManagementClient("http://localhost", "guest", "guest");
var vhost = client.GetVhostAsync("/").Result;
var errQueue = client.GetQueueAsync("EasyNetQ_Default_Error_Queue", 
vhost).Result;
var crit = new GetMessagesCriteria(long.MaxValue, 
Ackmodes.ack_requeue_true);
var errMsgs = client.GetMessagesFromQueueAsync(errQueue, 
crit).Result;

foreach (var errMsg in errMsgs)
{
    var pubRes = client.PublishAsync(client.GetExchangeAsync(errMsg.Exchange, vhost).Result,
                                 new PublishInfo(errMsg.RoutingKey, errMsg.Payload)).Result;
        }

这有效，但仅再次发布到错误队列，而不发布在原始队列上。另外，在此阶段，我不知道如何在邮件正文中添加/更新重试信息。

我已经探索了 this库来向邮件添加 header ，但是我没有看到正文中的计数是否未更新， header 中的计数如何/为什么被更新。

有什么方法可以在不依靠Advanced总线的情况下保持TriedCount(在这种情况下，我可能会使用RabbitMQ .Net客户端本身)？

Answer 1

为了以防万一，可以最终实现我自己的IErrorMessageSerializer(而不是实现整个IConsumerErrorStrategy，这似乎是一个过大的选择)。我在主体(而不是 header )中添加重试信息的原因是EasyNetQ不能处理 header 中的复杂类型(无论如何都不是开箱即用的)。因此，使用字典可以为不同的使用者提供更多控制。我在创建总线时注册了自定义序列化器，如下所示:

_defaultBus = RabbitHutch.CreateBus(currentConnString, serviceRegister => serviceRegister.Register<IErrorMessageSerializer>(serviceProvider => new RetryEnabledErrorMessageSerializer<IMessageType>(givenSubscriptionId)));

并像这样实现了Serialize方法:

 public class RetryEnabledErrorMessageSerializer<T> : IErrorMessageSerializer where T : class, IMessageType
 {
        public string Serialize(byte[] messageBody)
        {
             string stringifiedMsgBody = Encoding.UTF8.GetString(messageBody);
             var objectifiedMsgBody = JObject.Parse(stringifiedMsgBody);

             // Add/update RetryInformation into objectifiedMsgBody here
             // I have a dictionary that saves <key:consumerId, val: TryInfoObj>

             return JsonConvert.SerializeObject(objectifiedMsgBody);
        }
  }

实际的重试由一个简单的控制台应用程序/Windows服务通过EasyNetQ Management API定期进行:

            var client = new ManagementClient(AppConfig.BaseAddress, AppConfig.RabbitUsername, AppConfig.RabbitPassword);
            var vhost = client.GetVhostAsync("/").Result;
            var aliveRes = client.IsAliveAsync(vhost).Result;
            var errQueue = client.GetQueueAsync(Constants.EasyNetQErrorQueueName, vhost).Result;
            var crit = new GetMessagesCriteria(long.MaxValue, Ackmodes.ack_requeue_false);
            var errMsgs = client.GetMessagesFromQueueAsync(errQueue, crit).Result;
            foreach (var errMsg in errMsgs)
            {
                var innerMsg = JsonConvert.DeserializeObject<Error>(errMsg.Payload);
                var pubInfo = new PublishInfo(innerMsg.RoutingKey, innerMsg.Message);
                pubInfo.Properties.Add("type", innerMsg.BasicProperties.Type);
                pubInfo.Properties.Add("correlation_id", innerMsg.BasicProperties.CorrelationId);
                pubInfo.Properties.Add("delivery_mode", innerMsg.BasicProperties.DeliveryMode);
                var pubRes = client.PublishAsync(client.GetExchangeAsync(innerMsg.Exchange, vhost).Result,
                     pubInfo).Result;
            }

我的使用者本身就知道是否启用了重试功能，从而赋予了它更多的控制权，因此它可以选择处理重试的msg或忽略它。一旦忽略，则味精显然将不会再尝试；这就是EasyNetQ的工作方式。