带有rabbitmq的masstransit：为什么消息在响应时自动移动到***_skipped队列

Question

UCITY.API 请求 UCITY.USER 的“ucity_us_mobilephonecomplete”队列并接收消息，但 Masstransit 自动将其移至跳过队列

但其他队列工作正常

轨道交通记录：

2018-03-23 15:59:22.5727|DEBUG|MassTransit.Messages|SEND rabbitmq://192.168.1.142/us/ucity_us_mobilephonecomplete 0b350000-dafb-1866-5b48-08d59093fd36 MQNamespace.USER.IMobilephoneCompleteRequest

2018-03-23 15:59:22.6077|DEBUG|MassTransit.Messages|SKIP rabbitmq://192.168.1.142/us/bus-WIN-VH8418VRIC5-UCITY.API-bc4oyyg49ccgcqg6bdk3br7trf?durable=false&autodelete=true N/A

消息详情

他的服务器报告剩余 1 条消息。

Exchange    bus-WIN-VH8418VRIC5-UCITY.API-bc4oyyg49ccgcqg6bdk3br7trf_skipped
Routing Key 
Redelivered ●
Properties  
message_id: 0b350000-dafb-1866-d0af-08d59093bae2
delivery_mode:  2
headers:    
Content-Type:   application/vnd.masstransit+json
publishId:  1
MT-Reason:  dead-letter
MT-Host-MachineName:    WIN-VH8418VRIC5
MT-Host-ProcessName:    UCITY.API
MT-Host-ProcessId:  14648
MT-Host-Assembly:   UCITY.API
MT-Host-AssemblyVersion:    1.0.0.0
MT-Host-MassTransitVersion: 4.0.1.1390
MT-Host-FrameworkVersion:   4.0.30319.42000
MT-Host-OperatingSystemVersion: Microsoft Windows NT 6.2.9200.0
content_type:   application/vnd.masstransit+json
Payload
1060 bytes
Encoding: string
{

    "messageId": "0b350000-dafb-1866-d0af-08d59093bae2",

    "requestId": "0b350000-dafb-1866-8267-08d59093bade",

    "conversationId": "0b350000-dafb-1866-0ddf-08d59093bae2",

    "sourceAddress": "rabbitmq://192.168.1.142/us/ucity_us_mobilephonecomplete",

    "destinationAddress": "rabbitmq://192.168.1.142/us/bus-WIN-VH8418VRIC5-UCITY.API-bc4oyyg49ccgcqg6bdk3br7trf?durable=false&autodelete=true",

    "messageType": [

        "urn:message:MQNamespace:IBaseResponseModel"

    ],

    "message": {

        "code": 4212,

        "correlationId": "73989f86-51fd-4ae1-a467-a556a165f125",

        "message": "COMPLETEMOBILEPHONE FAILD!USERWECHAT MOBILEPHONE COMPLETED",

        "timeStamp": "2018-03-23T15:57:31.3218406+08:00"

    },

    "headers": {},

    "host": {

        "machineName": "WIN-VH8418VRIC5",

        "processName": "UCITY.USER",

        "processId": 8952,

        "assembly": "UCITY.USER",

        "assemblyVersion": "1.0.0.0",

        "frameworkVersion": "4.0.30319.42000",

        "massTransitVersion": "4.0.1.1390",

        "operatingSystemVersion": "Microsoft Windows NT 6.2.9200.0"
    }
}

跳过的队列有什么作用 谢谢回答

Answer 1

更改此方法以使用正确的响应类型，而不是基本接口。

