如何：启用流处理

项目
04/07/2023

Windows Communication Foundation (WCF) 可以使用缓冲传输或流传输发送消息。在默认的缓冲传输模式中，只有在一条消息全部传递完之后，接收方才能读取该消息。在流传输模式中，不必等到消息全部传递完，接收方便可以开始处理该消息。当传递的信息很长且可以依次处理时，流模式非常有用。当消息过长以致于无法全部缓冲时，流模式也非常有用。

若要启用流处理，请适当地定义 OperationContract 并在传输级别上启用流处理。

对数据进行流处理

若要对数据进行流处理，服务的 OperationContract 必须满足两个要求：
1. 保留要进行流处理的数据的参数必须是方法中的唯一参数。例如，如果要对输入消息进行流处理，则该操作必须正好具有一个输入参数。同样，如果要对输出消息进行流处理，则该操作必须正好具有一个输出参数或一个返回值。
2. 参数和返回值的类型中至少有一个必须是 Stream, Message 或 IXmlSerializable。
下面是流处理数据的协定的示例。
```
[ServiceContract(Namespace = "http://Microsoft.ServiceModel.Samples")]
public interface IStreamingSample
{
    [OperationContract]
    Stream GetStream(string data);
    [OperationContract]
    bool UploadStream(Stream stream);
    [OperationContract]
    Stream EchoStream(Stream stream);
    [OperationContract]
    Stream GetReversedStream();
}
```
```
<ServiceContract(Namespace:="http://Microsoft.ServiceModel.Samples")> _
Public Interface IStreamingSample
    <OperationContract()> _
    Function GetStream(ByVal data As String) As Stream
    <OperationContract()> _
    Function UploadStream(ByVal stream As Stream) As Boolean
    <OperationContract()> _
    Function EchoStream(ByVal stream As Stream) As Stream
    <OperationContract()> _
    Function GetReversedStream() As Stream

End Interface
```
GetStream 操作接收一些缓冲输入数据作为经过缓冲的 string，并返回经过流处理的 Stream。相反，UploadStream 接收一个经过流处理的 Stream，并返回一个经过缓冲的 bool。 EchoStream 是输入消息和输出消息都经过流处理的操作的示例，它接收和返回 Stream。最后，GetReversedStream 将不接收输入，并返回一个 Stream（已经过流处理）。

必须在绑定上启用流处理。设置 TransferMode 属性，可以采用下面的值之一：

Buffered，
Streamed，此值在两个方向上启用流通信。
StreamedRequest，此值仅启用请求流处理。
StreamedResponse，此值仅启用响应流处理。

BasicHttpBinding 公开绑定上的 TransferMode 属性，与 NetTcpBinding 和 NetNamedPipeBinding 一样。还可以在传输绑定元素上设置 TransferMode 属性，并且在自定义绑定中使用。

下面的示例演示如何通过代码和通过更改配置文件来设置 TransferMode。两个示例都将 maxReceivedMessageSize 属性设置为 64 MB，这是可以接收的最大消息大小。默认的 maxReceivedMessageSize 为 64 KB，对于流处理方案而言，此值通常太低。根据应用程序期望接收的消息的最大大小适当地设置此配额设置。还请注意，maxBufferSize 可控制缓冲的最大大小，应适当地进行设置。

下面示例中的配置段演示如何将 TransferMode 和自定义 HTTP 绑定上的 basicHttpBinding 属性设置为流处理。

<basicHttpBinding>
  <binding name="HttpStreaming" maxReceivedMessageSize="67108864"
           transferMode="Streamed"/>
</basicHttpBinding>
<!-- an example customBinding using Http and streaming-->
<customBinding>
  <binding name="Soap12">
    <textMessageEncoding messageVersion="Soap12WSAddressing10" />
    <httpTransport transferMode="Streamed" maxReceivedMessageSize="67108864"/>
  </binding>
</customBinding>

