AI 助理
备案控制台
开发者社区 云计算 文章 正文

Cowboy 开源 WebSocket 网络库

2017-12-20 2532
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介：

Cowboy.WebSockets 是一个托管在 GitHub 上的基于 .NET/C# 实现的开源 WebSocket 网络库，其完整的实现了 RFC 6455 (The WebSocket Protocol) 协议标准，并部分实现了 RFC 7692 (Compression Extensions for WebSocket) 协议标准。

WebSocket 可理解为建立在 TCP 连接通道上的更进一步的握手，并确定了消息封装格式。

通过定义控制帧 (Control Frame) 和数据帧 (Data Frame) 来控制通道内的通信和数据传输，下图用使用 ABNF 格式描述了帧头部的格式。

Cowboy.WebSockets 中对于 WebSocket 的 Client/Server 分别做了实现，分别对应代码中的：

Cowboy.WebSockets 的内部实现是基于 Cowboy.Sockets 中的 TAP 模式的 AsyncTcpSocketServer 和 AsyncTcpSocketClient 。关于 Cowboy.Sockets 可以参考文章《C#高性能TCP服务的多种实现方式》。

可通过 NuGet 查找 Cowboy 来获取 nuget 包。

WebSocket 服务端应用

实现 AsyncWebSocketServerModule 抽象类，其中 ModulePath 对应着 "ws://host:port/path" 中的 path 部分。可以实现多个 Module，将多个 Module 注入到 AsyncWebSocketServerModuleCatalog 中，或者采用反射机制等自动发现 Module。

复制代码 
  public class TestWebSocketModule : AsyncWebSocketServerModule
  {
      public TestWebSocketModule()
          : base(@"/test")
      {
      }
  
      public override async Task OnSessionStarted(AsyncWebSocketSession session)
      {
          Console.WriteLine(string.Format("WebSocket session [{0}] has connected.", session.RemoteEndPoint));
          await Task.CompletedTask;
      }
  
      public override async Task OnSessionTextReceived(AsyncWebSocketSession session, string text)
      {
          Console.Write(string.Format("WebSocket session [{0}] received Text --> ", session.RemoteEndPoint));
          Console.WriteLine(string.Format("{0}", text));
  
          await session.SendTextAsync(text);
      }
  
      public override async Task OnSessionBinaryReceived(AsyncWebSocketSession session, byte[] data, int offset, int count)
      {
          var text = Encoding.UTF8.GetString(data, offset, count);
          Console.Write(string.Format("WebSocket session [{0}] received Binary --> ", session.RemoteEndPoint));
          Console.WriteLine(string.Format("{0}", text));
  
          await session.SendBinaryAsync(Encoding.UTF8.GetBytes(text));
      }
  
      public override async Task OnSessionClosed(AsyncWebSocketSession session)
      {
          Console.WriteLine(string.Format("WebSocket session [{0}] has disconnected.", session.RemoteEndPoint));
          await Task.CompletedTask;
      }
  }
复制代码

实例化 AsyncWebSocketServer，并将 AsyncWebSocketServerModuleCatalog 实例注入，即可启动 WebSocket 的服务端监听。

复制代码 
  class Program
  {
      static AsyncWebSocketServer _server;
  
      static void Main(string[] args)
      {
          NLogLogger.Use();
  
          try
          {
              var catalog = new AsyncWebSocketServerModuleCatalog();
              catalog.RegisterModule(new TestWebSocketModule());
  
              var config = new AsyncWebSocketServerConfiguration();
              //config.SslEnabled = true;
              //config.SslServerCertificate = new System.Security.Cryptography.X509Certificates.X509Certificate2(@"D:\\Cowboy.pfx", "Cowboy");
              //config.SslPolicyErrorsBypassed = true;
  
              _server = new AsyncWebSocketServer(22222, catalog, config);
              _server.Listen();
  
              Console.WriteLine("WebSocket server has been started on [{0}].", _server.ListenedEndPoint);
              Console.WriteLine("Type something to send to clients...");
              while (true)
              {
                  try
                  {
                      string text = Console.ReadLine();
                      if (text == "quit")
                          break;
                      Task.Run(async () =>
                      {
                          //await _server.BroadcastText(text);
                          //Console.WriteLine("WebSocket server [{0}] broadcasts text -> [{1}].", _server.ListenedEndPoint, text);
                          await _server.BroadcastBinaryAsync(Encoding.UTF8.GetBytes(text));
                          Console.WriteLine("WebSocket server [{0}] broadcasts binary -> [{1}].", _server.ListenedEndPoint, text);
                      });
                  }
                  catch (Exception ex)
                  {
                      Console.WriteLine(ex.Message);
                  }
              }
  
              _server.Shutdown();
              Console.WriteLine("WebSocket server has been stopped on [{0}].", _server.ListenedEndPoint);
          }
          catch (Exception ex)
          {
              Logger.Get<Program>().Error(ex.Message, ex);
          }
  
          Console.ReadKey();
      }
  }