public async Task Consume(ConsumeContext<IMobilephoneCompleteRequest> context)
{
    var dispatch = new UserDispatch(context.Message.CorrelationId);
    try
    {
        var response = await dispatch.MobilephoneComplete(context.Message);
        return context.RespondAsync(response);
    }
    catch (UCException ucEx)
    {
        var response = new AuthenticationResponse(context.Message.CorrelationId)
        {
            Code = ucEx.Code,
            Message = ucEx.Message
        };
        return context.RespondAsync(response);
    }
    catch (Exception ex)
    {
        var response = new AuthenticationResponse(context.Message.CorrelationId)
        {
            Code = (int)EnumErrorStatus.UserError,
            Message = ex.Message
        };
        Console.WriteLine(JsonConvert.SerializeObject(ex));
        return context.RespondAsync(response);
    }
}

Answer 2

我也遇到了同样的问题。在我发现问题后的第二天，原因似乎是对当前程序集使用了 Autofac 注册方法，例如：

builder.RegisterConsumers(Assembly.GetExecutingAssembly());

使用以下（每个消费者的显式注册）解决了绑定问题：

builder.RegisterType<YouConsumerHandler>();

Answer 3

UCIT.API 代码

//execute request
public async Task<ResultOpenUserLogin> MobilephoneComplete(AuthenticationIdentity authenticationIdentity, ArguUserMobilephoneComplete arguUserMobilephoneComplete)
    {
        var response = await UserMobilephoneCompleteRequest.Client.Request(new UserMobilephoneCompleteRequest(CorrelationId)
        {
            DeviceType = authenticationIdentity.DeviceType,
            IdUser = authenticationIdentity.IdUser,
            Mobilephone = arguUserMobilephoneComplete.Mobilephone,
            Passcode = arguUserMobilephoneComplete.Passcode
        });// the request timeout
        if (response.Code != 200)
        {
            throw new UCException(response.Code, response.Message);
        }
        return new ResultOpenUserLogin()
        {
            Token = response.Token,
            UserInfoCompleted = !string.IsNullOrWhiteSpace(response.Mobilephone)
        };
    }


//UserMobilephoneCompleteRequest class detail
[RabbitMqUri(VirtualHost = "ucity_us", Queue = "ucity_us_mobilephonecomplete")]
public class UserMobilephoneCompleteRequest : BaseRequest, IMobilephoneCompleteRequest
{
    public UserMobilephoneCompleteRequest(Guid? correlationId) : base(correlationId)
    {

    }
    public int DeviceType { get; set; }

    public int IdUser { get; set; }

    public string Mobilephone { get; set; }

    public string Passcode { get; set; }

    public static IRequestClient<IMobilephoneCompleteRequest, IAuthenticationResponse> Client
    {
        get
        {
            return _requestClient;
        }
    }

    private static IRequestClient<IMobilephoneCompleteRequest, IAuthenticationResponse> _requestClient
    {
        get; set;
    }

    public override void CreateClient(IBusControl busControl, Uri address)
    {
        _requestClient = busControl.CreateRequestClient<IMobilephoneCompleteRequest, IAuthenticationResponse>(address, TimeSpan.FromSeconds(10));
    }
}

UCIT.USER 代码

//these codes work well
 [RabbitMqUri(VirtualHost = "ucity_us", Queue = "ucity_us_mobilephonecomplete")]
public class MobilephoneCompleteConsumer : BaseConsumer<MobilephoneCompleteConsumer>, IConsumer<IMobilephoneCompleteRequest>
{
    public async Task Consume(ConsumeContext<IMobilephoneCompleteRequest> context)
    {
        var dispatch = new UserDispatch(context.Message.CorrelationId);
        IBaseResponseModel response = null;
        try
        {
            response = await dispatch.MobilephoneComplete(context.Message);
        }
        catch (UCException ucEx)
        {
            response = new AuthenticationResponse(context.Message.CorrelationId)
            {
                Code = ucEx.Code,
                Message = ucEx.Message
            };
        }
        catch (Exception ex)
        {
            response = new AuthenticationResponse(context.Message.CorrelationId)
            {
                Code = (int)EnumErrorStatus.UserError,
                Message = ex.Message
            };
            Console.WriteLine(Newtonsoft.Json.JsonConvert.SerializeObject(ex));
        }
        await context.RespondAsync(response);
    }
}