在 Azure Service Fabric 服务中设置 SignalR 和服务总线

Question

我正在将现有的 Cloud Service WorkerRole 作为无状态服务移植到 Service Fabric。原始云服务使用 SignalR 和服务总线（作为 SignalR 背板）将通知发送到任何正在侦听的客户端。有一个 Startup 类可以进行一些设置：

class Startup
{
    public void Configuration(IAppBuilder app)
    {
        String connectionString = "Endpoint=sb://[name].servicebus.windows.net/;SharedSecretIssuer=owner;SharedSecretValue=[key]";
        GlobalHost.DependencyResolver.UseServiceBus(connectionString, "InSys");
        app.MapSignalR();
        Notifications.Hub = GlobalHost.ConnectionManager.GetHubContext<MyHub>();
    }
}

在 WorkerRole 的 OnStart() 方法中，我使用以下方法启动 OWIN：

var endpoint = RoleEnvironment.CurrentRoleInstance.InstanceEndpoints["HttpEndpoint"];
var baseUri = $"{endpoint.Protocol}://{endpoint.IPEndpoint}";
var app = WebApp.Start<Startup>(new StartOptions(url: baseUri));

如何为 Service Fabric 中的无状态服务完成此操作（即连接到 SignalR 服务总线背板）？

Answer 1

在https://github.com/marcinbudny/SignalRSelfHostScaleOut 的帮助下（这是一个使用 Redis 进行横向扩展的示例），我想我已经完成了这项工作。

在 ServiceManifest.xml 我添加了以下端点：

<Endpoint Protocol="http" Name="ServiceEndpoint" Type="Input" Port="8322" />

我还添加了一个 Startup 类：

public static class Startup
{
    public static void ConfigureApp(IAppBuilder app)
    {
        String connectionString = "Endpoint=sb://[name].servicebus.windows.net/;SharedSecretIssuer=owner;SharedSecretValue=[value]";
        GlobalHost.DependencyResolver.UseServiceBus(connectionString, "InSys");
        app.MapSignalR();
        Notifications.Hub = GlobalHost.ConnectionManager.GetHubContext<InSysMainHub>();
    }
}

还添加了一个 OwinCommunicationListener 类：

public class OwinCommunicationListener : ICommunicationListener
{
    private readonly ServiceEventSource eventSource;
    private readonly Action<IAppBuilder> startup;
    private readonly ServiceContext serviceContext;
    private readonly string endpointName;
    private readonly string appRoot;

    private IDisposable webApp;
    private string publishAddress;
    private string listeningAddress;

    public OwinCommunicationListener(Action<IAppBuilder> startup, ServiceContext serviceContext, ServiceEventSource eventSource, string endpointName)
        : this(startup, serviceContext, eventSource, endpointName, null)
    {
    }

    public OwinCommunicationListener(Action<IAppBuilder> startup, ServiceContext serviceContext, ServiceEventSource eventSource, string endpointName, string appRoot)
    {
        if (startup == null)
        {
            throw new ArgumentNullException(nameof(startup));
        }

        if (serviceContext == null)
        {
            throw new ArgumentNullException(nameof(serviceContext));
        }

        if (endpointName == null)
        {
            throw new ArgumentNullException(nameof(endpointName));
        }

        if (eventSource == null)
        {
            throw new ArgumentNullException(nameof(eventSource));
        }

        this.startup = startup;
        this.serviceContext = serviceContext;
        this.endpointName = endpointName;
        this.eventSource = eventSource;
        this.appRoot = appRoot;
    }


    public Task<string> OpenAsync(CancellationToken cancellationToken)
    {
        var serviceEndpoint = this.serviceContext.CodePackageActivationContext.GetEndpoint(this.endpointName);
        var protocol = serviceEndpoint.Protocol;
        int port = serviceEndpoint.Port;

        if (this.serviceContext is StatefulServiceContext)
        {
            StatefulServiceContext statefulServiceContext = (StatefulServiceContext) serviceContext;

            listeningAddress = string.Format(
                CultureInfo.InvariantCulture,
                "{0}://+:{1}/{2}{3}/{4}/{5}",
                protocol,
                port,
                string.IsNullOrWhiteSpace(appRoot)
                    ? string.Empty
                    : appRoot.TrimEnd('/') + '/',
                statefulServiceContext.PartitionId,
                statefulServiceContext.ReplicaId,
                Guid.NewGuid());
        }
        else if (serviceContext is StatelessServiceContext)
        {
            listeningAddress = string.Format(
                CultureInfo.InvariantCulture,
                "{0}://+:{1}/{2}",
                protocol,
                port,
                string.IsNullOrWhiteSpace(appRoot)
                    ? string.Empty
                    : appRoot.TrimEnd('/') + '/');
        }
        else
        {
            throw new InvalidOperationException();
        }

        publishAddress = listeningAddress.Replace("+", FabricRuntime.GetNodeContext().IPAddressOrFQDN);

        try
        {
            eventSource.Message("Starting web server on " + listeningAddress);
            webApp = WebApp.Start(listeningAddress, appBuilder => startup.Invoke(appBuilder));
            eventSource.Message("Listening on " + this.publishAddress);
            return Task.FromResult(this.publishAddress);
        }
        catch (Exception ex)
        {
            eventSource.Message("Web server failed to open endpoint {0}. {1}", this.endpointName, ex.ToString());
            StopWebServer();
            throw;
        }
    }

    public Task CloseAsync(CancellationToken cancellationToken)
    {
        this.eventSource.Message("Closing web server on endpoint {0}", this.endpointName);

        this.StopWebServer();

        return Task.FromResult(true);
    }

    public void Abort()
    {
        this.eventSource.Message("Aborting web server on endpoint {0}", this.endpointName);

        this.StopWebServer();
    }

    private void StopWebServer()
    {
        if (this.webApp != null)
        {
            try
            {
                this.webApp.Dispose();
            }
            catch (ObjectDisposedException)
            {
                // no-op
            }
        }
    }
}

最后我将无状态服务代码中的 CreateServiceInstanceListeners 方法更改为：

protected override IEnumerable<ServiceInstanceListener> CreateServiceInstanceListeners()
    {
        return new[]
        {
            new ServiceInstanceListener(serviceContext => new OwinCommunicationListener(Startup.ConfigureApp, serviceContext, ServiceEventSource.Current, "ServiceEndpoint"))
        };
    }

Answer 2

使用 Owin 侦听器创建无状态服务。然后在启动时配置 signalR 和背板（服务总线或 sql）。理想情况下，您将面临的问题是协商（Signalr 客户端与服务器之间的握手）此时尝试配置跨源请求，持久连接的示例代码如下所示。

还请注意 appBuilder.UseAesDataProtectorProvider("Your Key") 行，因为它很重要。这样做的结果是，您在大多数情况下最终都不会获得 HTTP 400 连接。这是因为 SignalR 会在握手时发出至少 2 个请求，而这些请求通常会到达两台不同的机器。

感谢marcin budny的解释。

var config = new HttpConfiguration();

// Configure your origins as required.

var cors = new EnableCorsAttribute("*", "*", "*");

config.EnableCors(cors);

FormatterConfig.ConfigureFormatters(config.Formatters);

RouteConfig.RegisterRoutes(config.Routes);

appBuilder.UseWebApi(config);

GlobalHost.DependencyResolver.UseServiceBus("yourconnection string comes here", "signalrbackplaneserver");
appBuilder.UseAesDataProtectorProvider("some password");
appBuilder.Map("/echo", map =>
{
                map.UseCors(CorsOptions.AllowAll).RunSignalR<MyEndPoint>();
});