快捷搜索：  汽车  科技
当前位置： 爱玩科技 > 图文 > 正文

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）

逗爷 325

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）这其中主要包括4个基本组件：<center> （图2-1）事件驱动模型图 </center>从分类上，掌门在Socket长连接领域覆盖度很高，其中最核心场景为教学场景，单个会话信令最高可达80帧QPS，百兆峰值带宽下上课高峰时间对服务器冲击很大，高性能、可控制、可度量、可伸缩是生产系统服务提供的基础要求，如何设计一套高性能高可扩展性系统对团队提出了很大的考验，高性能事件驱动模型的探索正是方向之一。建议面向读者：对SocketIO或者对事件驱动设计有兴趣的开发人员。事件本是GUI领域最常用的概念，前端开发人员最常接触的一些GUI事件和事件模型框架比后端开发人员相对使用的更多，GUI中通常定义的一些事件，比如client、touch、doubleclick、multitouch、open、close等等都是对于GUI层面一些交互的抽象，这些具体的事件注册和响应也通常由G

背景

SocketIO原生基于NodeJS实现的Web长连接技术方案，H5原生场景下通常使用websocket作为基础协议进行网络通信（客户端支持多语言），SocketIO对于长连接场景下的业务形态进行了很多方面的抽象和实现，比如：命名空间、用户、房间等关系模型，技术形态下同样也进行了多方面的快速支持，比如ssl证书、websocket文本、二进制、双向Ack、心跳等API，作为一个Web长连接解决方案，SocketIO不失为一个很棒的基础协议支持框架，接入快速，模型简单，掌门在Socket通信接入侧选型上也选择了SocketIO协议。

如果是一个擅长java技术栈的后端来说，Netty-socketio（官方地址: https://github.com/mrniko/net... ）(4.4k star)的确是实现socketio服务的不二之选，这个项目由近几年比较火的redis官方推荐Java客户端连接工具redisson(11.6k star)作者(mrniko)于13年开发，已经有7年之久，已经处于事实上的停更状态，这给使用该项目作为web长连接后端框架来说，选择它通常并不是那么容易接受，而我们希望通过一系列的性能改造使地它能够更好地被我们所用。

业务背景

基于Socket的业务场景大致可以分为以下部分（其中加粗部分为掌门在Socket业务领域覆盖）：

  • 聊天场景： 即时通信
  • 直播场景：互动、弹幕
  • 智能家居IoT：监控、远程控制
  • 游戏场景：互动
  • 交通场景：位置共享
  • 教学场景：在线白板
  • 音视频：WebRTC信令协商
  • SLB长连接场景：网关

从分类上，掌门在Socket长连接领域覆盖度很高，其中最核心场景为教学场景，单个会话信令最高可达80帧QPS，百兆峰值带宽下上课高峰时间对服务器冲击很大，高性能、可控制、可度量、可伸缩是生产系统服务提供的基础要求，如何设计一套高性能高可扩展性系统对团队提出了很大的考验，高性能事件驱动模型的探索正是方向之一。

建议面向读者：对SocketIO或者对事件驱动设计有兴趣的开发人员。

概念

事件本是GUI领域最常用的概念，前端开发人员最常接触的一些GUI事件和事件模型框架比后端开发人员相对使用的更多，GUI中通常定义的一些事件，比如client、touch、doubleclick、multitouch、open、close等等都是对于GUI层面一些交互的抽象，这些具体的事件注册和响应也通常由GUI系统本身提供，而背后的实现逻辑，则无外乎下图所描述的事件模型实现。

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）(1)

<center> （图2-1）事件驱动模型图 </center>

这其中主要包括4个基本组件：

  • 事件队列（event queue）：接收事件的入口，存储待处理事件；
  • 分发器（event mediator）：将不同的事件分发到不同的业务逻辑单元；
  • 事件通道（event channel）：分发器与处理器之间的联系渠道；
  • 事件处理器（event processor）：实现业务逻辑，处理完成后会发出事件，触发下一步操作。

事件驱动模型的三要素：

  • 事件源：能够接收外部事件的源体；
  • 侦听器：能够接收事件源通知的对象；
  • 事件处理程序：用于处理事件的对象。

