Socket 异步

摘要：

　　System.Net.Sockets.Sockte 类有一组增强功能，提供可供专用的高性能套接字应用程序使用的可选异步模式，SocketAsyncEventArgs 类就是这一组增强功能的一部分。 该类专为需要高性能的网络服务器应用程序而设计。 应用程序可以完全使用增强的异步模式，也可以仅仅在目标热点区域（例如，在接收大量数据时）使用此模式。

　　这些增强功能的主要特点是可以避免在异步套接字 I/O 量非常大时发生重复的对象分配和同步。

　　在新的 System.Net.Sockets.Sockte 类增强功能中，异步套接字操作由分配的可重用 SocketAsyncEventArgs 对象描述并由应用程序维护。 高性能套接字应用程序非常清楚地知道必须保持的重叠的套接字操作的量。 应用程序可以根据自身需要创建任意多的SocketAsyncEventArgs 对象。 例如，如果服务器应用程序需要总是有 15 个未完成的套接字接收操作来支持传入客户端连接率，那么可以为此分配 15 个可重用的 SocketAsyncEventArgs 对象。

　　使用此类执行异步套接字操作的模式包含以下步骤：

  1. 分配一个新的 SocketAsyncEventArgs 上下文对象，或者从应用程序池中获取一个空闲的此类对象。
  2. 将该上下文对象的属性设置为要执行的操作（例如，完成回调方法、数据缓冲区、缓冲区偏移量以及要传输的最大数据量）。
  3. 调用适当的套接字方法 (xxxAsync) 以启动异步操作。
  4. 如果异步套接字方法 (xxxAsync) 返回 true，则在回调中查询上下文属性来获取完成状态。
  5. 如果异步套接字方法 (xxxAsync) 返回 false，则说明操作是同步完成的。 可以查询上下文属性来获取操作结果。
  6. 将该上下文重用于另一个操作，将它放回到应用程序池中，或者将它丢弃。

　　新的异步套接字操作上下文对象的生命周期由应用程序代码引用和异步 I/O 引用决定。 在对异步套接字操作上下文对象的引用作为一个参数提交给某个异步套接字操作方法之后，应用程序不必保留该引用。 在完成回调返回之前将一直引用该对象。 但是，应用程序保留对上下文的引用是有好处的，这样该引用就可以重用于将来的异步套接字操作。

　　下面的代码示例实现使用 SocketAsyncEventArgs 类的套接字服务器的连接逻辑。 接受连接之后，从客户端读取的所有数据都将发回客户端。 客户端模式的读取和回送会一直继续到客户端断开连接

　一、服务端　

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

namespace ConsoleTest
{
    public class Server
    {
        /// <summary>
        /// 同时处理的最大连接数
        /// </summary>
        private int m_numConnections;

        /// <summary>
        /// 每个Socket IO可使用的缓冲区大小
        /// </summary>
        private int m_receiveBufferSize;

        BufferManager m_bufferManager;
        const int opsToPreAlloc = 2;
        Socket listenSocket;
        SocketAsyncEventArgsPool m_readWritePool;

        /// <summary>
        /// 服务端接受的总数据大小
        /// </summary>
        int m_totalBytesRead;

        /// <summary>
        /// 连接到服务端的客户端数量
        /// </summary>
        int m_numConnectedSockets;
        Semaphore m_maxNumberAcceptedClients;

        public Server(int numConnections, int receiveBufferSize)
        {
            m_totalBytesRead = 0;
            m_numConnectedSockets = 0;
            m_numConnections = numConnections;
            m_receiveBufferSize = receiveBufferSize;
            m_bufferManager = new BufferManager(receiveBufferSize * numConnections * opsToPreAlloc, receiveBufferSize);
            m_readWritePool = new SocketAsyncEventArgsPool(numConnections);
            m_maxNumberAcceptedClients = new Semaphore(numConnections, numConnections);
        }