复制代码

WebSocket 客户端应用

客户端侧在实例化 AsyncWebSocketClient 时有两种方式：

  1. 实现 IAsyncWebSocketClientMessageDispatcher 接口；
  2. 直接构造函数注入接受各种事件的 Func<> 实现；
复制代码 
  public interface IAsyncWebSocketClientMessageDispatcher
  {
      Task OnServerConnected(AsyncWebSocketClient client);
      Task OnServerTextReceived(AsyncWebSocketClient client, string text);
      Task OnServerBinaryReceived(AsyncWebSocketClient client, byte[] data, int offset, int count);
      Task OnServerDisconnected(AsyncWebSocketClient client);
  
      Task OnServerFragmentationStreamOpened(AsyncWebSocketClient client, byte[] data, int offset, int count);
      Task OnServerFragmentationStreamContinued(AsyncWebSocketClient client, byte[] data, int offset, int count);
      Task OnServerFragmentationStreamClosed(AsyncWebSocketClient client, byte[] data, int offset, int count);
  }
复制代码

下面的 DEMO 采用了方式二。

复制代码 
  class Program
  {
      static AsyncWebSocketClient _client;
  
      static void Main(string[] args)
      {
          NLogLogger.Use();
  
          Task.Run(async () =>
          {
              try
              {
                  var config = new AsyncWebSocketClientConfiguration();
                  //config.SslTargetHost = "Cowboy";
                  //config.SslClientCertificates.Add(new System.Security.Cryptography.X509Certificates.X509Certificate2(@"D:\\Cowboy.cer"));
                  //config.SslPolicyErrorsBypassed = true;
  
                  //var uri = new Uri("ws://echo.websocket.org/");
                  //var uri = new Uri("wss://127.0.0.1:22222/test");
                  var uri = new Uri("ws://127.0.0.1:22222/test");
                  _client = new AsyncWebSocketClient(uri,
                      OnServerTextReceived,
                      OnServerBinaryReceived,
                      OnServerConnected,
                      OnServerDisconnected,
                      config);
                  await _client.Connect();
  
                  Console.WriteLine("WebSocket client has connected to server [{0}].", uri);
                  Console.WriteLine("Type something to send to server...");
                  while (_client.State == WebSocketState.Open)
                  {
                      try
                      {
                          string text = Console.ReadLine();
                          if (text == "quit")
                              break;
                          Task.Run(async () =>
                          {
                              //await _client.SendText(text);
                              //Console.WriteLine("Client [{0}] send text -> [{1}].", _client.LocalEndPoint, text);
                              await _client.SendBinaryAsync(Encoding.UTF8.GetBytes(text));
                              Console.WriteLine("Client [{0}] send binary -> [{1}].", _client.LocalEndPoint, text);
                          }).Forget();
                      }
                      catch (Exception ex)
                      {
                          Console.WriteLine(ex.Message);
                      }
                  }
  
                  await _client.Close(WebSocketCloseCode.NormalClosure);
                  Console.WriteLine("WebSocket client has disconnected from server [{0}].", uri);
              }
              catch (Exception ex)
              {
                  Logger.Get<Program>().Error(ex.Message, ex);
              }
          }).Wait();
  
          Console.ReadKey();
      }
  
      private static async Task OnServerConnected(AsyncWebSocketClient client)
      {
          Console.WriteLine(string.Format("WebSocket server [{0}] has connected.", client.RemoteEndPoint));
          await Task.CompletedTask;
      }
  
      private static async Task OnServerTextReceived(AsyncWebSocketClient client, string text)
      {
          Console.Write(string.Format("WebSocket server [{0}] received Text --> ", client.RemoteEndPoint));
          Console.WriteLine(string.Format("{0}", text));
  
          await Task.CompletedTask;
      }
  
      private static async Task OnServerBinaryReceived(AsyncWebSocketClient client, byte[] data, int offset, int count)
      {
          var text = Encoding.UTF8.GetString(data, offset, count);
          Console.Write(string.Format("WebSocket server [{0}] received Binary --> ", client.RemoteEndPoint));
          Console.WriteLine(string.Format("{0}", text));
  
          await Task.CompletedTask;
      }
  
      private static async Task OnServerDisconnected(AsyncWebSocketClient client)
      {
          Console.WriteLine(string.Format("WebSocket server [{0}] has disconnected.", client.RemoteEndPoint));
          await Task.CompletedTask;
      }
  }
