基于.NET 7 的 QUIC 实现 Echo 服务的详细过程_Asp.net

前言

随着今年6月份的 http/3 协议的正式发布，它背后的网络传输协议 quic，凭借其高效的传输效率和多路并发的能力，也大概率会取代我们熟悉的使用了几十年的 tcp，成为互联网的下一代标准传输协议。

在去年 .net 6 发布的时候，已经可以看到 http/3 和 quic 支持的相关内容了，但是当时 http/3 的 rfc 还没有定稿，所以也只是预览功能，而 quic 的 api 也没有在 .net 6 中公开。

在最新的 .net 7 中，.net 团队公开了 quic api，它是基于 msquic 库来实现的，提供了开箱即用的支持，命名空间为 system.net.quic。

quic api

下面的内容中，我会介绍如何在 .net 中使用 quic。

下面是 system.net.quic 命名空间下，比较重要的几个类。

quicconnection

表示一个 quic 连接，本身不发送也不接收数据，它可以打开或者接收多个quic 流。

quiclistener

用来监听入站的 quic 连接，一个 quiclistener 可以接收多个 quic 连接。

quicstream

表示 quic 流，它可以是单向的（quicstreamtype.unidirectional），只允许创建方写入数据，也可以是双向的（quicstreamtype.bidirectional），它允许两边都可以写入数据。

小试牛刀

下面是一个客户端和服务端应用使用 quic 通信的示例。

1.分别创建了 quicclient 和 quicserver 两个控制台程序。

项目的版本为 .net 7，并且设置 enablepreviewfeatures = true。

下面创建了一个 quiclistener，监听了本地端口 9999，指定了 alpn 协议版本。

console.writeline("quic server running...");

// 创建 quiclistener
var listener = await quiclistener.listenasync(new quiclisteneroptions
{ 
    applicationprotocols = new list<sslapplicationprotocol> { sslapplicationprotocol.http3  },
    listenendpoint = new ipendpoint(ipaddress.loopback,9999), 
    connectionoptionscallback = (connection,ssl, token) => valuetask.fromresult(new quicserverconnectionoptions()
    {
        defaultstreamerrorcode = 0,
        defaultcloseerrorcode = 0,
        serverauthenticationoptions = new sslserverauthenticationoptions()
        {
            applicationprotocols = new list<sslapplicationprotocol>() { sslapplicationprotocol.http3 },
            servercertificate = generatemanualcertificate()
        }
    }) 
});

因为 quic 需要 tls 加密，所以要指定一个证书，generatemanualcertificate 方法可以方便地创建一个本地的测试证书。

x509certificate2 generatemanualcertificate()
{
    x509certificate2 cert = null;
    var store = new x509store("kestrelwebtransportcertificates", storelocation.currentuser);
    store.open(openflags.readwrite);
    if (store.certificates.count > 0)
    {
        cert = store.certificates[^1];

        // rotate key after it expires
        if (datetime.parse(cert.getexpirationdatestring(), null) < datetimeoffset.utcnow)
        {
            cert = null;
        }
    }
    if (cert == null)
    {
        // generate a new cert
        var now = datetimeoffset.utcnow;
        subjectalternativenamebuilder sanbuilder = new();
        sanbuilder.adddnsname("localhost");
        using var ec = ecdsa.create(eccurve.namedcurves.nistp256);
        certificaterequest req = new("cn=localhost", ec, hashalgorithmname.sha256);
        // adds purpose
        req.certificateextensions.add(new x509enhancedkeyusageextension(new oidcollection
        {
            new("1.3.6.1.5.5.7.3.1") // serverauth

        }, false));
        // adds usage
        req.certificateextensions.add(new x509keyusageextension(x509keyusageflags.digitalsignature, false));
        // adds subject alternate names
        req.certificateextensions.add(sanbuilder.build());
        // sign
        using var crt = req.createselfsigned(now, now.adddays(14)); // 14 days is the max duration of a certificate for this
        cert = new(crt.export(x509contenttype.pfx));

        // save
        store.add(cert);
    }
    store.close();

    var hash = sha256.hashdata(cert.rawdata);
    var certstr = convert.tobase64string(hash);
    //console.writeline($"\n\n\n\n\ncertificate: {certstr}\n\n\n\n"); // <-- you will need to put this output into the js api call to allow the connection
    return cert;
}

阻塞线程，直到接收到一个 quic 连接，一个 quiclistener 可以接收多个连接。

var connection = await listener.acceptconnectionasync();

console.writeline($"client [{connection.remoteendpoint}]: connected");

接收一个入站的 quic 流, 一个 quicconnection 可以支持多个流。

var stream = await connection.acceptinboundstreamasync();

console.writeline($"stream [{stream.id}]: created");

接下来，使用 system.io.pipeline 处理流数据，读取行数据，并回复一个 ack 消息。

console.writeline();

await processlinesasync(stream);

console.readkey();      