下面的代码段演示如何将 TransferMode 和自定义 HTTP 绑定上的 basicHttpBinding 属性设置为流处理。

public static Binding CreateStreamingBinding()
{
    BasicHttpBinding b = new BasicHttpBinding();
    b.TransferMode = TransferMode.Streamed;
    return b;
}

Public Shared Function CreateStreamingBinding() As Binding
    Dim b As New BasicHttpBinding()
    b.TransferMode = TransferMode.Streamed
    Return b
End Function

下面的代码段演示如何将自定义 TCP 绑定上的 TransferMode 属性设置为流处理。

public static Binding CreateStreamingBinding()
{
    TcpTransportBindingElement transport = new TcpTransportBindingElement();
    transport.TransferMode = TransferMode.Streamed;
    BinaryMessageEncodingBindingElement encoder = new BinaryMessageEncodingBindingElement();
    CustomBinding binding = new CustomBinding(encoder, transport);
    return binding;
}

Public Shared Function CreateStreamingBinding() As Binding
    Dim transport As New TcpTransportBindingElement()
    transport.TransferMode = TransferMode.Streamed
    Dim binding As New CustomBinding(New BinaryMessageEncodingBindingElement(), _
                                     transport)
    Return binding
End Function

GetStream、UploadStream 和 EchoStream 操作都可以处理直接从文件发送数据或将接收的数据直接保存到文件的情况。下面的代码适用于 GetStream。

public Stream GetStream(string data)
{
    //this file path assumes the image is in
    // the Service folder and the service is executing
    // in service/bin
    string filePath = Path.Combine(
        System.Environment.CurrentDirectory,
        ".\\..\\image.jpg");
    //open the file, this could throw an exception
    //(e.g. if the file is not found)
    //having includeExceptionDetailInFaults="True" in config
    // would cause this exception to be returned to the client
    try
    {
        FileStream imageFile = File.OpenRead(filePath);
        return imageFile;
    }
    catch (IOException ex)
    {
        Console.WriteLine(
            String.Format("An exception was thrown while trying to open file {0}", filePath));
        Console.WriteLine("Exception is: ");
        Console.WriteLine(ex.ToString());
        throw ex;
    }
}

Public Function GetStream(ByVal data As String) As Stream Implements IStreamingSample.GetStream
    'this file path assumes the image is in
    ' the Service folder and the service is executing
    ' in service/bin 
    Dim filePath = Path.Combine(System.Environment.CurrentDirectory, ".\..\image.jpg")
    'open the file, this could throw an exception 
    '(e.g. if the file is not found)
    'having includeExceptionDetailInFaults="True" in config 
    ' would cause this exception to be returned to the client
    Try
        Return File.OpenRead(filePath)
    Catch ex As IOException
        Console.WriteLine(String.Format("An exception was thrown while trying to open file {0}", filePath))
        Console.WriteLine("Exception is: ")
        Console.WriteLine(ex.ToString())
        Throw ex
    End Try
End Function

编写自定义流

若要在发送或接收数据流的每个块区时对其进行特殊处理，可从 Stream 派生一个自定义流类。与自定义流的示例一样，下面的代码包含 GetReversedStream 方法和 ReverseStream 类。

GetReversedStream 创建并返回 ReverseStream 的新实例。当系统从 ReverseStream 对象中读取时，发生实际处理。 ReverseStream.Read 方法从基础文件中读取字节块区，反转字节，然后返回反转的字节。此方法不会反转整个文件内容；一次只能反转一个字节块区。本示例演示在从流中读取内容或将内容写入到流中时如何执行流处理。

class ReverseStream : Stream
{

    FileStream inStream;
    internal ReverseStream(string filePath)
    {
        //opens the file and places a StreamReader around it
        inStream = File.OpenRead(filePath);
    }
    public override bool CanRead
    {
        get { return inStream.CanRead; }
    }

    public override bool CanSeek
    {
        get { return false; }
    }