复制代码

相关资料



本文转自匠心十年博客园博客，原文链接：http://www.cnblogs.com/gaochundong/p/cowboy_websockets.html，如需转载请自行联系原作者

关键词：
网络库
开源WebSocket
网络开源
WebSocket网络
网络websocket
嗯哼9925
目录
相关文章
杨枝甘露z
|
2月前
|
C++
基于Reactor模型的高性能网络库之地址篇
这段代码定义了一个 InetAddress 类，是 C++ 网络编程中用于封装 IPv4 地址和端口的常见做法。该类的主要作用是方便地表示和操作一个网络地址（IP + 端口）
杨枝甘露z
160 58
杨枝甘露z
|
2月前
|
网络协议 算法 Java
基于Reactor模型的高性能网络库之Tcpserver组件-上层调度器
TcpServer 是一个用于管理 TCP 连接的类，包含成员变量如事件循环（EventLoop）、连接池（ConnectionMap）和回调函数等。其主要功能包括监听新连接、设置线程池、启动服务器及处理连接事件。通过 Acceptor 接收新连接，并使用轮询算法将连接分配给子事件循环（subloop）进行读写操作。调用链从 start() 开始，经由线程池启动和 Acceptor 监听，最终由 TcpConnection 管理具体连接的事件处理。
杨枝甘露z
57 2
杨枝甘露z
|
2月前
基于Reactor模型的高性能网络库之Tcpconnection组件
TcpConnection 由 subLoop 管理 connfd，负责处理具体连接。它封装了连接套接字，通过 Channel 监听可读、可写、关闭、错误等
杨枝甘露z
75 1
杨枝甘露z
|
2月前
|
负载均衡 算法 安全
基于Reactor模式的高性能网络库之线程池组件设计篇
EventLoopThreadPool 是 Reactor 模式中实现“一个主线程 + 多个工作线程”的关键组件，用于高效管理多个 EventLoop 并在多核 CPU 上分担高并发 I/O 压力。通过封装 Thread 类和 EventLoopThread，实现线程创建、管理和事件循环的调度，形成线程池结构。每个 EventLoopThread 管理一个子线程与对应的 EventLoop（subloop），主线程（base loop）通过负载均衡算法将任务派发至各 subloop，从而提升系统性能与并发处理能力。
杨枝甘露z
107 3
杨枝甘露z
|
2月前
基于Reactor模式的高性能网络库github地址
https://github.com/zyi30/reactor-net.git
杨枝甘露z
45 0
杨枝甘露z
|
2月前
基于Reactor模型的高性能网络库之Poller(EpollPoller)组件
封装底层 I/O 多路复用机制（如 epoll）的抽象类 Poller，提供统一接口支持多种实现。Poller 是一个抽象基类，定义了 Channel 管理、事件收集等核心功能，并与 EventLoop 绑定。其子类 EPollPoller 实现了基于 epoll 的具体操作，包括事件等待、Channel 更新和删除等。通过工厂方法可创建默认的 Poller 实例，实现多态调用。
杨枝甘露z
194 60
杨枝甘露z
|
2月前
基于Reactor模型的高性能网络库之Channel组件篇
Channel 是事件通道，它绑定某个文件描述符 fd，注册感兴趣的事件（如读/写），并在事件发生时分发给对应的回调函数。
杨枝甘露z
155 60
杨枝甘露z
|
2月前
|
安全 调度
基于Reactor模型的高性能网络库之核心调度器：EventLoop组件
它负责：监听事件（如 I/O 可读写、定时器）、分发事件、执行回调、管理事件源 Channel 等。
杨枝甘露z
168 57
杨枝甘露z
|
2月前
基于Reactor模型的高性能网络库之时间篇
是一个用于表示时间戳（精确到微秒）**的简单封装类
杨枝甘露z
125 57
蓝易云
|
2月前
|
JSON 网络安全 数据格式
Python网络请求库requests使用详述
总结来说，`requests`库非常适用于需要快速、简易、可靠进行HTTP请求的应用场景，它的简洁性让开发者避免繁琐的网络代码而专注于交互逻辑本身。通过上述方式，你可以利用 `requests`处理大部分常见的HTTP请求需求。
蓝易云
267 51