        public void Init()
        {
            m_bufferManager.InitBuffer();
            SocketAsyncEventArgs readWriteEventArg;
            for (int i = 0; i < m_numConnections; i++)
            {
                readWriteEventArg = new SocketAsyncEventArgs();
                readWriteEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
                readWriteEventArg.UserToken = new AsyncUserToken();
                m_bufferManager.SetBuffer(readWriteEventArg);
                m_readWritePool.Push(readWriteEventArg);
            }
        }

        /// <summary>
        /// 开始监听
        /// </summary>
        public void Start(IPEndPoint localEndPoint)
        {
            listenSocket = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
            listenSocket.Bind(localEndPoint);
            listenSocket.Listen(100);

            //开始监听客户端连接
            StartAccept(null);

            Console.WriteLine("按任意键终止服务器...");
            Console.ReadKey();
        }

        /// <summary>
        /// 处理来自客户机的连接请求
        /// </summary>
        /// <param name="acceeptEventArg"></param>
        public void StartAccept(SocketAsyncEventArgs acceptEventArg)
        {
            if (acceptEventArg == null)
            {
                acceptEventArg = new SocketAsyncEventArgs();
                acceptEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(AcceptEventArg_Completed);
            }
            else
            {
                acceptEventArg.AcceptSocket = null;
            }

            m_maxNumberAcceptedClients.WaitOne();
            //异步接收客户端连接，操作完成后触发AcceptEventArg_Completed事件
            bool willRaiseEvent = listenSocket.AcceptAsync(acceptEventArg);
            //如果是同步完成
            if (!willRaiseEvent)
            {
                ProcessAccept(acceptEventArg);
            }
        }

        /// <summary>
        /// 接收到数据后触发
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        void IO_Completed(object sender, SocketAsyncEventArgs e)
        {
            switch (e.LastOperation)
            {
                case SocketAsyncOperation.Receive:
                    ProcessReceive(e);
                    break;
                case SocketAsyncOperation.Send:
                    ProcessSend(e);
                    break;
                default:
                    throw new ArgumentException("The last operation completed on the socket was not a receive or send");
            }
        }

        /// <summary>
        /// 侦听到客户端连接时触发
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        void AcceptEventArg_Completed(object sender, SocketAsyncEventArgs e)
        {
            ProcessAccept(e);
        }

        /// <summary>
        ///
        /// </summary>
        /// <param name="e"></param>
        private void ProcessAccept(SocketAsyncEventArgs e)
        {
            Interlocked.Increment(ref m_numConnectedSockets);
            Console.WriteLine("已接受客户端连接. There are {0} clients connected to the server",
            m_numConnectedSockets);

            SocketAsyncEventArgs readEventArgs = m_readWritePool.Pop();

            ((AsyncUserToken)readEventArgs.UserToken).Socket = e.AcceptSocket;

            //异步接受数据，接受完成则触发AcceptEventArg_Completed事件
            bool willRaiseEvent = e.AcceptSocket.ReceiveAsync(readEventArgs);

            if (!willRaiseEvent)
            {
                ProcessReceive(readEventArgs);
            }
            //循环接受下一个客户端连接
            StartAccept(e);
        }

        private void ProcessReceive(SocketAsyncEventArgs e)
        {
            AsyncUserToken token = e.UserToken as AsyncUserToken;
            if (e.BytesTransferred > 0 && e.SocketError == SocketError.Success)
            {
                Interlocked.Add(ref m_totalBytesRead, e.BytesTransferred);
                Console.WriteLine("服务器已经接受 {0} 字节数据", m_totalBytesRead);

                string received = Encoding.Unicode.GetString(e.Buffer, e.Offset, e.BytesTransferred);

                for (int i = 73728; i < e.Buffer.Length; i++)
                {
                    byte o = e.Buffer[i];
                }

                Console.WriteLine(received);

                //异步发送是将Buffer空间的所有数据拷贝到基础系统发送缓冲区
                e.SetBuffer(e.Offset, e.BytesTransferred);
                string str1 = Encoding.Unicode.GetString(e.Buffer);
                //发送数据到客户端
                bool willRaiseEvent = token.Socket.SendAsync(e);
                if (!willRaiseEvent)
                {
                    ProcessSend(e);
                }
            }
            else
            {
                CloseClientSocket(e);
            }
        }

