C#中使用async和await实现异步Udp通讯的示例代码

在之前的C#版本中, 如果我们想要进行异步的Udp, 需要单开线程接收消息, C#7.1开始, 我们可以使用async/await关键字来编写异步代码, 我们今天一起来探索怎么实现.

C/S架构

我们要实现两个app, 一个客户端和一个服务器, 各自都可以发消息和收消息.
发消息很简单, 收消息的话需要一直在端口上监听.

udp相比tcp来说简单了很多, 不需要一直保持连接, 也不需要处理发送回调, 因为udp不可靠, 只要发了就不管, 丢了也与我无关. 而且因为不需要保证顺序, 所以没有发送缓存, 只要请求发送, 立马就发, 收到的包也不会堆积, 肯定是整包, 所以我们也不需要处理粘包问题.

整个实现的关键点有:

• Sockets.socket: socket类, tcp和udp共用.

• System.Net.IPEndPoint: 端口类, tcp和udp共用.

• Sockets.socket.Bind: 绑定本地端口方法, 主要是服务器使用.

• Sockets.socket.Create: 绑定远端端口方法, 主要是客户端使用.

• Sockets.socket.SendTo: 向指定端口发送数据, 主要是服务器使用.

• Sockets.socket.ReceiveFrom: 从指定端口接收数据, 主要是服务器使用.

• Sockets.socket.Send: 从绑定的端口发送数据, 主要是客户端使用.

• Sockets.socket.Receive: 从绑定的端口接收数据, 主要是客户端使用.

• async 关键字: 标识方法为异步方法.

• await 关键字: 标识异步执行方法, 等待返回.

• System.Threading.Tasks.Task: 异步任务类

客户端实现

我们先来研究客户端, 服务器的实现大致相同, 只是有细微的差别.

客户端主流程和实现

// 构建socket对象
Socket udpSocket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);

// 连接远端口, 用于向远端发送消息, 这里是自己的机器同时当服务器和客户端, 所以都是127...
// 注意这里的连接只是将`socket`对象与ip和端口绑定, 不是tcp中的连接概念.
// 内部会分配新的本地端口, 发送给远端, 供远端使用
var endPoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 8060);
udpSocket.Connect(endPoint);

// 发送消息
SendMessageToServer("客户端说:Hello Server!");

// 监听消息
StartRecvMessage();
Console.ReadKey();

客户端发送消息实现

static void SendMessageToServer(string message)
{
   udpSocket.Send(Encoding.UTF8.GetBytes(message));
}

因为之前已经和远端口绑定了, 所以客户端可以直接发送消息, 在内部会自动分配一个客户端自己的本地端口, 服务器端使用这个端口来向本客户端发送消息, 我们会在服务器实现中看到.

客户端监听消息实现

// 从byte中转换string
static string ConverBytesToString(Decoder decoder, byte[] bytes, int len)
{
    var nchar = decoder.GetCharCount(bytes, 0, len);

    var bytesChar = new char[nchar];
    nchar = decoder.GetChars(bytes, 0, len, bytesChar, 0);

    var result = new string(bytesChar, 0, nchar);
    return result;
}

// 从连接的端口接收消息, 返回读取到的字节数
static int SocketRecvMessage()
{
    var nread = udpSocket.Receive(buffer);
    return nread;
}

// 开始异步接收消息
static async void StartRecvMessage()
{
    Console.WriteLine("客户端开始监听: " + udpSocket.LocalEndPoint);
    var decoder8 = Encoding.UTF8.GetDecoder();

    while (true)
    {
        var nread = await Task.Run<int>(SocketRecvMessage);
        var message = ConverBytesToString(decoder8, buffer, nread);

        Console.WriteLine($"收到来自服务器的消息: {message}");
    }
}

上面的代码中, 主要的部分是:

async/await/Task.Run<int>(xxx):

• async:标识方法StartRecvMessage将采用异步方式执行

