C/C++教程
websocket-cometd源码阅读-初始化(二)
本文主要是介绍websocket-cometd源码阅读-初始化(二)，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
常规配置
<!-- CometD Servlet -->
<servlet>
<servlet-name>cometd</servlet-name>
<!--<1>-->
<servlet-class>org.cometd.annotation.server.CometDServlet</servlet-class>
<!--liqiang todo 600000-->
<init-param>
<param-name>maxProcessing</param-name>
<param-value>600000</param-value>
</init-param>
<init-param>
<param-name>timeout</param-name>
<param-value>20000</param-value>
</init-param>
<init-param>
<param-name>interval</param-name>
<param-value>0</param-value>
</init-param>
<init-param>
<param-name>maxInterval</param-name>
<param-value>10000</param-value>
</init-param>
<init-param>
<param-name>handshakeReconnect</param-name>
<param-value>true</param-value>
</init-param>
<init-param>
<param-name>maxLazyTimeout</param-name>
<param-value>5000</param-value>
</init-param>
<init-param>
<param-name>long-polling.multiSessionInterval</param-name>
<param-value>2000</param-value>
</init-param>
<init-param>
<param-name>services</param-name>
<param-value>org.cometd.examples.ChatService</param-value>
</init-param>
<init-param>
<param-name>ws.cometdURLMapping</param-name>
<param-value>/cometd/*</param-value>
</init-param>
<!--容器启动时调用init方法初始化 而不是第一次调用时-->
<load-on-startup>1</load-on-startup>
<async-supported>true</async-supported>
</servlet>
<1>
通过ServletInit为切入点进行初始化
org.cometd.server.CometDServlet#init
@Override
public void init() throws ServletException {
try {
boolean export = false;
//可以通过ServletContext如果在别的地方自定义初始化 需要指定此值 比如自己根据spring容器初始化了
_bayeux = (BayeuxServerImpl)getServletContext().getAttribute(BayeuxServer.ATTRIBUTE);
if (_bayeux == null) {
export = true;
//初始化
_bayeux = newBayeuxServer();
//这里主要是设置Servlet参数到_bayeux 供后续初始化使用
// Transfer all servlet init parameters to the BayeuxServer implementation
for (String initParamName : Collections.list(getInitParameterNames())) {
_bayeux.setOption(initParamName, getInitParameter(initParamName));
}
//添加ServletContext到_bayeux
// Add the ServletContext to the options
_bayeux.setOption(ServletContext.class.getName(), getServletContext());
}
//调用start进行组件初始化<2>
_bayeux.start();
if (export) {
//设置到Attribute
getServletContext().setAttribute(BayeuxServer.ATTRIBUTE, _bayeux);
}
} catch (Exception x) {
throw new ServletException(x);
}
}
<2>
org.cometd.server.BayeuxServerImpl#doStart
@Override
protected void doStart() throws Exception {
//<3>初始化内置的渠道
initializeMetaChannels();
//<4>初始化消息序列化转换器
initializeJSONContext();
//<5>初始化Transport
initializeServerTransports();
//如果线程池为空 创建线程池
if (_executor == null) {
//<7>
_executor = new MarkedReference<>(newExecutor(), true);
}
addBean(_executor.getReference());
//如果Schedule为空 创建Scheduler
if (_scheduler == null) {
_scheduler = new MarkedReference<>(newScheduler(), true);
}
addBean(_scheduler.getReference());
//配置获得validateMessageFields 默认为true 是否进行消息格式校验
_validation = getOption(VALIDATE_MESSAGE_FIELDS_OPTION, true);
//配置获得broadcastToPublisher 默认为true 此消息是否广播到发布者,比如都订阅了同一个渠道,是否广播给自己
_broadcastToPublisher = getOption(BROADCAST_TO_PUBLISHER_OPTION, true);
super.doStart();
long defaultSweepPeriod = 997;
//获得配置的sweepPeriodOption 会话扫描周期检查会话是否需要移除 默认997 毫秒单位
long sweepPeriodOption = getOption(SWEEP_PERIOD_OPTION, defaultSweepPeriod);
if (sweepPeriodOption < 0) {
sweepPeriodOption = defaultSweepPeriod;
}
long sweepPeriod = sweepPeriodOption;
//开启session检查,检查是否需要剔除
schedule(new Runnable() {
@Override
public void run() {
//并行执行asyncSweep的四个任务 执行完后指定周期后 开启下一轮 实现了定时任务效果
asyncSweep().whenComplete((r, x) -> schedule(this, sweepPeriod));
}
}, sweepPeriod);
}
<3>
org.cometd.server.BayeuxServerImpl#initializeMetaChannels
protected void initializeMetaChannels() {
//握手<8>
createChannelIfAbsent(Channel.META_HANDSHAKE);
//续约连接
createChannelIfAbsent(Channel.META_CONNECT);
//订阅渠道
createChannelIfAbsent(Channel.META_SUBSCRIBE);
//取消订阅
createChannelIfAbsent(Channel.META_UNSUBSCRIBE);
//断开连接
createChannelIfAbsent(Channel.META_DISCONNECT);
}
<4>
org.cometd.server.BayeuxServerImpl#initializeJSONContext
protected void initializeJSONContext() throws Exception {
//默认通过 option jsonContext去获取
Object option = getOption(AbstractServerTransport.JSON_CONTEXT_OPTION);
if (option == null) {
//未配置则获取默认的
_jsonContext = new JettyJSONContextServer();
} else {
//如果我们有配置的是class全名称
if (option instanceof String) {
Class<?> jsonContextClass = Thread.currentThread().getContextClassLoader().loadClass((String)option);
if (JSONContextServer.class.isAssignableFrom(jsonContextClass)) {
_jsonContext = (JSONContextServer)jsonContextClass.getConstructor().newInstance();
} else {
throw new IllegalArgumentException("Invalid " + JSONContextServer.class.getName() + " implementation class");
}
} else if (option instanceof JSONContextServer) {//如果是context对象
_jsonContext = (JSONContextServer)option;
} else {
throw new IllegalArgumentException("Invalid " + JSONContextServer.class.getName() + " implementation class");
}
}
_options.put(AbstractServerTransport.JSON_CONTEXT_OPTION, _jsonContext);
}
<5>
org.cometd.server.BayeuxServerImpl#initializeServerTransports
protected void initializeServerTransports() {
if (_transports.isEmpty()) {
//初始化Transport 没重定义则创建默认 指定了则创建指定的 注:反射创建 会传入bayeux
String option = (String)getOption(TRANSPORTS_OPTION);
if (option == null) {
//未定义则初始化处理websocket 和长轮询的Transport处理器 JSONP的处理器
// Order is important, see #findHttpTransport()
//<6>
ServerTransport transport = newWebSocketTransport();
if (transport != null) {
addTransport(transport);
}
addTransport(newJSONTransport());
addTransport(new JSONPTransport(this));
} else {
//如果有进行类的全名称配置 根据累的全名称创建
for (String className : option.split(",")) {
ServerTransport transport = newServerTransport(className.trim());
if (transport != null) {
addTransport(transport);
}
}
if (_transports.isEmpty()) {
throw new IllegalArgumentException("Option '" + TRANSPORTS_OPTION +
"' does not contain a valid list of server transport class names");
}
}
}
//如果没有配置_allowedTransports 将transport加入到 _allowedTransports//liqiangtodo 暂时不晓得干嘛的
if (_allowedTransports.isEmpty()) {
String option = (String)getOption(ALLOWED_TRANSPORTS_OPTION);
if (option == null) {
_allowedTransports.addAll(_transports.keySet());
} else {
for (String transportName : option.split(",")) {
if (_transports.containsKey(transportName)) {
_allowedTransports.add(transportName);
}
}
if (_allowedTransports.isEmpty()) {
throw new IllegalArgumentException("Option '" + ALLOWED_TRANSPORTS_OPTION +
"' does not contain at least one configured server transport name");
}
}
}
//逐个调用transport init方法完成Transport的初始化 Transport 内部的相关参数自定义配置可以通过Option拿到
List<String> activeTransports = new ArrayList<>();
for (String transportName : _allowedTransports) {
ServerTransport serverTransport = getTransport(transportName);
if (serverTransport instanceof AbstractServerTransport) {
//调用init方法进行初始化
((AbstractServerTransport)serverTransport).init();
//加入到已激活的transpor
activeTransports.add(serverTransport.getName());
}
}
if (_logger.isDebugEnabled()) {
_logger.debug("Active transports: {}", activeTransports);
}
}
<6>
org.cometd.server.BayeuxServerImpl#newWebSocketTransport
private ServerTransport newWebSocketTransport() {
try {
ClassLoader loader = Thread.currentThread().getContextClassLoader();
//加载服务端websoket组件 websocket组件
loader.loadClass("javax.websocket.server.ServerContainer");
//初始化websoketTransport
String transportClass = "org.cometd.server.websocket.javax.WebSocketTransport";
ServerTransport transport = newServerTransport(transportClass);
if (transport == null) {
_logger.info("JSR 356 WebSocket classes available, but " + transportClass +
" unavailable: JSR 356 WebSocket transport disabled");
}
return transport;
} catch (Exception x) {
return null;
}
}
<7>
org.cometd.server.BayeuxServerImpl#newExecutor
private Executor newExecutor() {
String name = _name + "-Executor";
//默认线程大小为128
int maxThreads = (int)getOption(EXECUTOR_MAX_THREADS, 128);
QueuedThreadPool executor = new QueuedThreadPool(maxThreads, 0);
executor.setName(name);
executor.setReservedThreads(0);
return executor;
}
<8>
org.cometd.server.BayeuxServerImpl#createChannelIfAbsent
@Override
public MarkedReference<ServerChannel> createChannelIfAbsent(String channelName, Initializer... initializers) {
ChannelId channelId;
boolean initialized = false;
//尝试根据channelName获取 判断是否存在
ServerChannelImpl channel = _channels.get(channelName);
if (channel == null) {
// Creating the ChannelId will also normalize the channelName.
//尝试通过处理过的channelId获取
channelId = new ChannelId(channelName);
String id = channelId.getId();
if (!id.equals(channelName)) {
channelName = id;
channel = _channels.get(channelName);
}
} else {
channelId = channel.getChannelId();
}
//表示没有被初始化
if (channel == null) {
//新建一个channel
ServerChannelImpl candidate = new ServerChannelImpl(this, channelId);
//放入_channels
channel = _channels.putIfAbsent(channelName, candidate);
if (channel == null) {
// My candidate channel was added to the map, so I'd better initialize it
channel = candidate;
if (_logger.isDebugEnabled()) {
_logger.debug("Added channel {}", channel);
}
try {
//通知 Initializer实现 可以对ServerChannelImpl做自定义配置
for (Initializer initializer : initializers) {
notifyConfigureChannel(initializer, channel);
}
//调用listeners中ChannelListener的configureChannel方法可以对channel进行自定义配置
for (BayeuxServer.BayeuxServerListener listener : _listeners) {
if (listener instanceof ServerChannel.Initializer) {
notifyConfigureChannel((Initializer)listener, channel);
}
}
} finally {
channel.initialized();
}
//调用listeners中ChannelListener的channelAdded表示已经被初始化
for (BayeuxServer.BayeuxServerListener listener : _listeners) {
if (listener instanceof BayeuxServer.ChannelListener) {
notifyChannelAdded((ChannelListener)listener, channel);
}
}
initialized = true;
}
} else {
channel.resetSweeperPasses();
// Double check if the sweeper removed this channel between the check at the top and here.
// This is not 100% fool proof (e.g. this thread is preempted long enough for the sweeper
// to remove the channel, but the alternative is to have a global lock)
_channels.putIfAbsent(channelName, channel);
}
// Another thread may add this channel concurrently, so wait until it is initialized
channel.waitForInitialized();
return new MarkedReference<>(channel, initialized);
}
这篇关于websocket-cometd源码阅读-初始化(二)的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
您可能喜欢
-
怎么实现ansible playbook 备份代码中命名包含时间戳功能？-icode9专业技术文章分享11-22
-
ansible 的archive 参数是什么意思？-icode9专业技术文章分享11-22
-
ansible 中怎么只用archive 排除某个目录？-icode9专业技术文章分享11-22
-
exclude_path参数是什么作用？-icode9专业技术文章分享11-22
-
微信开放平台第三方平台什么时候调用数据预拉取和数据周期性更新接口？-icode9专业技术文章分享11-22
-
uniapp 实现聊天消息会话的列表功能怎么实现？-icode9专业技术文章分享11-22
-
在Mac系统上将图片中的文字提取出来有哪些方法？-icode9专业技术文章分享11-22
-
excel 表格中怎么固定一行显示不滚动？-icode9专业技术文章分享11-22
-
怎么将 -rwxr-xr-x 修改为 drwxr-xr-x？-icode9专业技术文章分享11-22
-
在Excel中怎么将小数向上取整到最接近的整数？-icode9专业技术文章分享11-22
-
Excel中常见的取整函数有哪些？-icode9专业技术文章分享11-22
-
错误信息：does not match the actual quantity supplied 提示是什么意思？-icode9专业技术文章分享11-22
-
WebSocket学习：初学者的简单指南11-21
-
获取apk的md5值有哪些方法？-icode9专业技术文章分享11-20
-
xml报文没有传 IdentCode ，为什么正常解析没报错呢？-icode9专业技术文章分享11-20
栏目导航