        private void ProcessSend(SocketAsyncEventArgs e)
        {
            if (e.SocketError == SocketError.Success)
            {
                AsyncUserToken token = (AsyncUserToken)e.UserToken;
                bool willRaiseEvent = token.Socket.ReceiveAsync(e);
                if (!willRaiseEvent)
                {
                    ProcessReceive(e);
                }
            }
            else
            {
                CloseClientSocket(e);
            }
        }

        private void CloseClientSocket(SocketAsyncEventArgs e)
        {
            AsyncUserToken token = e.UserToken as AsyncUserToken;

            try
            {
                token.Socket.Shutdown(SocketShutdown.Send);
            }
            catch (Exception) { }
            token.Socket.Close();

            Interlocked.Decrement(ref m_numConnectedSockets);

            m_maxNumberAcceptedClients.Release();
            Console.WriteLine("A client has been disconnected from the server. There are {0} clients connected to the server", m_numConnectedSockets);

            m_readWritePool.Push(e);
        }

    }
}

　　

using System.Net.Sockets;

namespace ConsoleTest
{
    class AsyncUserToken
    {
        Socket m_socket;

        public AsyncUserToken() : this(null) { }

        public AsyncUserToken(Socket socket)
        {
            m_socket = socket;
        }

        public Socket Socket
        {
            get { return m_socket; }
            set { m_socket = value; }
        }

    }
}

　　

using System.Collections.Generic;
using System.Net.Sockets;

namespace ConsoleTest
{
    /// <summary>
    /// 创建一个大型缓冲区，缓冲区可以进行分割并指定给 SocketAsyncEventArgs 对象以便用在每个套接字 I/O 操作中
    /// </summary>
    public class BufferManager
    {
        /// <summary>
        /// 缓冲区大小
        /// </summary>
        int m_numBytes;

        /// <summary>
        /// BufferManager维护的底层字节数组缓冲区
        /// </summary>
        byte[] m_buffer;

        /// <summary>
        /// 记录偏移量
        /// </summary>
        Stack<int> m_freeIndexPool;

        /// <summary>
        /// 缓存区当前位置
        /// </summary>
        int m_currentIndex;

        /// <summary>
        ///
        /// </summary>
        int m_bufferSize;

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="totalBytes">缓冲区字节总数</param>
        /// <param name="bufferSize">要缓存的数据</param>
        public BufferManager(int totalBytes, int bufferSize)
        {
            m_numBytes = totalBytes;
            m_currentIndex = 0;
            m_bufferSize = bufferSize;
            m_freeIndexPool = new Stack<int>();
        }

        /// <summary>
        /// 初始化缓冲区
        /// </summary>
        public void InitBuffer()
        {
            m_buffer = new byte[m_numBytes];
        }

        /// <summary>
        /// 设置要用于异步套接字方法的数据缓冲区。
        /// </summary>
        /// <param name="args"></param>
        /// <returns></returns>
        public bool SetBuffer(SocketAsyncEventArgs args)
        {
            if (m_freeIndexPool.Count > 0)
            {
                args.SetBuffer(m_buffer, m_freeIndexPool.Pop(), m_bufferSize);
            }
            else
            {
                //缓存区大小容量不够
                if ((m_numBytes - m_bufferSize) < m_currentIndex) { return false; }
                args.SetBuffer(m_buffer, m_currentIndex, m_bufferSize);
                m_currentIndex += m_bufferSize;
            }
            return true;
        }