• await: 标识要等待的操作, 而这种操作是需要耗时的, 比如socket, io等, 也可以是单纯就是要等待多久(Task.Delay(500); // 等待500ms).

• Task.Run<int>(xxx): 将耗时的操作包装为异步任务(类似开了一个线程来执行该操作).

udpSocket.Receive(buffer): 从连接好的远端口接收消息, 这是一个阻断性的操作, 在消息回来之前会停留在这里不动.

上面的异步还能写成下面的形式, 只是将耗时操作推迟到了更具体的操作而已:

// 从连接的端口接收消息, 返回读取到的字节数
static async Task<int> SocketRecvMessage()
{
    var nread = await Task.Run<int>(() => udpSocket.Receive(buffer));
    return nread;
}

// 开始异步接收消息
static async void StartRecvMessage()
{
    Console.WriteLine("客户端开始监听: " + udpSocket.LocalEndPoint);
    var decoder8 = Encoding.UTF8.GetDecoder();

    while (true)
    {
        var nread = await SocketRecvMessage();
        var message = ConverBytesToString(decoder8, buffer, nread);

        Console.WriteLine($"收到来自服务器的消息: {message}");
    }
}

我们还能进一步简化代码:

// 开始异步接收消息
static async void StartRecvMessage()
{
    Console.WriteLine("客户端开始监听: " + udpSocket.LocalEndPoint);
    var decoder8 = Encoding.UTF8.GetDecoder();

    while (true)
    {
        var nread = await Task.Run<int>(() => udpSocket.Receive(buffer));
        var message = ConverBytesToString(decoder8, buffer, nread);

        Console.WriteLine($"收到来自服务器的消息: {message}");
    }
}

服务器实现

服务器和客户端的实现差别很小.

主要区别在于服务器针对的是很多客户端, 所以在收发消息上对于端口的处理不一样.

服务器主流程和实现

// 构建socket对象
Socket udpSocket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);

// 绑定本地端口, 监听来自于各个客户端的消息
var endPoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 8060);
udpSocket.Bind(endPoint);

// 监听消息
StartRecvMessage();
Console.ReadKey();

服务器发送消息实现

// 向指定的客户端端口发送消息
// 注意这里和客户端的实现不一样, 还是因为服务器会对应多个客户端, 所以每次发送都需要指明目的地
static void SendMessageToClient(EndPoint endPoint, string message)
{
    udpSocket.SendTo(Encoding.UTF8.GetBytes(message), endPoint);
}

服务器监听消息实现

static (int, EndPoint) SocketRecvMessage()
{
    EndPoint endPoint = new IPEndPoint(IPAddress.Any, 0);

    var nread = udpSocket.ReceiveFrom(buffer, ref endPoint);
    return (nread, endPoint);
}

static async void StartRecvMessage()
{
    Console.WriteLine("服务器开始监听: " + udpSocket.LocalEndPoint);
    var decoder8 = Encoding.UTF8.GetDecoder();

    while(true)
    {
        var (nread, endPoint) = await Task.Run<(int, EndPoint)>(SocketRecvMessage);
        var message = ConverBytesToString(decoder8, buffer, nread);

        Console.WriteLine($"收到来自客户端[{endPoint}]的消息: {message}");

        SendMessageToClient(endPoint, "服务器对你说Hi!");
    }
}

上面的代码中, 主要的差别在对于端口的处理上:

• SocketRecvMessage返回的是一个元组(int, EndPoint): 即读取到的字节数, 还有客户端的端口信息.

• ReceiveFrom: 接收消息指定了端口, 服务器每次接收消息都要使用端口信息用来标识发送消息的客户端.

优化过后的代码为:

static async void StartRecvMessage()
{
    Console.WriteLine("服务器开始监听: " + udpSocket.LocalEndPoint);
    var decoder8 = Encoding.UTF8.GetDecoder();

    while(true)
    {
        EndPoint endPoint = new IPEndPoint(IPAddress.Any, 0);
        var nread = await Task.Run<int>(() => udpSocket.ReceiveFrom(buffer, ref endPoint));
        var message = ConverBytesToString(decoder8, buffer, nread);

        Console.WriteLine($"收到来自客户端[{endPoint}]的消息: {message}");

        SendMessageToClient(endPoint, "服务器对你说Hi!");
    }
}