    public override bool CanWrite
    {
        get { return false; }
    }

    public override void Flush()
    {
        throw new Exception("This stream does not support writing.");
    }

    public override long Length
    {
        get { throw new Exception("This stream does not support the Length property."); }
    }

    public override long Position
    {
        get
        {
            return inStream.Position;
        }
        set
        {
            throw new Exception("This stream does not support setting the Position property.");
        }
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        int countRead = inStream.Read(buffer, offset, count);
        ReverseBuffer(buffer, offset, countRead);
        return countRead;
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        throw new Exception("This stream does not support seeking.");
    }

    public override void SetLength(long value)
    {
        throw new Exception("This stream does not support setting the Length.");
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        throw new Exception("This stream does not support writing.");
    }
    public override void Close()
    {
        inStream.Close();
        base.Close();
    }
    protected override void Dispose(bool disposing)
    {
        inStream.Dispose();
        base.Dispose(disposing);
    }
    void ReverseBuffer(byte[] buffer, int offset, int count)
    {

        int i, j;

        for (i = offset, j = offset + count - 1; i < j; i++, j--)
        {
            byte currenti = buffer[i];
            buffer[i] = buffer[j];
            buffer[j] = currenti;
        }
    }
}

Friend Class ReverseStream
    Inherits Stream

    Private inStream As FileStream

    Friend Sub New(ByVal filePath As String)
        'opens the file and places a StreamReader around it
        inStream = File.OpenRead(filePath)
    End Sub

    Public Overrides ReadOnly Property CanRead() As Boolean
        Get
            Return inStream.CanRead
        End Get
    End Property

    Public Overrides ReadOnly Property CanSeek() As Boolean
        Get
            Return False
        End Get
    End Property

    Public Overrides ReadOnly Property CanWrite() As Boolean
        Get
            Return False
        End Get
    End Property

    Public Overrides Sub Flush()
        Throw New Exception("This stream does not support writing.")
    End Sub

    Public Overrides ReadOnly Property Length() As Long
        Get
            Throw New Exception("This stream does not support the Length property.")
        End Get
    End Property

    Public Overrides Property Position() As Long
        Get
            Return inStream.Position
        End Get
        Set(ByVal value As Long)
            Throw New Exception("This stream does not support setting the Position property.")
        End Set
    End Property

    Public Overrides Function Read(ByVal buffer() As Byte, _
                                   ByVal offset As Integer, _
                                   ByVal count As Integer) As Integer

        Dim countRead = inStream.Read(buffer, _
                                      offset, _
                                      count)
        ReverseBuffer(buffer, _
                      offset, _
                      countRead)
        Return countRead
    End Function

    Public Overrides Function Seek(ByVal offset As Long, _
                                   ByVal origin As SeekOrigin) As Long
        Throw New Exception("This stream does not support seeking.")
    End Function

    Public Overrides Sub SetLength(ByVal value As Long)
        Throw New Exception("This stream does not support setting the Length.")
    End Sub

    Public Overrides Sub Write(ByVal buffer() As Byte, _
                               ByVal offset As Integer, _
                               ByVal count As Integer)
        Throw New Exception("This stream does not support writing.")
    End Sub

    Public Overrides Sub Close()
        inStream.Close()
        MyBase.Close()
    End Sub

    Protected Overloads Overrides Sub Dispose(ByVal disposing As Boolean)
        inStream.Dispose()
        MyBase.Dispose(disposing)
    End Sub

    Private Sub ReverseBuffer(ByVal buffer() As Byte, _
                              ByVal offset As Integer, _
                              ByVal count As Integer)

        Dim i = offset
        Dim j = offset + count - 1

        Do While i < j
            Dim currenti = buffer(i)
            buffer(i) = buffer(j)
            buffer(j) = currenti
            i += 1
            j -= 1
        Loop

    End Sub
End Class

Share via

如何：启用流处理

对数据进行流处理

编写自定义流

另请参阅

其他资源