        /// <summary>
        /// 释放缓冲区
        /// </summary>
        /// <param name="args"></param>
        public void FreeBuffer(SocketAsyncEventArgs args)
        {
            m_freeIndexPool.Push(args.Offset);
            args.SetBuffer(null, 0, 0);
        }
    }
}

　　

using System;
using System.Collections.Generic;
using System.Net.Sockets;

namespace ConsoleTest
{
    /// <summary>
    /// 代表一个可重用的SocketAsyncEventArgs对象的集合。
    /// </summary>
    public class SocketAsyncEventArgsPool
    {
        Stack<SocketAsyncEventArgs> m_pool;

        public SocketAsyncEventArgsPool(int capacity)
        {
            m_pool = new Stack<SocketAsyncEventArgs>(capacity);
        }

        /// <summary>
        /// 添加SocketAsyncEventArg实例到池
        /// </summary>
        /// <param name="item"></param>
        public void Push(SocketAsyncEventArgs item)
        {
            if (item == null) throw new ArgumentNullException("参数不能为空！");

            lock (m_pool)
            {
                m_pool.Push(item);
            }
        }

        /// <summary>
        /// 从池中删除SocketAsyncEventArgs实例
        /// </summary>
        /// <returns></returns>
        public SocketAsyncEventArgs Pop()
        {
            lock (m_pool)
            {
                return m_pool.Pop();
            }
        }

        /// <summary>
        /// 缓冲池中的实例个数
        /// </summary>
        public int Count
        {
            get { return m_pool.Count; }
        }
    }
}

　　

二、客户端调用：

using System.Collections.Generic;
using System.Net.Sockets;

namespace ConsoleTest
{
    /// <summary>
    /// 创建一个大型缓冲区，缓冲区可以进行分割并指定给 SocketAsyncEventArgs 对象以便用在每个套接字 I/O 操作中
    /// </summary>
    public class BufferManager
    {
        /// <summary>
        /// 缓冲区大小
        /// </summary>
        int m_numBytes;

        /// <summary>
        /// BufferManager维护的底层字节数组缓冲区
        /// </summary>
        byte[] m_buffer;

        /// <summary>
        /// 记录偏移量
        /// </summary>
        Stack<int> m_freeIndexPool;

        /// <summary>
        /// 缓存区当前位置
        /// </summary>
        int m_currentIndex;

        /// <summary>
        ///
        /// </summary>
        int m_bufferSize;

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="totalBytes">缓冲区字节总数</param>
        /// <param name="bufferSize">要缓存的数据</param>
        public BufferManager(int totalBytes, int bufferSize)
        {
            m_numBytes = totalBytes;
            m_currentIndex = 0;
            m_bufferSize = bufferSize;
            m_freeIndexPool = new Stack<int>();
        }

        /// <summary>
        /// 初始化缓冲区
        /// </summary>
        public void InitBuffer()
        {
            m_buffer = new byte[m_numBytes];
        }

        /// <summary>
        /// 设置要用于异步套接字方法的数据缓冲区。
        /// </summary>
        /// <param name="args"></param>
        /// <returns></returns>
        public bool SetBuffer(SocketAsyncEventArgs args)
        {
            if (m_freeIndexPool.Count > 0)
            {
                args.SetBuffer(m_buffer, m_freeIndexPool.Pop(), m_bufferSize);
            }
            else
            {
                //缓存区大小容量不够
                if ((m_numBytes - m_bufferSize) < m_currentIndex) { return false; }
                args.SetBuffer(m_buffer, m_currentIndex, m_bufferSize);
                m_currentIndex += m_bufferSize;
            }
            return true;
        }

        /// <summary>
        /// 释放缓冲区
        /// </summary>
        /// <param name="args"></param>
        public void FreeBuffer(SocketAsyncEventArgs args)
        {
            m_freeIndexPool.Push(args.Offset);
            args.SetBuffer(null, 0, 0);
        }
    }
}

　　

　　

时间： 2024-10-10 16:10:38

Socket 异步的相关文章

socket异步编程--libevent的使用