下面是完整的代码:

// --- AsyncUdpClient.cs

using System;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;

namespace test
{
    class AsyncUdpClient
    {
        static Socket udpSocket;
        static byte[] buffer = new byte[4096];

        public static void Main()
        {
            udpSocket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);

            var endPoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 8060);
            //udpSocket.Bind(endPoint);
            udpSocket.Connect(endPoint);

            SendMessageToServer("客户端说:Hello Server!");
            StartRecvMessage();

            Console.ReadKey();
        }

        static void SendMessageToServer(string message)
        {
            udpSocket.Send(Encoding.UTF8.GetBytes(message));
        }

        static async void StartRecvMessage()
        {
            Console.WriteLine("客户端开始监听: " + udpSocket.LocalEndPoint);
            var decoder8 = Encoding.UTF8.GetDecoder();

            while (true)
            {
                var nread = await Task.Run<int>(() => udpSocket.Receive(buffer));
                var message = ConverBytesToString(decoder8, buffer, nread);

                Console.WriteLine($"收到来自服务器的消息: {message}");

                #region 交互
                Console.WriteLine("是否继续监听?[yes|no]");
                var str = await Task.Run<string>(() => Console.ReadLine());
                if (str == "yes")
                {
                    Console.WriteLine("继续监听...");
                    continue;
                }

                Console.WriteLine("客户端停止监听.");
                return;
                #endregion
            }
        }

        static string ConverBytesToString(Decoder decoder, byte[] bytes, int len)
        {
            var nchar = decoder.GetCharCount(bytes, 0, len);

            var bytesChar = new char[nchar];
            nchar = decoder.GetChars(bytes, 0, len, bytesChar, 0);

            var result = new string(bytesChar, 0, nchar);
            return result;
        }
    }
}

// --- AsyncUdpServer.cs

using System;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;

namespace test
{
    static class AsyncUdpServer
    {
        static Socket udpSocket;
        static byte[] buffer = new byte[4096];

        public static void Main()
        {
            udpSocket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);

            var endPoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 8060);
            udpSocket.Bind(endPoint);
            //udpSocket.Connect(endPoint);

            StartRecvMessage();
            Console.ReadKey();
        }

        static void SendMessageToClient(EndPoint endPoint, string message)
        {
            udpSocket.SendTo(Encoding.UTF8.GetBytes(message), endPoint);
        }

        static async void StartRecvMessage()
        {
            Console.WriteLine("服务器开始监听: " + udpSocket.LocalEndPoint);
            var decoder8 = Encoding.UTF8.GetDecoder();

            while(true)
            {
                EndPoint endPoint = new IPEndPoint(IPAddress.Any, 0);
                var nread = await Task.Run<int>(() => udpSocket.ReceiveFrom(buffer, ref endPoint));
                var message = ConverBytesToString(decoder8, buffer, nread);

                Console.WriteLine($"收到来自客户端[{endPoint}]的消息: {message}");

                SendMessageToClient(endPoint, "服务器对你说Hi!");

#region 交互
                Console.WriteLine("是否继续监听?[yes|no]");
                var str = await Task.Run<string>(()=> Console.ReadLine());
                if (str == "yes")
                {
                    Console.WriteLine("继续监听...");
                    continue;
                }

                Console.WriteLine("服务器停止监听.");
                return;
#endregion

            }
        }

        static string ConverBytesToString(Decoder decoder, byte[] bytes, int len)
        {
            var nchar = decoder.GetCharCount(bytes, 0, len);

            var bytesChar = new char[nchar];
            nchar = decoder.GetChars(bytes, 0, len, bytesChar, 0);

            var result = new string(bytesChar, 0, nchar);
            return result;
        }
    }
}

来源：https://blog.csdn.net/woodengm/article/details/125929812