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

Question

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

更新后的 TriedCount 适用于进程内重新发布，但不适用于通过 EasyNetQ Hosepipe 或 EasyNetQ 管理客户端重新发布时。 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 库以向消息添加标头，但我看不到正文中的计数是否没有更新，如何/为什么更新标头中的计数。

有没有办法在不使用高级总线的情况下保留 TriedCount（在这种情况下，我可能会使用 RabbitMQ .Net 客户端本身）？

Answer 1

以防万一它对其他人有所帮助，我最终实现了自己的IErrorMessageSerializer（而不是实现整个IConsumerErrorStrategy，这似乎有点矫枉过正）。我在正文（而不是标题）中添加重试信息的原因是 EasyNetQ 不处理标题中的复杂类型（无论如何都不是开箱即用的）。因此，使用字典可以为不同的消费者提供更多控制。我在创建总线时注册了自定义序列化程序，如下所示：

_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 管理 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 的工作原理。