HTML中对于Body的标签预埋的事件，除了浏览器本身提供可声明行为监听，还可以针对框架本身进行扩展。

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）(2)

<center> （图2-2）github的body标签对应的事件 </center>

抛开GUI领域，事件驱动模型在其他领域发挥的作用也远比想象的要多，基于笔者的理解，大致能抽象一下如下的场景：

  • 可声明的行为（静态的）
  • 可状态化的行为抽象描述（动态的）

可状态化的行为，可以通过状态图来描述，而状态图通常是事件驱动模型设计中非常重要的建模模型。作为面向对象语言的Java开发人员，面向对象设计和状态图之间总会有一个成员变量表示这个对象的当前状态，而状态的变化（生命周期）用事件驱动模型设计思路不谋而合，比如下图展示的是课堂中的熔断逻辑状态图。

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）(3)

<center> （图2-3）单个会话生命周期信令熔断状态图 </center>

事件驱动在Java编程领域常用的基于guava的EventBus、RxJava的RxBus都是使用率较高的事件驱动框架，为了和SocketIO框架整合，本篇也重复造了一个轮子，功能更多、性能更强。

SocketIO事件驱动分析

SocketIO在事件驱动设计中大致可以划分为2个场景：

  • netty网络通信事件
  • 业务对象模型事件

SocketIO使用Netty作为底层的网络通信框架，而Netty使用NIO的API进行设计，Netty的事件驱动资料较为丰富，本篇不再重复。

业务对象模型

常用的连接事件

  • 'CONNECT'
  • 'DISCONNECT'
  • 'EVENT'
  • 'ACK'
  • 'ERROR'
  • 'BINARY_EVENT'
  • 'BINARY_ACK'

SocketIO服务端的事件API

基本上可以定义成单个会话的生命周期状态图。

  • onConnect
  • onDisconnect
  • onEvent
  • onPing

SocketIO事件扫描注册器

public interface AnnotationScanner { //扫描的注解类 Class<? extends Annotation> getScanAnnotation(); //给扫描的类注解方法添加Listener void addListener(Namespace namespace Object object Method method Annotation annotation); //验证扫描的类注解方法 void validate(Method method Class<?> clazz); }

注解扫描实现

  • OnConnectScanner
  • OnDisconnectScanner
  • OnEventScanner

优化与改造

SocketIO自身的事件模型框架给开发者提供了很大的扩展性，基于这些扩展性我们可以进行2个维度的优化改造，一方面针对SocketIO自身的业务事件进行二次优化和扩展，另一方面，基于生产的要求和功能的高度扩展性，迭代出可插拔的事件插件功能包。

基于上述两个维度，利用事件驱动模型可进行改造的方向主要围绕以下几个内容：

