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

目录

• C/S架构
• 客户端实现
• 客户端主流程和实现
• 客户端发送消息实现
• 客户端监听消息实现
• 服务器实现
• 服务器主流程和实现
• 服务器发送消息实现
• 服务器监听消息实现
• 总结

在之前的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;
        }
    }
}

总结

今天我们使用aync/await关键字实现了异步的udp通讯.

主要是了解和实践异步关键字的知识和使用, 同时对传统的单开线程来进行udp通讯方式进行了优化, 这

样的好处是不需要自己维护多线程环境, 不需要保证线程安全, 各种锁之类的操作.

udp通讯本身很简单, 只要搞清楚Bind, Connect还有端口的概念即可.

aync/await对于长期写同步代码或者使用异步callback形式回调的同学来说, 可能会有一定的理解困难,

但是其实也就那么回事, 我们简单理解为协程即可(只是比协程使用起来更方便).

到此这篇关于C#中使用async和await实现异步Udp通讯的示例代码的文章就介绍到这了,更多相关C# 异步Udp通讯内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！