这篇文章介绍下libevent在socket异步编程中的应用.在一些对性能要求较高的网络应用程序中,为了防止程序阻塞在socket I/O操作上造成程序性能的下降,需要使用异步编程,即程序准备好读写的函数(或接口)并向系统注册,然后在需要的时候只向系统提交读写的请求之后就继续做自己的事情,实际的读写操作由系统在合适的时候调用我们程序注册的接口进行.异步编程会给一些程序猿带来一些理解和编写上的困难,因为我们通常写的一些简单的程序都是顺序执行的,而异步编程将程序的执行顺序打乱了,有些代码什么情况下执

Socket异步存储示例

异步客户端存储示例: using System; using System.Net; using System.Net.Sockets; using System.Threading; using System.Text; // State object for receiving data from remote device. public class StateObject { // Client socket. public Socket workSocket = null; // Si

libevent的使用(socket异步编程)

这篇文章介绍下libevent在socket异步编程中的应用.在一些对性能要求较高的网络应用程序中,为了防止程序阻塞在socket I/O操作上造成程序性能的下降,需要使用异步编程,即程序准备好读写的函数(或接口)并向系统注册,然后在需要的时候只向系统提交读写的请求之后就继续做自己的事情,实际的读写操作由系统在合适的时候调用我们程序注册的接口进行.异步编程会给一些程序猿带来一些理解和编写上的困难,因为我们通常写的一些简单的程序都是顺序执行的,而异步编程将程序的执行顺序打乱了,有些代码什么情况下执

socket 异步事件 WSAAsyncSelect 用法

WSAAsyncSelect 实现给异步socket给了另一种实现方式,就是通过窗口消息的方式来提醒对socket接收还是发送 msdn有非常全面的解释:https://msdn.microsoft.com/en-us/library/windows/desktop/ms741540(v=vs.85).aspx 本文几解释以下我认为难懂的几点: 1.The socket created by the accept function has the same properties as the l

.net平台下socket异步通讯

1,首先添加两个windows窗体项目,一个作为服务端server,一个作为客户端Client 2,然后添加服务端代码,添加命名空间,界面上添加TextBox控件 using System.Net; using System.Net.Sockets; 3,FormLoad方法添加代码 private void Form1_Load(object sender, EventArgs e) { IPEndPoint epServer = new IPEndPoint(IPAddress.Parse(

网络编程之Socket异步编程

看了上篇socket入门,相信你已经对socket有一定的了解了http://www.cnblogs.com/nsky/p/4501782.html 现在来回顾一下.上篇在循环接收客户端连接和循环接收消息的两个方法.都使用了多线程,因为Accept()方法是同步的,所有用了多线程异步执行. 1 static void Main(string[] args) 2 { 3 string ip = "192.168.1.2"; 4 int points = 8000; 5 6 Socket

socket 异步接收连接和接收数据

using System; using System.Collections.Generic; using System.ComponentModel; using System.Data; using System.Drawing; using System.Linq; using System.Text; using System.Threading.Tasks; using System.Windows.Forms; using System.Net.Sockets; using Syst

C# socket异步 服务端

转自:  http://blog.csdn.net/qq_20282263/article/details/54310737 1 private Dictionary<string, Session> SessionPool = new Dictionary<string, Session>(); 2 private Dictionary<string, string> MsgPool = new Dictionary<string, string>();

Socket层实现系列 — 信号驱动的异步等待

主要内容:Socket的异步通知机制. 内核版本:3.15.2 我的博客:http://blog.csdn.net/zhangskd 概述 socket上定义了几个IO事件:状态改变事件.有数据可读事件.有发送缓存可写事件.有IO错误事件. 对于这些事件,socket中分别定义了相应的事件处理函数,也称回调函数. Socket I/O事件的处理过程中,要使用到sock上的两个队列:等待队列和异步通知队列,这两个队列中 都保存着等待该Socket I/O事件的进程. Q:为什么要使用两个队列,等待