// 处理流数据
async task processlinesasync(quicstream stream)
{
    var reader = pipereader.create(stream);  
    var writer = pipewriter.create(stream);

    while (true)
    {
        readresult result = await reader.readasync();
        readonlysequence<byte> buffer = result.buffer;

        while (tryreadline(ref buffer, out readonlysequence<byte> line))
        {
            // 读取行数据
            processline(line);

            // 写入 ack 消息
            await writer.writeasync(encoding.utf8.getbytes($"ack: {datetime.now.tostring("hh:mm:ss")} \n"));
        } 
      
        reader.advanceto(buffer.start, buffer.end);
 
        if (result.iscompleted)
        {
            break;
        } 
    }

    console.writeline($"stream [{stream.id}]: completed");

    await reader.completeasync();  
    await writer.completeasync();    
} 

bool tryreadline(ref readonlysequence<byte> buffer, out readonlysequence<byte> line)
{ 
    sequenceposition? position = buffer.positionof((byte)'\n');

    if (position == null)
    {
        line = default;
        return false;
    } 
    
    line = buffer.slice(0, position.value);
    buffer = buffer.slice(buffer.getposition(1, position.value));
    return true;
} 

void processline(in readonlysequence<byte> buffer)
{
    foreach (var segment in buffer)
    {
        console.writeline("recevied -> " + system.text.encoding.utf8.getstring(segment.span));
    }

    console.writeline();
}

以上就是服务端的完整代码了。

接下来我们看一下客户端 quicclient 的代码。

直接使用 quicconnection.connectasync 连接到服务端。

console.writeline("quic client running...");

await task.delay(3000);

// 连接到服务端
var connection = await quicconnection.connectasync(new quicclientconnectionoptions
{
    defaultcloseerrorcode = 0,
    defaultstreamerrorcode = 0,
    remoteendpoint = new ipendpoint(ipaddress.loopback, 9999),
    clientauthenticationoptions = new sslclientauthenticationoptions
    {
        applicationprotocols = new list<sslapplicationprotocol> { sslapplicationprotocol.http3 },
        remotecertificatevalidationcallback = (sender, certificate, chain, errors) =>
        {
            return true;
        }
    }
});

创建一个出站的双向流。

// 打开一个出站的双向流
var stream = await connection.openoutboundstreamasync(quicstreamtype.bidirectional); 

var reader = pipereader.create(stream);
var writer = pipewriter.create(stream);

后台读取流数据，然后循环写入数据。

// 后台读取流数据
_ = processlinesasync(stream);

console.writeline(); 

// 写入数据
for (int i = 0; i < 7; i++)
{
    await task.delay(2000);

    var message = $"hello quic {i} \n";

    console.write("send -> " + message);  

    await writer.writeasync(encoding.utf8.getbytes(message)); 
}

await writer.completeasync(); 

console.readkey();

processlinesasync 和服务端一样，使用 system.io.pipeline 读取流数据。

async task processlinesasync(quicstream stream)
{
    while (true)
    {
        readresult result = await reader.readasync();
        readonlysequence<byte> buffer = result.buffer;

        while (tryreadline(ref buffer, out readonlysequence<byte> line))
        { 
            // 处理行数据
            processline(line);
        }
     
        reader.advanceto(buffer.start, buffer.end); 
     
        if (result.iscompleted)
        {
            break;
        }
    }

    await reader.completeasync();
    await writer.completeasync();

} 

bool tryreadline(ref readonlysequence<byte> buffer, out readonlysequence<byte> line)
{ 
    sequenceposition? position = buffer.positionof((byte)'\n');

    if (position == null)
    {
        line = default;
        return false;
    }
 
    line = buffer.slice(0, position.value);
    buffer = buffer.slice(buffer.getposition(1, position.value));
    return true;
}

void processline(in readonlysequence<byte> buffer)
{
    foreach (var segment in buffer)
    {
        console.write("recevied -> " + system.text.encoding.utf8.getstring(segment.span));
        console.writeline();
    }

    console.writeline();
}

到这里，客户端和服务端的代码都完成了，客户端使用 quic 流发送了一些消息给服务端，服务端收到消息后在控制台输出，并回复一个 ack 消息，因为我们创建了一个双向流。

程序的运行结果如下

我们上面说到了一个 quicconnection 可以创建多个流，并行传输数据。

改造一下服务端的代码，支持接收多个 quic 流。

var cts = new cancellationtokensource();

while (!cts.iscancellationrequested)
{
    var stream = await connection.acceptinboundstreamasync();

    console.writeline($"stream [{stream.id}]: created");

    console.writeline();

    _ = processlinesasync(stream); 
} 

console.readkey();

对于客户端，我们用多个线程创建多个 quic 流，并同时发送消息。

默认情况下，一个 quic 连接的流的限制是 100，当然你可以设置 quicconnectionoptions 的 maxinboundbidirectionalstreams 和 maxinboundunidirectionalstreams 参数。

for (int j = 0; j < 5; j++)
{
    _ = task.run(async () => {

        // 创建一个出站的双向流
        var stream = await connection.openoutboundstreamasync(quicstreamtype.bidirectional); 
      
        var writer = pipewriter.create(stream); 

        console.writeline();
 
        await task.delay(2000);
        
        var message = $"hello quic [{stream.id}] \n";

        console.write("send -> " + message);

        await writer.writeasync(encoding.utf8.getbytes(message));

        await writer.completeasync(); 
    });  
}

最终程序的输出如下