如何添加活动

Question

我用RabbitMq总线创建了服务。现在，我正在尝试在公共汽车上添加活动(我尝试了我找到的每一个注册选项)

    private void ConfigureBus(IServiceCollectionConfigurator configurator)
    {
        configurator.AddActivity(typeof(StepAcitivity));
        configurator.AddActivitiesFromNamespaceContaining<StepAcitivity>();
        configurator.AddActivity<StepAcitivity, StepArguments, StepLog>();
        configurator.AddConsumer<PingConsumer>();
    }

我还在集装箱注册了StepAcitivity：

    services.AddScoped<StepAcitivity>();

来自另一个服务:首先:测试按预期工作的ping请求/响应：

    IRequestClient<IPingRequest, IPingResponse> client =
                   busControl.CreateRequestClient<IPingRequest, IPingResponse>(serviceAddress, TimeSpan.FromSeconds(10));
                IPingResponse s = await client.Request(request);

第二:创建和执行活动

    string address = config.GetValue<string>("executorAddress");
    Uri sendAddress = new Uri($"queue:{address}");
    RoutingSlipBuilder builder = new RoutingSlipBuilder(NewId.NextGuid());
    
    builder.AddActivity("SetepA", sendAddress);
    builder.AddActivity("StepB", sendAddress);
    builder.AddActivity("StepC", sendAddress);
    RoutingSlip routingSlip = builder.Build();
    return routingSlip;

执行：

    await bus.Execute(routingSlip);

ping请求/响应按预期工作的问题，但routingSlip并没有失败，但我在活动端没有接到任何调用。在rabbitMQ管理中，可以看到消息被移动到WorkerService_skipped队列中。

，

1. ，我在工人服务部门的活动注册中缺少了什么？(总线看起来不错，因为我可以运行post IPingRequest请求并在同一个地址上读取响应)如果消息移动到跳过的队列中，为什么对await bus.Execute(routingSlip)的调用没有下降？我的编辑版本是: 6.2.3