热门文章

最新文章

  • 1
    AI 基础知识从 0.6 到 0.7—— 彻底拆解深度神经网络训练的五大核心步骤
  • 2
    【PID】基于人工神经网络的PID控制器，用于更好的系统响应研究（Matlab&Simulink代码实现）
  • 3
    【FFNN负荷预测】基于人工神经网络的空压机负荷预测（Matlab代码实现）
  • 4
    阿里云服务器ECS实例规格族是什么？不同规格CPU型号、处理器主频及网络性能参数均不同
  • 5
    手表只会报步数？别闹了，用神经网络优化健康监测才是王道
  • 6
    【多智能体系统】遭受DoS攻击的网络物理多智能体系统的弹性模型预测控制MPC研究（Simulink仿真实现）
  • 7
    具有多种最大功率点跟踪（MPPT）方法的光伏发电系统（P&O-增量法-人工神经网络-模糊逻辑控制-粒子群优化）之使用粒子群算法的最大功率点追踪（MPPT）（Simulink仿真实现）
  • 8
    水声网络(UAN)仿真的信道建模（Matlab代码实现）
  • 9
    【PSO-LSTM】基于PSO优化LSTM网络的电力负荷预测（Python代码实现）
  • 10
    【水质预测】基于模糊神经网络FNN水质预测研究（Matlab代码实现）
  • 1
    【电机轴承监测】基于matlab声神经网络电机轴承监测研究（Matlab代码实现）
    46
  • 2
    【创新无忧】基于白鲨算法WSO优化广义神经网络GRNN电机故障诊断（Matlab代码实现）
    54
  • 3
    【图像分割】基于神经气体网络的图像分割与量化（Matlab代码实现）
    34
  • 4
    【多智能体系统】遭受DoS攻击的网络物理多智能体系统的弹性模型预测控制MPC研究（Simulink仿真实现）
    78
  • 5
    阿里云服务器ECS实例规格族是什么？不同规格CPU型号、处理器主频及网络性能参数均不同
    85
  • 6
    【PID】基于人工神经网络的PID控制器，用于更好的系统响应研究（Matlab&Simulink代码实现）
    130
  • 7
    【PSO-LSTM】基于PSO优化LSTM网络的电力负荷预测（Python代码实现）
    64
  • 8
    AI 基础知识从 0.6 到 0.7—— 彻底拆解深度神经网络训练的五大核心步骤
    383
  • 9
    【FFNN负荷预测】基于人工神经网络的空压机负荷预测（Matlab代码实现）
    92
  • 10
    【水质预测】基于模糊神经网络FNN水质预测研究（Matlab代码实现）
    67

    • 相关课程

    更多
  • TCP/IP 网络基础
  • 云计算工程师解析与实战-网络专家篇（体验版）
  • 云网络白皮书-阿里云网络系列课
  • 企业上云攻略-阿里云网络产品应用系列教程
  • Linux网络进阶 - TCP/IP协议及OSI七层模型
  • 网络管理者必知-2分钟了解新出台的《网络安全法》

    • 相关电子书

    更多
  • Session：更加安全、可靠的数据中心网络产品更新
  • Session：极简易用的全球化网络产品更新
  • Session：弹性、高可用、可观测的应用交付网络产品更新
    • 下一篇
    阿里云 EMR Serverless Spark 版开启免费公测