  • onData事件注册器改造
  • EventBus改造
  • 基于事件的原生事件API改造
  • 基于事件的Telemetry改造
  • 基于事件的Ops改造
  • NettyEpoll优化（补充）

论证

在进行改造之前，我们做一次常规的微基准测试，目的很简单，SocketIO提供的基于反射的事件驱动和NativeCall大致的性能对比到底如何。

测试点1：Event的反射调用基准测试

原理分析：反射需要执行一个相当昂贵的方法查找来获取描述特定方法的对象。同时，当一个方法被调用时，这要求 Java 虚拟机去运行本地代码，相比直接调用，这需要一个很长的运行时间。然而，现代 Java 虚拟机知道一个被称为“类型膨胀”的概念：基于 JNI 的方法调用会被动态生成的字节码给替换掉，而这些方法调用的字节码被注入到一个动态生成的类中。（即使 Java 虚拟机自身也使用代码生成！）毕竟，Java 的类型膨胀系统仍存在生成非常一般的代码的缺点，例如，仅能使用基本类型的装箱类型以至于性能缺陷不能完全解决。

跳过安全检查同样在反射中可以做到，便利的同时谁也不希望自己没有一点点防备。为此，特别写了一个JMH用于实验，作为此次改造依据。

微基准测试报告：

配置：19MacPro，i9-8C16G

反射实现：(1)jdk的methodInvoker实现，(2)cglib的fastInvoker实现，(3)编制成原生字节码NativeCall

结论：cglib(1倍速) < jdk(1倍速) << nativeCall(284倍速)

# JMH version: 1.22 # VM version: JDK 1.8.0_101 Java HotSpot(TM) 64-Bit Server VM 25.101-b13 # VM invoker: /Library/Java/JavaVirtualMachines/jdk1.8.0_101.jdk/Contents/Home/jre/bin/java # VM options: -Dfile.encoding=UTF-8 # Warmup: 10 iterations 10 s each # Measurement: 10 iterations 10 s each # Timeout: 10 min per iteration # Threads: 1 thread will synchronize iterations # Benchmark mode: Average time time/op JMH Report Benchmark Mode Cnt Score Error Units MethodInoveBenchmark.cgilib2 avgt 10 2.885 ± 0.094 ms/op MethodInoveBenchmark.jdk avgt 10 2.841 ± 0.050 ms/op MethodInoveBenchmark.nativeCall avgt 10 0.017 ± 0.001 ms/op

测试点2：Javassist静态编译、ByteBuddy动态编译

对接上面的性能压测篇，可以得到编织（字节码编译，完成字节码加载步骤后等价于硬编码）在某些固定调用场景下的作用随着量级的提升，整体的QPS差异是巨大的，这里主要围绕javasist（静态编译）和bytebuddy（动态 编译 （也称运行时编译））两种常用组件进行介绍如何对socketio的反射调用进行编译器的字节码编制，ByteBuddy底层ASM二次抽象框架，二者原理及优劣势在此处不做对比。

Javasist、ByteBuddy都支持字节码静态和动态编译

SocketIO对于信令时间的执行使用method.invoke()，对应的参数假设固定位2位，并且格式为。

public void invoke(SocketIOClient client AckRequest request)

Javassist伪代码：

<dependency> <groupId>org.javassist</groupId> <artifactId>javassist</artifactId> <version>3.26.0-GA</version> </dependency> private static final ClassPool classPool = ClassPool.getDefault(); static { try { classPool.insertClassPath(new ClassClassPath(AbstractMessageInvoker.class)); } catch (Exception e) { Log.error("AbstractMessageInvoker class is failed." e); } } //对需要扫描的注解方法进行字节码编译，修改并重新生成class文件，对事件加入前置过滤器和事件拦截器。 public static IMessageInvoker invokeEnhance(Object bean Method method String eventName CombineFilter filter CombineHandlerInterceptor combineHandlerInterceptor) throws Exception { String methodName = method.getName(); Class<?>[] parsClass = method.getParameterTypes(); CtClass result = classPool.makeClass(IMessageInvoker.class.getPackage().getName() "." bean.getClass().getSimpleName() methodName); result.setSuperclass(classPool.get(AbstractMessageInvoker.class.getCanonicalName())); //设置构造方法 setConstructor(result bean); //设置成员方法 setMethod(result method parsClass eventName); ... return (IMessageInvoker) result.toClass().getConstructor(bean.getClass() String.class CombineFilter.class CombineHandlerInterceptor.class).newInstance(bean eventName filter combineHandlerInterceptor); }

ByteBuddy伪代码：

<dependency> <groupId>net.bytebuddy</groupId> <artifactId>byte-buddy</artifactId> <version>1.10.5</version> </dependency> //对扫描的方法进行构造，并加入代理执行器 AbstractMessageInvoker invoker = (AbstractMessageInvoker) new ByteBuddy() .redefine(bean.getClass()) .method(named(method.getName()).and(takesArguments(method.getParameterTypes()))).intercept(MethodDelegation.to(MethodInterceptor.class)) .make() .load(MessageInvokeEnhance.class.getClassLoader() ClassLoadingStrategy.Default.WRAPPER) .getLoaded() .newInstance(); return invoker; } //在代理期中完成事件的前置过滤器和事件拦截器 @RuntimeType public void invoke(@Origin Method method @Argument(0)SocketIOClient socketIOClient @Argument(0)AckRequest ackRequest @Argument(0)String eventName @SuperCall Callable<?> callable) { try { boolean ret = filters.filter(socketIOClient ackRequest eventName); if (ret) { return; } ... callable.call(); } catch (Exception e) { log.error("MethodInterceptor error" e); } finally { } }

改造点

DataListener.onData事件注册器改造

SocketIO源码：原生通过反射Method.invoke进行调用。

namespace.addEventListener(annotation.value() objectType new DataListener<Object>() { @Override public void onData(SocketIOClient client Object data AckRequest ackSender) { try { Object[] args = new Object[method.getParameterTypes().length]; if (socketIOClientIndex != -1) { args[socketIOClientIndex] = client; } if (ackRequestIndex != -1) { args[ackRequestIndex] = ackSender; } if (!dataIndexes.isEmpty()) { int dataIndex = dataIndexes.iterator().next(); args[dataIndex] = data; } //基于反射的调用 method.invoke(object args); } catch (InvocationTargetException e) { throw new SocketIOException(e.getCause()); } catch (Exception e) { throw new SocketIOException(e); } } });

改造后源码：现改造成编制调用，并加入到事件监听器。

socketIOServer.addEventListener(eventName objectType new DataListener<Object>() { @Override public void onData(SocketIOClient client Object data AckRequest ackSender) { try { //扫描并添加编译后的基于字节码执行逻辑 final IMessageInvoker thisInvoker = messageInvokeMap.get(eventName); if (Objects.isNull(thisInvoker)) { return;} if (parsClass.length == 2) { thisInvoker.invoke(client ackSender); } else if (parsClass.length == 3) { thisInvoker.invoke(client ackSender data); } else if (parsClass.length == 4) { thisInvoker.invoke(client ackSender data); } } catch (Exception e) { throw new SocketIOException(e); } } });

高性能EventBus

阅读guava的EventBus源码设计思路为定制EventBus提供很好的API模型参考。

SocketIO场景中，信令事件的下发通常使用Netty原生的workerGroup线程池的NioEventLoop Thread进行，Eventbus在此完成事件模型从通信事件到业务事件的解耦， 通常同步调用支持会是基本的诉求，其次，为了保证通信线程的高效，耗时的业务事件需要EventBus以异步的形式去执行。

EventBus通常包含以下几个功能API:

  • 事件扫描注册；
  • 事件的post接口（同步、异步）；
  • 事件声明式注解。

异步和同步在基本情况下，原则上仅仅是多了一层任务分派执行者从本线程变成了其他线程。高性能的EventBus设计，我们依旧把精力放在事件的调用上，同时针对各种业务形态，定制高效的异步Actor线程池， 消除锁竞争 是提升线程池效率的利器，实现IO和业务解耦，用一些讨巧的命名获取可以更好理解这些模型，大致可理解为：

  • 1对1模型：一个线程分配一个独立队列，互相不干扰；
  • 大班课模型：按照业务划分，使用多个队列，具备同一特征的任务分配到同一个队列，这些队列的数量可进行扩展，参考Kafka Partitation队列；
  • 不可描述模型：常规线程池模型，使用一个任务队列，在队列的选型上可以参考LMAX Disruptor①，Tomcat8.5 提供QueuableCachedThreadPool。

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）(4)

<center> （图4-1）某种优化过的1对1线程池模型图 </center>

扩展性：

EventBus的扩展性同SocketIO事件一样使用可插播的过滤器和拦截器来完成，预制事前Filter，事中Interceptor完成事件的过程管控。

优雅关闭：

EventBus在提供异步功能的前提下，做到任务的可追踪可度量、消息不丢失，做好优雅关闭是另一件重要事情，对于定制的线程池，需要根据队列和正在运行的任务消费情况进行生命周期的关闭。

跨线程：

异步带来的另外一个问题跨线程问题，在监控和流控场景造成了很大的适配性难题，单一局部场景下，InheritThreadLocal能解燃眉之急，侵入的编程传递也能完成功能，如何做到低侵入和高性能的适配是EventBus考虑的另一个难题，好在EventBus场景比较受控，贯彻约定大于配置思路，按需调用定制型的submitAPI。

常用的低侵入跨线程同步方案：②

  • java agent
  • transmittable-thread-local

度量：

EventBus度量常用的指标维度：

  • QPS：1m、5m
  • ThreadPool：active、queue、max
  • processTime：min、max、avg

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）(5)

<center>（图4-2）Actor线程池监控</center>

②Skywalking使用Javaagent扫描TraceCrossThread注解进行跨线程同步

基于事件的Telemetry改造

Telemetry通常由Log、Metric、Trace③构成，这里简单介绍如何通过事件驱动完成三者的Collector模型。

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）(6)

<center> （图4-3）监控作用域维恩图 </center>

产品上线，监控先行

SocketIO原生接口缺失：相对于参考作者的(redisson Pro)版本使用dropwizard metrics进行扩展；

SocketIO需要对监控的部分代码已经改造，以支持高效率的监控Guage信息和可扩展的拦截的其他埋点信息。

Metrics可以使用opentelemetry④/dropwizardmetrics/prometheus-client。

SocketIO可根据基础事件埋点的Metrics案例：

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）(7)

<center> （图4-4）SocketIO常规事件监控 </center>

③对于Log、Metric、Trace三者之间的关系可参考sw作者吴晟的译文： https://wu-sheng.github.io/me...

④基本上opentelemetry想要实现的Log Metrics Trace未来目标都是我们想要的，但是opentelemetry更新的进度实在难以接受，所以使用成熟的第三方会更加容易生产上线。opentelemetry对于Log的支持目前还未开始。

监控埋点和收集

本章节简单介绍如何使用SocketIO事件模型框架通过事件和扩展点低侵入式完成各种监控埋点和收集。

Log:

画像埋点就是对非结构化数据的特征抽象，当需要将服务端的用户日志UserLog纳入到画像分析时，用户特征状态和事件之间可以抽象出一种对照表。

LogStateRegister.registe(ConnectEvent) LogStateRegister.registe(DisconnectEvent)

增加一个监控监听完成对特征事件的监听，并根据特征进行日志的输出，完成部分用画画像日志的低侵入埋点。

void onEvent(ConnectEvent event) { LogStateRegister.log(event) }

增加异常Log收集和反馈：类似于SpringMVC的ExceptionHandler的基于事件的统一的异常反馈机制。

public class SocketIOServerExceptionFacade { @Autowired private SocketIOServer socketIOServer; @EventHandler public void onSessionInvokerExceptionEvent(SessionInvokerExceptionEvent ex) { log.error(ex.getMessage ex.getException()); ex.getRequest().sendAckData(PacketUtil.socketError(ex.getAppid() ex.getGroupId() ex.getErrorCode() ex.getErrorMsg())); } }

Metric:

同样，针对特征指标的抽取埋点，同样可以低侵入完成，大致过程如下：

1）对发送包耗时和QPS进行监控；

2）定义Meter：SocketMessageMeter.java；

3）将Meter注册到MetricsRegister；

4）新增一个事件监听完成MeterAPI的调用；

5）完成收集并输出。

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）(8)

<center> （图4-5）SocketIO信令事件监控 感谢@郭浩大佬供图 </center>

定义Histograms(Histograms在高QPS情况下性能有比较大的影响，所以通常建议自己实现timerange结果再输出到Meter)

SocketMessageHistograms.java

Tracer：

为了达到尽可能少的侵入代码，通常集成skywalking plugins 6.5.0 （netty-socketio.plugin）进行链路Trace收集。

Trace链路常规需求：

  • 提供用户接入连接现场链路；
  • 提供用户断开连接的现场链路；
  • 提供用户关键信令的链路；

Trace更多定制项：（部分功能需要定制代码）针对超高QPS信令来说，实际情况需要提供更多可扩展性：

  • 采样拦截；
  • 特征值(染色)拦截；
  • 固定event拦截；
  • 魔数拦截（二进制协议）；

针对netty-socketio高性能的定位，对sw的期望更加体现在关键节点的完整性扩展性，而非给所有信令铺上一张大网，大鱼小鱼一把抓，所以需要进一步定制socketio-plugin按需所取，针对定制后的socketio的入口，出口，链路进行拦截，原生的事件可以提供较为入门的功能使用，插件开发基础难度不高，具体开发内容篇幅内容不展开。

这里以官方插件为例：

Skywalking-agent打包时使用share将代码全部到打到该jar，避免冲突问题，需要手动打socketio-plugins包。

备注6.5.0新增了SOCKET_IO，6.5.0以下版本使用该包降级会出现异常，降级使用需要定制该域。

插件定制打包过程

下载打包，因为会有checkstyle强制检查，这里暂时跳过。

git checkout v6.5.0 mvn package -Dcheckstyle.skip=true

或者

<plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-checkstyle-plugin</artifactId> <executions> <execution> <id>checkstyle-validation</id> <phase>none</phase> </execution> </executions> </plugin>

定制内容

1). 替换所有字段：

NettySocketIOConnectionInterceptor.java

public static final OfficialComponent SOCKET_IO = new OfficialComponent(76 "SocketIO"); span.setComponent(SOCKET_IO);

否则在低版本中会有Field异常

ERROR 2019-12-26 14:43:43:933 epollEventLoopGroup-9-3 InstMethodsInter : class[class com.corundumstudio.socketio.namespace.Namespace] before method[onDisconnect] intercept failure java.lang.NoSuchFieldError: SOCKET_IO at org.apache.skywalking.apm.plugin.netty.socketio.NettySocketIOConnectionInterceptor.beforeMethod(NettySocketIOConnectionInterceptor.java:44) at org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstMethodsInter.intercept(InstMethodsInter.java:82) at com.corundumstudio.socketio.namespace.Namespace.onDisconnect(Namespace.java) at com.corundumstudio.socketio.transport.NamespaceClient.onDisconnect(NamespaceClient.java:115) at com.corundumstudio.socketio.handler.ClientHead.onChannelDisconnect(ClientHead.java:183) at com.corundumstudio.socketio.transport.WebSocketTransport.channelInactive(WebSocketTransport.java:148)

2). 重命名6.x.0，加入到plugins中，查看日志：

INFO 2019-12-26 15:37:56:434 localhost-startStop-1 AgentClassLoader : /usr/skywalking-apm/agent/plugins/apm-netty-socketio-plugin-6.3.0.jar loaded.

3). 源码分析：

netty-socketio-plugin能提供的埋点源码可以看到trace的出入口拦截方式。

a）拦截SocketIOClient编译代理对象：NamespaceClient：

protected ClassMatch enhanceClass() { return byName("com.corundumstudio.socketio.transport.NamespaceClient"); }

b）拦截加入、离开房间事件：

public ElementMatcher<MethodDescription> getMethodsMatcher() { //对加入、退出房间行为进行拦截 return named("joinRoom").or(named("leaveRoom")); }

c）拦截基础连接和信令事件：

new InstanceMethodsInterceptPoint() { @Override public ElementMatcher<MethodDescription> getMethodsMatcher() { //加入基础信令事件拦截 return named("onEvent"); } @Override public String getMethodsInterceptor() { return "org.apache.skywalking.apm.plugin.netty.socketio.NettySocketIOOnEventInterceptor"; } @Override public boolean isOverrideArgs() { return false; } } new InstanceMethodsInterceptPoint() { @Override public ElementMatcher<MethodDescription> getMethodsMatcher() { //对连接事件进行拦截 return named("onConnect"); } @Override public String getMethodsInterceptor() { return "org.apache.skywalking.apm.plugin.netty.socketio.NettySocketIOConnectionInterceptor"; } @Override public boolean isOverrideArgs() { return false; } } new InstanceMethodsInterceptPoint() { @Override public ElementMatcher<MethodDescription> getMethodsMatcher() { //对断开连接事件进行来接 return named("onDisconnect"); } @Override public String getMethodsInterceptor() { return "org.apache.skywalking.apm.plugin.netty.socketio.NettySocketIOConnectionInterceptor"; } @Override public boolean isOverrideArgs() { return false; } }

4). 在SW-UI中结果展示：

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）(9)

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）(10)

<center> （图4-6）SocketIO接入流程Trace链路 </center>

更多功能参考：

  1. Log集成Skywalking toolkit-TID的集成
  2. 获取TraceContext中的内容
  3. 常用的异步事件驱动中跨线程处理

基于事件的Ops改造

redis和mysql client提供了很多强有力的客户端工具，管理人员能通过简单的命令设置或者拿到需要的第一手信息，同样，在线上运行的socket机器是个典型的有状态服务，同样需要一系列的管理手段，Ops工具的改造同样值得投入，特别在近期的压测过程中能够更快速做到信息响应和控制，堪称好用。

定义并注册以下事件的实现：

SocketIO-Cli：telnet hacker $->auth password #验证客户端 $->socketio info #显示socketioinfo $->socketio show rooms #显示房间 $->socketio show namespaces #显示命名空间 $->socketio kill sessionId #强制踢出会话 $->socketio performance 100 #显示某个Top100性能指标

socket类编程的三个步骤（SocketIO高性能事件驱动设计探索）(11)

<center> （图4-7）SocketIO-Ops工具客户端 </center>

优化

Netty Epoll改造

Epoll事件模型动态支持

socketio只能根据配置开关：useLinuxNativeEpoll=true，调整成优先尝试策略

public static boolean useNettyEpoll() { if (useEpoll) { try { Class.forName("io.netty.channel.epoll.Native"); return true; } catch (Throwable error) { LOGGER.warn("can not load netty epoll switch nio model."); } } return false; }

epoll没有描述符限制，用户态拷贝到内核态只需要一次使用事件通知，通过epoll_ctl注册fd 一旦该fd 就绪，内核就采用callback机制激活对应的fd

优点：没有fd限制，所支持的 FD 上限是操作系统的最大文件句柄数（65535），1G 内存大概支持 10W 句柄，支持百万连接的话，16G 内存就可以搞定

效率高，使用回调通知而不是轮询方式，不会随着 FD 数目增加效率下降，通过 callback 机制通知，内核和用户空间 mmap 同一块内存实现

/** * 在Netty 4中实现了一个新的ByteBuf内存池，它是一个纯Java版本的 jemalloc （Facebook也在用）。 * 现在，Netty不会再因为用零填充缓冲区而浪费内存带宽了。不过，由于它不依赖于GC，开发人员需要小心内存泄漏。 * 如果忘记在处理程序中释放缓冲区，那么内存使用率会无限地增长。 * Netty默认不使用内存池，需要在创建客户端或者服务端的时候进行指定 */ b.option(ChannelOption.ALLOCATOR PooledByteBufAllocator.DEFAULT); b.childOption(ChannelOption.ALLOCATOR PooledByteBufAllocator.DEFAULT);

结语

SocketIO在更多地承担着网络通信和对象模型封装工作，而往往在整个产品中，整个功能的复杂度并不是1 1=2的叠加关系，而是各种交叉和耦合关系，事件驱动开发模型，在后期迭代的长尾优势效应会越发明显。优势好坏之边界，从来就是架构开发致力于划清的对象，是非之间，一如哈姆雷特之多，二如薛定谔的猫之困惑，笔者认为对于改造好坏一说并无统一定论，重构也亦无银弹一说，仅限于对能抓到老鼠的猫就是好猫的认可，适合即妥当，关于SocketIO更多的底层改造内容未来会在在线课堂和交互中台迭代以适配更加丰富的交互场景。

新一代面向应用的通信协议框架RSocket诞生至今，一直不温不火，但貌似并不妨碍成为未来的主流服务间的标准通信方案，未来是否可以跨越多语言无缝对接，站在巨人的肩膀上快速完成业务对于通信的高定制要求，不失为一个未来方向。

作者介绍：徐敏，开源爱好者，10年 互联网开发经验，现任掌门课堂云架构师，主要负责课堂云实时交互中台和4层网关架构设计开发工作。

网站首页

返回栏目

猜您喜欢：

相关文章