2

我有一些实验代码基本上只是试图使一个简单的场景工作。我有一个客户端将数据流式传输到多个服务。我遇到的问题是，如果其中一项服务没有正常关闭，我会得到一个似乎无法处理的 EndpointNotFoundException。以下是我处理失败的尝试。实际上，我想从频道列表中删除失败的服务频道，并继续将数据流式传输到仍在运行的服务。计时器只是让服务有机会在数据流开始之前启动。

let prices = returns a seq of data that is streamed.

type ReplayDataStream(prices) =
  let evt = new Event<_>()
  member x.Replay() = 
                    async { for line, delay in prices do
                                do! Async.Sleep(delay)
                                evt.Trigger(line) }
                                |> Async.StartImmediate

  member x.PriceChanged = evt.Publish


let main() =
    let addresses = new ResizeArray<EndpointAddress>()

    let announcementService = new AnnouncementService()

    let createChannels addresses =
        let channels = new ResizeArray<IInputDataService>()
        for (address:EndpointAddress) in addresses do
                let channelFactory = new ChannelFactory<IInputDataService>(new BasicHttpBinding(), address)
                let channel = channelFactory.CreateChannel()
                (channel :?> ICommunicationObject).Faulted.Add(fun x -> 
                                                                        (channel :?> ICommunicationObject).Abort()
                                                                        channels.Remove(channel) |> ignore
                                                               )
                channels.Add(channel)
        channels

    let sendMessage(args:ElapsedEventArgs) =
        let channels = createChannels addresses
        for financialDataStream in prices do
        let replayDataStreamA = new ReplayDataStream(financialDataStream)
        for channel in channels do
            try
            //This is where it blows up and the try block isn't catching the exception.
            replayDataStreamA.PriceChanged.Add(channel.InputStringData)
            with
            | :? EndpointNotFoundException as ex -> Console.WriteLine(ex.ToString())
            | :? CommunicationException as ex -> Console.WriteLine(ex.ToString())
            | :? Exception as ex -> Console.WriteLine(ex.ToString())
            replayDataStreamA.Replay()

    let timer = new System.Timers.Timer()
    timer.Enabled <- true
    timer.AutoReset <- false
    timer.Interval <- 30000.0
    timer.Start()
    timer.Elapsed.Add(sendMessage)

    announcementService.OnlineAnnouncementReceived.Add(fun e -> 
                                                                Console.WriteLine(e.EndpointDiscoveryMetadata.Address)
                                                                addresses.Add(e.EndpointDiscoveryMetadata.Address)
                                                                )

    announcementService.OfflineAnnouncementReceived.Add(fun e -> 
                                                                Console.WriteLine(e.EndpointDiscoveryMetadata.Address)
                                                                addresses.Remove(e.EndpointDiscoveryMetadata.Address) |> ignore
                                                                )

    let announcementServiceHost = new ServiceHost(announcementService)
    try
        announcementServiceHost.AddServiceEndpoint(new UdpAnnouncementEndpoint());
        announcementServiceHost.Open();
    with 
    | :? System.ServiceModel.CommunicationException as ex -> Console.WriteLine(ex.ToString())
    | :? System.TimeoutException as ex -> Console.WriteLine(ex.ToString())


    printfn "%s" "Hit any key to close."
    Console.ReadKey() |> ignore
4

2 回答 2

2

在用 C# 重写我的代码之后，我终于明白我做错了什么。这就是 PriceChanged 事件处理程序的外观。我需要在 lambda 本身中捕获异常。现在我需要编写一些实际上看起来像生产代码的东西。:)

replayDataStreamA.PriceChanged.Add( fun x -> 
                                                            try
                                                            channel.InputStringData x
                                                            with 
                                                            | :? System.ServiceModel.CommunicationException as ex -> (channel :?> ICommunicationObject).Abort()
                                                            )

对于后代，这里是整个方法：

let sendMessage(args:ElapsedEventArgs) =
            if(addresses.Count > 0) then
                for address in addresses do
                    let channelFactory = new ChannelFactory<IInputDataService>(new BasicHttpBinding(), address)
                    let channel = channelFactory.CreateChannel()
                    for financialDataStream in prices do
                    let replayDataStreamA = new ReplayDataStream(financialDataStream)
                    replayDataStreamA.PriceChanged.Add( fun x -> 
                                                        try
                                                        channel.InputStringData x
                                                        with 
                                                        | :? System.ServiceModel.CommunicationException as ex -> (channel :?> ICommunicationObject).Abort()
                                                        )
                    replayDataStreamA.Replay()
于 2011-05-27T19:56:05.130 回答
0

Sky Sanders 的解释很有意义，应该适用于这种情况。这是博客的链接。

为 Faulted 事件提供订阅者与在异常处理程序中调用 channel.Abort() 做的事情并不完全相同。

PriceChanged.Add() 等价于 PriceChanged += ：您正在为处理程序订阅 Price changed 事件。放置一个 try/with 块将捕获订阅时抛出的异常（想想您的事件中的自定义添加/删除实现），这不是您想要的。您正在寻找一种在调用 InputStringData 时处理异常的方法。这个思考过程自然会导致你的解决方案。

在 C# 生产代码中，在事件端引发异常的点周围放置一个 try/catch 块。通过重新抛出捕获订阅者和 Debug.Assert 抛出的异常，警告开发人员应在订阅者端处理所有异常。在您的代码中，这意味着在 evt.Trigger() 处发出警告并重新抛出的 try/with 块。

您可以公开异步块，而不是在声明时运行它。这应该为您提供更高级别的编排能力：在 sendMessage 中。有一个特殊的 API 可以在一个中心位置捕获异常、处理取消和超时，这确实值得研究。

于 2011-05-27T03:55:40.360 回答