我的员工服务探测器：

    {
      "resultId": "a4984523-5ffc-4ff4-8ceb-498029dbfeff",
      "probeId": "561351df-42da-44a7-83ce-29a467966663",
      "startTimestamp": "2020-07-06T12:51:32.0106524Z",
      "duration": "00:00:00.0427169",
      "host": {
        "machineName": "SLAVAF-LAP",
        "processName": "WorkerService",
        "processId": 38212,
        "assembly": "WorkerService",
        "assemblyVersion": "1.0.0.0",
        "frameworkVersion": "3.1.5",
        "greenPipesVersion": "3.0.1.0",
        "operatingSystemVersion": "Microsoft Windows NT 6.2.9200.0"
      },
      "results": {
        "bus": {
          "address": "rabbitmq://localhost:0/SLAVAFLAP_WorkerService_bus_x37yyyd4rz1m1wcfbdcndk4enp?temporary=true",
          "host": {
            "Type": "RabbitMQ",
            "Host": "localhost",
            "Port": 5672,
            "VirtualHost": "/",
            "Username": "slava",
            "Password": "****",
            "Heartbeat": 0,
            "Ssl": false,
            "connected": true,
            "receiveEndpoint": [
              {
                "name": "WorkerService",
                "started": true,
                "transport": {
                  "type": "RabbitMQ",
                  "ExchangeName": "WorkerService",
                  "ExchangeArguments": [],
                  "ExchangeType": "fanout",
                  "Durable": true,
                  "AutoDelete": false,
                  "QueueName": "WorkerService",
                  "Exclusive": false,
                  "QueueArguments": [
                    [
                      "x-queue-type",
                      {}
                    ]
                  ],
                  "BindingArguments": [],
                  "PrefetchCount": 16,
                  "PurgeOnStartup": false,
                  "ExclusiveConsumer": false,
                  "NoAck": false,
                  "BindQueue": true,
                  "ConsumeArguments": [],
                  "topology": {
                    "exchange": [
                      {
                        "Name": "WorkerService",
                        "Type": "fanout",
                        "Durable": true,
                        "AutoDelete": false
                      },
                      {
                        "Name": "Common.Messages.Ping:IPingRequest",
                        "Type": "fanout",
                        "Durable": true,
                        "AutoDelete": false
                      }
                    ],
                    "queue": {
                      "Name": "WorkerService",
                      "Durable": true,
                      "AutoDelete": false,
                      "Exclusive": false,
                      "argument": {
                        "key": "x-queue-type",
                        "value": "quorum"
                      }
                    },
                    "exchange-binding": {
                      "Source": "Common.Messages.Ping:IPingRequest",
                      "Destination": "WorkerService"
                    },
                    "queue-binding": {
                      "Source": "WorkerService",
                      "Destination": "WorkerService"
                    }
                  }
                },
                "filters": [
                  {
                    "filterType": "deadLetter",
                    "filters": {
                      "filterType": "dead-letter"
                    }
                  },
                  {
                    "filterType": "rescue",
                    "filters": [
                      {
                        "filterType": "generateFault"
                      },
                      {
                        "filterType": "moveFault"
                      }
                    ]
                  },
                  {
                    "filterType": "deserialize",
                    "deserializers": {
                      "json": {
                        "contentType": "application/vnd.masstransit+json"
                      },
                      "bson": {
                        "contentType": "application/vnd.masstransit+bson"
                      },
                      "xml": {
                        "contentType": "application/vnd.masstransit+xml"
                      }
                    },
                    "consumePipe": {
                      "filters": {
                        "filterType": "dispatchPipe",
                        "outputType": "MassTransit.ConsumeContext<Common.Messages.Ping.IPingRequest>",
                        "filters": {
                          "filterType": "MessageFilter",
                          "Type": "Common.Messages.Ping.IPingRequest"
                        },
                        "consumer": {
                          "type": "WorkerService.Consumers.PingConsumer",
                          "consumerFactory": {
                            "source": "scope",
                            "consumerType": "WorkerService.Consumers.PingConsumer",
                            "provider": "dependencyInjection"
                          },
                          "filters": {
                            "filterType": "split",
                            "ConsumerType": "WorkerService.Consumers.PingConsumer",
                            "filters": {
                              "filterType": "ScopedLoggingFilter"
                            }
                          },
                          "consume": {
                            "method": "Consume(ConsumeContext<Common.Messages.Ping.IPingRequest> context)"
                          }
                        }
                      }
                    }
                  }
                ]
              },
              {
                "name": "SLAVAFLAP_WorkerService_bus_x37yyyd4rz1m1wcfbdcndk4enp",
                "started": true,
                "transport": {
                  "type": "RabbitMQ",
                  "ExchangeName": "SLAVAFLAP_WorkerService_bus_x37yyyd4rz1m1wcfbdcndk4enp",
                  "ExchangeArguments": [],
                  "ExchangeType": "fanout",
                  "Durable": false,
                  "AutoDelete": true,
                  "QueueName": "SLAVAFLAP_WorkerService_bus_x37yyyd4rz1m1wcfbdcndk4enp",
                  "Exclusive": false,
                  "QueueExpiration": "00:01:00",
                  "QueueArguments": [],
                  "BindingArguments": [],
                  "PrefetchCount": 16,
                  "PurgeOnStartup": false,
                  "ExclusiveConsumer": false,
                  "NoAck": false,
                  "BindQueue": true,
                  "ConsumeArguments": [],
                  "topology": {
                    "exchange": {
                      "Name": "SLAVAFLAP_WorkerService_bus_x37yyyd4rz1m1wcfbdcndk4enp",
                      "Type": "fanout",
                      "Durable": false,
                      "AutoDelete": true
                    },
                    "queue": {
                      "Name": "SLAVAFLAP_WorkerService_bus_x37yyyd4rz1m1wcfbdcndk4enp",
                      "Durable": false,
                      "AutoDelete": false,
                      "Exclusive": false,
                      "argument": {
                        "key": "x-expires",
                        "value": 60000
                      }
                    },
                    "queue-binding": {
                      "Source": "SLAVAFLAP_WorkerService_bus_x37yyyd4rz1m1wcfbdcndk4enp",
                      "Destination": "SLAVAFLAP_WorkerService_bus_x37yyyd4rz1m1wcfbdcndk4enp"
                    }
                  }
                },
                "filters": [
                  {
                    "filterType": "deadLetter",
                    "filters": {
                      "filterType": "dead-letter"
                    }
                  },
                  {
                    "filterType": "rescue",
                    "filters": [
                      {
                        "filterType": "generateFault"
                      },
                      {
                        "filterType": "moveFault"
                      }
                    ]
                  },
                  {
                    "filterType": "deserialize",
                    "deserializers": {
                      "json": {
                        "contentType": "application/vnd.masstransit+json"
                      },
                      "bson": {
                        "contentType": "application/vnd.masstransit+bson"
                      },
                      "xml": {
                        "contentType": "application/vnd.masstransit+xml"
                      }
                    },
                    "consumePipe": {}
                  }
                ]
              }
            ]
          }
        }
      }
    }

Answer 1

每个活动都需要一个单独的端点来执行，如果活动支持补偿，则需要一个额外的补偿端点。您不能在同一个端点上配置您的活动，并且不应该与其他服务共享端点。

使用AddActivity()就足以配置容器，您不需要在活动中调用AddScoped。

此外，您还可以使用ConfigureEndpoints并传递注册上下文，为所有活动、sagas和消费者配置总线端点。

您所包含的探测(很好，谢谢)显示您只有一个接收端点，WorkerService，它有一个使用者来处理IPingRequest消息.