NServiceBus向单端点发消息并在另一端点等待回复的实现咨询

针对无共享依赖的双服务架构下的两个核心问题，可直接按以下方案落地：

问题1：service1不感知响应具体类型、仅提取特定字段的处理方案 #

• 放弃两端共享强类型消息契约的依赖模式，利用消息传输的结构化特性做松散反序列化。RabbitMQ传输的消息本质是序列化后的结构化文本（默认JSON格式），只要字段名匹配即可完成目标字段提取，完全不需要感知响应的完整类型定义。
• service1侧仅定义包含所需特定字段的最小契约，比如只包含你需要提取的RequestId、Result、ProcessTime等字段的接口或浅类，不需要引用service2的任何消息代码。
• 关闭NServiceBus默认的强类型消息校验：序列化配置中关闭类型名称绑定校验，以Newtonsoft序列化为例，配置为TypeNameHandling.None，避免因两端消息类型的程序集名、命名空间不一致导致反序列化报错。
• 接收响应时可直接用JObject（Newtonsoft）/JsonNode（System.Text.Json）作为接收类型，拿到原始结构化对象后直接读取目标字段即可，无需解析整个响应结构。

问题2：异步API等待匹配响应的实现方案 #

该需求完全可实现，不需要从零开发匹配逻辑，NServiceBus原生能力即可支撑请求-响应的一一对应，具体实现思路如下：

• 首先纠正消息发送模式的选型：请求场景不要用Publish()（事件广播，会发给所有订阅方，易产生重复处理），改用Send()做点对点投递，更符合请求响应的语义。如果要求响应必须固定投递到endpoint2(queue2)，只需在发送请求的SendOptions中调用SetReplyTo("endpoint2")即可指定响应投递目标。
• 利用消息头的关联ID天然做请求-响应匹配：NServiceBus发送消息时会自动生成全局唯一的MessageId，同时在消息头写入NServiceBus.CorrelationId标记；service2处理完请求调用Reply()返回响应时，会自动把请求里的CorrelationId原样写回响应消息头，不需要手动传递这个标识。
• 不要自行实现WaitingReply()做全局队列阻塞等待，直接用NServiceBus原生的IMessageSession.Request<TResponse>()异步API：该API内部会维护一个以CorrelationId为键的并发字典，每个发送出去的请求会绑定一个带超时的TaskCompletionSource，当监听endpoint2的处理器收到响应时，会根据响应头的CorrelationId找到对应的等待任务，自动触发结果返回，天然保证请求和响应一一匹配，不会出现串包问题。
• 结合问题1的松散反序列化能力，Request()方法的泛型参数直接用JObject/自定义最小字段契约即可，不需要依赖service2的响应类型。

参考实现代码 #

// 服务启动时NServiceBus基础配置
var endpointConfig = new EndpointConfiguration("Service1Instance");
var transport = endpointConfig.UseTransport<RabbitMQTransport>();
transport.ConnectionString("RabbitMQ连接字符串");
// 配置请求路由：MyRequest类型消息发往endpoint1对应的queue1
var routing = transport.Routing();
routing.RouteToEndpoint(typeof(MyRequest), "endpoint1");
// 关闭强类型校验，支持松散反序列化
endpointConfig.UseSerialization<NewtonsoftSerializer>()
    .Settings(s => s.TypeNameHandling = TypeNameHandling.None);
endpointConfig.SendFailedMessagesTo("error_queue");
var messageSession = await Endpoint.Start(endpointConfig);

// 异步API方法实现
public async Task<object> SendRequest(string str, CancellationToken ct)
{
    var request = new MyRequest(str);
    var sendOptions = new SendOptions();
    sendOptions.SetDestination("endpoint1");
    sendOptions.SetReplyTo("endpoint2"); // 指定响应固定投递到endpoint2
    // 发送请求并等待匹配响应，设置10秒超时避免无限挂起
    var response = await messageSession.Request<JObject>(request, sendOptions, ct)
        .WaitAsync(TimeSpan.FromSeconds(10), ct);
    
    // 直接提取需要的特定字段返回，无需感知完整响应结构
    return new {
        RequestId = response.Value<string>("RequestId"),
        Success = response.Value<bool>("Success"),
        Payload = response.Value<string>("ResultData")
    };
}

额外注意事项 #

• 若需要完全自行实现响应匹配逻辑（比如自定义队列消费逻辑），只需维护一个ConcurrentDictionary<string, TaskCompletionSource<JObject>>：发请求时将生成的MessageId作为键存入TCS，收到响应时从消息头取出CorrelationId，找到对应TCS调用SetResult()即可，匹配逻辑非常轻量。
• 必须配置请求超时，避免service2宕机、消息丢失导致接口永久挂起。
• 多实例部署场景下，只要保证CorrelationId正确透传，不会出现多实例间响应串流的问题。

内容的提问来源于stack exchange，提问作者Nikita Kondratev