皮皮网

1.eworkerԴ??
2.图解+源码讲解 Nacos 客户端动态监听配置机制
3.ForkjoinPool -1

eworkerԴ??

       文件处理一直都是前端人的心头病，如何控制好文件大小，文件太大上传不了，文件下载时间太长，tcp直接给断开了?等

效果

       为了方便大家有意义的学习，这里就先放效果图，红酒交易盘源码如果不满足直接返回就行，不浪费大家的时间。

文件上传

       文件上传实现，分片上传，暂停上传，恢复上传，文件合并等

文件下载

       为了方便测试，我上传了1个1g的大文件拿来下载，前端用的是流的方式来保存文件的，具体的可以看这个apiTransformStream

正文

       本项目的地址是：/post/

       requestIdleCallback有不明白的可以看这里：/post/

       接下来咋们来计算文件的hash,计算文件的hash需要使用spark-md5这个库，

全量计算文件hashexportasyncfunctioncalcHashSync(file:File){ //对文件进行分片，每一块文件都是分为2MB，这里可以自己来控制constsize=2**;letchunks:any[]=[];letcur=0;while(cur<file.size){ chunks.push({ file:file.slice(cur,压力测试平台源码cur+size)});cur+=size;}//可以拿到当前计算到第几块文件的进度lethashProgress=0returnnewPromise(resolve=>{ constspark=newSparkMD5.ArrayBuffer();letcount=0;constloadNext=(index:number)=>{ constreader=newFileReader();reader.readAsArrayBuffer(chunks[index].file);reader.onload=e=>{ //累加器不能依赖index，count++;//增量计算md5spark.append(e.target?.resultasArrayBuffer);if(count===chunks.length){ //通知主线程，计算结束hashProgress=;resolve({ hashValue:spark.end(),progress:hashProgress});}else{ //每个区块计算结束，通知进度即可hashProgress+=/chunks.length//计算下一个loadNext(count);}};};//启动loadNext(0);});}

       全量计算文件hash，在文件小的时候计算是很快的，但是在文件大的情况下，计算文件的hash就会非常慢，并且影响主进程哦?

抽样计算文件hash

       抽样就是取文件的一部分来继续，原理如下：

/***抽样计算hash值大概是1G文件花费1S的时间**采用抽样hash的方式来计算hash*我们在计算hash的时候，将超大文件以2M进行分割获得到另一个chunks数组，阅读app源码定制*第一个元素(chunks[0])和最后一个元素(chunks[-1])我们全要了*其他的元素(chunks[1,2,3,4....])我们再次进行一个分割，这个时候的分割是一个超小的大小比如2kb，我们取*每一个元素的头部，尾部，中间的2kb。*最终将它们组成一个新的文件，我们全量计算这个新的文件的hash值。*@paramfile{ File}*@returns*/exportasyncfunctioncalcHashSample(file:File){ returnnewPromise(resolve=>{ constspark=newSparkMD5.ArrayBuffer();constreader=newFileReader();//文件大小constsize=file.size;letoffset=2**;letchunks=[file.slice(0,offset)];//前面2mb的数据letcur=offset;while(cur<size){ //最后一块全部加进来if(cur+offset>=size){ chunks.push(file.slice(cur,cur+offset));}else{ //中间的前中后去两个字节constmid=cur+offset/2;constend=cur+offset;chunks.push(file.slice(cur,cur+2));chunks.push(file.slice(mid,mid+2));chunks.push(file.slice(end-2,end));}//前取两个字节cur+=offset;}//拼接reader.readAsArrayBuffer(newBlob(chunks));//最后Kreader.onload=e=>{ spark.append(e.target?.resultasArrayBuffer);resolve({ hashValue:spark.end(),progress:});};});}

       这个设计是不是发现挺灵活的，真是传播客改名源码个人才哇

       在这两个的基础上，咋们还可以分别使用web-worker和requestIdleCallback来实现，源代码在hereヾ(≧▽≦*)o

       这里把我电脑配置说一下，公司给我分的电脑配置比较lower,8g内存的老机器。计算(3.3g文件的)hash的结果如下:

       结果很显然，全量无论怎么弄，都是比抽样的更慢。

文件分片的方式

       这里可能大家会说，文件分片方式不就是等分吗，其实还可以根据网速上传的海鸥优选app源码速度来实时调整分片的大小哦！

consthandleUpload1=async(file:File)=>{ if(!file)return;constfileSize=file.sizeletoffset=2**letcur=0letcount=0//每一刻的大小需要保存起来，方便后台合并constchunksSize=[0,2**]constobj=awaitcalcHashSample(file)as{ hashValue:string};fileHash.value=obj.hashValue;//todo判断文件是否存在存在则不需要上传，也就是秒传while(cur<fileSize){ constchunk=file.slice(cur,cur+offset)cur+=offsetconstchunkName=fileHash.value+"-"+count;constform=newFormData();form.append("chunk",chunk);form.append("hash",chunkName);form.append("filename",file.name);form.append("fileHash",fileHash.value);form.append("size",chunk.size.toString());letstart=newDate().getTime()//todo上传单个碎片constnow=newDate().getTime()consttime=((now-start)/).toFixed(4)letrate=Number(time)///速率有最大和最小可以考虑更平滑的过滤比如1/tanif(rate<0.5)rate=0.5if(rate>2)rate=2offset=parseInt((offset/rate).toString())chunksSize.push(offset)count++}//todo可以发送合并操作了} ATTENTION!!!?如果是这样上传的文件碎片，如果中途断开是无法续传的（每一刻的网速都是不一样的），除非每一次上传都把chunksSize(分片的数组)保存起来哦

控制/post/

图解+源码讲解 Nacos 客户端动态监听配置机制

       图解+源码讲解 Nacos 客户端动态监听配置机制

       在人生中第一要紧的是发现自己。为了这个目的，各位时常需要孤独和深思 —— 南森 Nacos 源码分析系列相关文章

       从零开始看 Nacos 源码环境搭建

       图解+源码讲解 Nacos 客户端发起注册流程

       图解+源码讲解 Nacos 服务端处理注册请求逻辑

       图解+源码讲解 Nacos 客户端下线流程

       图解+源码讲解 Nacos 服务端处理下线请求

       图解+源码讲解 Nacos 客户端发起心跳请求

       图解+源码讲解 Nacos 服务端处理心跳请求

       图解+源码讲解 Nacos 服务端处理配置获取请求

       图解+源码讲解 Nacos 客户端动态监听配置机制

NacosConfigAutoConfiguration

       我们看到这里面其实注入了一个 Nacos 配置刷新的关键 NacosContextRefresherBean

@Configuration@ConditionalOnProperty(name?=?"spring.cloud.nacos.config.enabled",?matchIfMissing?=?true)public?class?NacosConfigAutoConfiguration?{ //?Nacos?配置属性@Beanpublic?NacosConfigProperties?nacosConfigProperties(ApplicationContext?context)?{ if?(context.getParent()?!=?null&&?BeanFactoryUtils.beanNamesForTypeIncludingAncestors(context.getParent(),?NacosConfigProperties.class).length?>?0)?{ return?BeanFactoryUtils.beanOfTypeIncludingAncestors(context.getParent(),NacosConfigProperties.class);}return?new?NacosConfigProperties();}//?Nacos?配置刷新属性@Beanpublic?NacosRefreshProperties?nacosRefreshProperties()?{ return?new?NacosRefreshProperties();}//?Nacos?刷新历史@Beanpublic?NacosRefreshHistory?nacosRefreshHistory()?{ return?new?NacosRefreshHistory();}//?Nacos?配置管理@Beanpublic?NacosConfigManager?nacosConfigManager(NacosConfigProperties?nacosConfigProperties)?{ return?new?NacosConfigManager(nacosConfigProperties);}//?Nacos?配置刷新@Beanpublic?NacosContextRefresher?nacosContextRefresher(NacosConfigManager?nacosConfigManager,NacosRefreshHistory?nacosRefreshHistory)?{ return?new?NacosContextRefresher(nacosConfigManager,?nacosRefreshHistory);}}NacosContextRefresher 配置中心刷新public?NacosContextRefresher(NacosConfigManager?nacosConfigManager,NacosRefreshHistory?refreshHistory)?{ //?获取配置属性信息this.nacosConfigProperties?=?nacosConfigManager.getNacosConfigProperties();//?刷新历史this.nacosRefreshHistory?=?refreshHistory;//?获取配置服务this.configService?=?nacosConfigManager.getConfigService();//?是否开启刷新，是truethis.isRefreshEnabled?=?this.nacosConfigProperties.isRefreshEnabled();}获取配置服务 getConfigService

       nacosConfigManager.getConfigService()，这行代码其实就是为了创建 NcaosConfigService 对象，我们看看你是怎么创建的，其实核心代码就是通过 NacosFactory 反射创建的 NcaosConfigService 对象，这个对象是一个核心对象后续会讲到的

public?static?ConfigService?createConfigService(Properties?properties)?throws?NacosException?{ try?{ //?加载?NacosConfigService?类Class<?>?driverImplClass?=?Class.forName("com.alibaba.nacos.client.config.NacosConfigService");//?获取构造器Constructor?constructor?=?driverImplClass.getConstructor(Properties.class);//?创建实例ConfigService?vendorImpl?=?(ConfigService)?constructor.newInstance(properties);return?vendorImpl;}?catch?(Throwable?e)?{ throw?new?NacosException(NacosException.CLIENT_INVALID_PARAM,?e);}}监听器

       NacosContextRefresher 实现了 ApplicationListener ，一看这就是一个监听器了，我们看看这个在监听器里面做了什么操作

@Overridepublic?void?onApplicationEvent(ApplicationReadyEvent?event)?{ //?这是一个?CAS?操作，只设置一次if?(this.ready.compareAndSet(false,?true))?{ //?注册?Nacos?监听器对于应用this.registerNacosListenersForApplications();}}注册 Nacos 监听/**

       register Nacos Listeners. 注册Nacos监听器 */ private void registerNacosListenersForApplications() { // 默认是 true if (isRefreshEnabled()) { // 遍历Nacos属性资源中心 for (NacosPropertySource propertySource : NacosPropertySourceRepository .getAll()) { if (!propertySource.isRefreshable()) { continue; } // 获取资源ID ?String dataId = propertySource.getDataId(); // 通过组和 dataId 注册 Nacos 监听器 registerNacosListener(propertySource.getGroup(), dataId); } } }

       private void registerNacosListener(final String groupKey, final String dataKey) { // 构建 Key 信息 String key = NacosPropertySourceRepository.getMapKey(dataKey, groupKey); // 在 listenerMap中放入了 key 对应 AbstractSharedListener 响应的方法 Listener listener = listenerMap.computeIfAbsent(key, lst -> new AbstractSharedListener() { @Override public void innerReceive(String dataId, String group, String configInfo) { // 刷新次数 refreshCountIncrement(); // 记录刷新历史，就是改变历史 nacosRefreshHistory.addRefreshRecord(dataId, group, configInfo); // 发布刷新事件 applicationContext.publishEvent( new RefreshEvent(this, null, "Refresh Nacos config")); } }); // 向配置服务中添加监听器 configService.addListener(dataKey, groupKey, listener);

       }

####?向配置服务中添加监听器&nbsp;&nbsp;&nbsp;&nbsp;此时调用的是?NacosConfigService?中的?addListener?方法，但是最终执行的还是?ClientWorker?中的?addTenantListeners?方法，后面会进行分析?ClientWorker?这个类的```java@Overridepublic?void?addListener(String?dataId,?String?group,?Listener?listener)?throws?NacosException?{ //?这个?ClientWorker?worker?也是一个核心类worker.addTenantListeners(dataId,?group,?Arrays.asList(listener));}构建 CacheData 信息

       此时向 ClientWorker 中的 CacheData 中添加数据，之后遍历监听器添加到 CacheData 中

创建 CacheData 对象public?CacheData(ConfigFilterChainManager?configFilterChainManager,?String?name,?String?dataId,?String?group,String?tenant)?{ //?dataId?不能为空if?(null?==?dataId?||?null?==?group)?{ throw?new?IllegalArgumentException("dataId="?+?dataId?+?",?group="?+?group);}this.name?=?name;this.configFilterChainManager?=?configFilterChainManager;this.dataId?=?dataId;?//?设置dataIdthis.group?=?group;?//?设置组信息this.tenant?=?tenant;?//?设置租户listeners?=?new?CopyOnWriteArrayList<ManagerListenerWrap>();?//?装饰器集合this.isInitializing?=?true;//?加载缓存数据从本地磁盘this.content?=?loadCacheContentFromDiskLocal(name,?dataId,?group,?tenant);//?计算本地缓存信息的MD5this.md5?=?getMd5String(content);}向 CacheData 中添加数据public?void?addTenantListeners(String?dataId,?String?group,List<?extends?Listener>?listeners)throws?NacosException?{ //?DefaultGroupgroup?=?null2defaultGroup(group);String?tenant?=?agent.getTenant();?//?是?""//?向缓存数据中添加监听器CacheData?cache?=?addCacheDataIfAbsent(dataId,?group,?tenant);for?(Listener?listener?:?listeners)?{ cache.addListener(listener);}}public?CacheData?addCacheDataIfAbsent(String?dataId,?String?group,?String?tenant)throws?NacosException?{ //?获取Key信息String?key?=?GroupKey.getKeyTenant(dataId,?group,?tenant);//?在缓存?Map?中获取缓存数据CacheData?cacheData?=?cacheMap.get(key);//?如果不为空的情况下那么就返回，如果为空那么就创建一个?CacheDataif?(cacheData?!=?null)?{ return?cacheData;}//?创建一个?CacheData?cacheData?=?new?CacheData(configFilterChainManager,?agent.getName(),dataId,?group,?tenant);//?将创建好的?cacheData?放入缓存?Map?中CacheData?lastCacheData?=?cacheMap.putIfAbsent(key,?cacheData);//?如果缓存数据为空的话那么从配置中心拉取，不过此时不为空if?(lastCacheData?==?null)?{ //fix?issue?#?if?(enableRemoteSyncConfig)?{ String[]?ct?=?getServerConfig(dataId,?group,?tenant,?L);cacheData.setContent(ct[0]);}//?计算任务IDint?taskId?=?cacheMap.size()?/?(int)?ParamUtil.getPerTaskConfigSize();//?设置任务IDcacheData.setTaskId(taskId);lastCacheData?=?cacheData;}//?缓存数据初始化完成//?reset?so?that?server?not?hang?this?checklastCacheData.setInitializing(true);LOGGER.info("[{ }]?[subscribe]?{ }",?agent.getName(),?key);MetricsMonitor.getListenConfigCountMonitor().set(cacheMap.size());//?返回最新的缓存数据return?lastCacheData;}

       到这里 CacheData 对象 和 cacheMap 集合已经构建完成了，后续会用到这个数据的

NacosConfigService 分析

       NacosConfigService这个类在创建的时候主要做了什么事情,这这里面创建了一个 ClientWorker对象，这个对象是一个核心的类，有关于配置的一些操作都是归功于 ClientWorker类

public?NacosConfigService(Properties?properties)?throws?NacosException?{ ......this.agent?=?new?MetricsHttpAgent(new?ServerHttpAgent(properties));this.agent.start();//?核心工作类this.worker?=?new?ClientWorker(this.agent,this.configFilterChainManager,?properties);}核心配置类 ClientWorker

       分析一下这个类都在做什么事情，都有哪些核心方法 其实能看到里面有一个构造函数、添加缓存数据、添加监听器、检查配置中心相关方法、获取服务配置、解析数据响应、移除缓存数据、删除监听器以及 shutdown方法

构造函数

       看到这里其实看到了定义了两个调度线程池，一个是用于配置检测的，一个是用于执行长轮询服务的

@SuppressWarnings("PMD.ThreadPoolCreationRule")public?ClientWorker(final?HttpAgent?agent,final?ConfigFilterChainManager?configFilterChainManager,?final?Properties?properties){ this.agent?=?agent;this.configFilterChainManager?=?configFilterChainManager;//?初始化操作init(properties);//?定义一个调度线程池，只有一个线程还是守护线程this.executor?=?Executors.newScheduledThreadPool(1,?new?ThreadFactory()?{ @Overridepublic?Thread?newThread(Runnable?r)?{ Thread?t?=?new?Thread(r);t.setName("com.alibaba.nacos.client.Worker."?+?agent.getName());t.setDaemon(true);return?t;}});//?定义一个多个线程的调度线程池，线程个数和CPU?核心数有关，也是守护线程，是一个长轮询this.executorService?=?Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(),?new?ThreadFactory()?{ @Overridepublic?Thread?newThread(Runnable?r)?{ Thread?t?=?new?Thread(r);t.setName("com.alibaba.nacos.client.Worker.longPolling."?+agent.getName());t.setDaemon(true);return?t;}});//?定义一个定时的调度任务，第一次执行的时候延时1毫秒，后续毫秒调度一次this.executor.scheduleWithFixedDelay(new?Runnable()?{ @Overridepublic?void?run()?{ try?{ //?检查配置信息方法checkConfigInfo();}?catch?(Throwable?e)?{ LOGGER.error("["?+?agent.getName()?+?"]?"+?"[sub-check]?rotate?check?error",?e);}}},?1L,?L,?TimeUnit.MILLISECONDS);}检查配置服务方法

       这个 cacheMap 包含了一些任务信息，这里面的任务是怎么来的呢，他是在添加监听器的时候添加的，上面已经分析过了

public?NacosContextRefresher(NacosConfigManager?nacosConfigManager,NacosRefreshHistory?refreshHistory)?{ //?获取配置属性信息this.nacosConfigProperties?=?nacosConfigManager.getNacosConfigProperties();//?刷新历史this.nacosRefreshHistory?=?refreshHistory;//?获取配置服务this.configService?=?nacosConfigManager.getConfigService();//?是否开启刷新，是truethis.isRefreshEnabled?=?this.nacosConfigProperties.isRefreshEnabled();}0长轮询任务 LongPollingRunnable

ForkjoinPool -1

        ForkJoin是用于并行执行任务的框架， 是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。Fork就是把一个大任务切分为若干子任务并行的执行，Join就是合并这些子任务的执行结果，最后得到这个大任务的结果。

        下面是一个是一个简单的Join/Fork计算过程，将1—数字相加

        通常这样个模型，你们会想到什么？

        Release Framework ？ 常见的处理模型是什么？ task pool - worker pool的模型。 但是Forkjoinpool 采取了完全不同的模型。

        ForkJoinPool一种ExecutorService的实现，运行ForkJoinTask任务。ForkJoinPool区别于其它ExecutorService，主要是因为它采用了一种工作窃取(work-stealing)的机制。所有被ForkJoinPool管理的线程尝试窃取提交到池子里的任务来执行，执行中又可产生子任务提交到池子中。

        ForkJoinPool维护了一个WorkQueue的数组(数组长度是2的整数次方，自动增长)。每个workQueue都有任务队列(ForkJoinTask的数组)，并且用base、top指向任务队列队尾和队头。work-stealing机制就是工作线程挨个扫描任务队列，如果队列不为空则取队尾的任务并执行。示意图如下

        流程图：

        pool属性

        workQueues是pool的属性，它是WorkQueue类型的数组。externalPush和externalSubmit所创建的workQueue没有owner(即不是worker)，且会被放到workQueues的偶数位置；而createWorker创建的workQueue（即worker）有owner，且会被放到workQueues的奇数位置。

        WorkQueue的几个重要成员变量说明如下：

        这是WorkQueue的config，高位跟pool的config值保持一致，而低位则是workQueue在workQueues数组的位置。

        从workQueues属性的介绍中，我们知道，不是所有workQueue都有worker，没有worker的workQueue称为公共队列（shared queue），config的第位就是用来判断是否是公共队列的。在externalSubmit创建工作队列时，有：

        q.config = k | SHARED_QUEUE;

        其中q是新创建的workQueue，k就是q在workQueues数组中的位置，SHARED_QUEUE=1<<，注意这里config没有保留mode的信息。

        而在registerWorker中，则是这样给workQueue的config赋值的：

        w.config = i | mode;

        w是新创建的workQueue，i是其在workQueues数组中的位置，没有设置SHARED_QUEUE标记位

        scanState是workQueue的属性，是int类型的。scanState的低位可以用来定位当前worker处于workQueues数组的哪个位置。每个worker在被创建时会在其构造函数中调用pool的registerWorker，而registerWorker会给scanState赋一个初始值，这个值是奇数，因为worker是由createWorker创建，并会被放到WorkQueues的奇数位置，而createWorker创建worker时会调用registerWorker。

        简言之，worker的scanState初始值是奇数，非worker的scanstate初始值=INACTIVE=1<<，小于0（非worker的workQueue在externalSubmit中创建）。

        当每次调用signalWork（或tryRelease）唤醒worker时，worker的高位就会加1

        另外，scanState<0表示worker未激活，当worker调用runtask执行任务时，scanState会被置为偶数，即设置scanState的最右边一位为0。

        worker休眠时，是这样存储的

        worker的唤醒类似这样：

        在worker休眠的4行伪码中，让ctl的低位的值变为worker.scanState，这样下次就可以通过scanState唤醒该worker。唤醒该worker时，把该worker的preStack设置为ctl低位的值，这样下下次唤醒的worker就是scanState等于该preStack的worker。

        这里通过preStack保存下一个worker，这个worker比当前worker更早地在等待，所以形成一个后进先出的栈。

        runState是int类型的值，控制整个pool的运行状态和生命周期，有下面几个值（可以好几个值同时存在）：

        如果runState值为0，表示pool尚未初始化。

        RSLOCK表示锁定pool，当添加worker和pool终止时，就要使用RSLOCK锁定整个pool。如果由于runState被锁定，导致其他操作等待runState解锁（通常用wait进行等待），当runState设置了RSIGNAL，表示runState解锁，并通知（notifyAll）等待的操作。

        剩下4个值都跟runState生命周期有关，都可以顾名思义：

        当需要停止时，设置runState的STOP值，表示准备关闭，这样其他操作看到这个标记位，就不会继续操作，比如tryAddWorker看到STOP就不会再创建worker：

        而tryTerminate对这些生命周期状态的处理则是这样的：

        当前top和base的初始值为 INITIAL_QUEUE_CAPACITY >>>1= (1 << )>>>1 = /2。然后push一个task之后，top+=1，也就是说，top对应的位置是没有task的，最近push进来的task在top-1的位置。而base的位置则能对应到task，base对应最先放进队列的task，top-1对应最后放进队列的task。

        qlock值含义：1: locked, < 0: terminate; else 0

        即当qlock值位0时，可以正常操作，值=1时，表示锁定

        int SQMASK=0xe，则任何整数跟SQMASK位与后，得到的数就是偶数。

        证明：

        注意这里化为二进制是 ，尤其注意最右边第一位是0，任何数跟最右边第一位是0的数位与后，得到的数就是偶数，因为位与之后，第一位就是0，比如s=A&SQMASK，A可以是任意整数，然后把s按二进制进行多项式展开，则有s=2 n1+2 n2 ……+2^nn，这里n≥1，所以s可以被2整除，即s是偶数。

        所以一个数是奇数还是偶数，看其最右边第一位即可。

        我们知道workQueue有externalPush创建的和createWorker创建的worker，两种方式创建的workQueue，其放置到workQueues的位置是不同的，前者放到workQueue的偶数位置，而后者则放到奇数位置。不同workQueue找到自己在workQueues的位置的算法有点不同。

        下面看一下forkjoin框架获取workQueues中的偶数位置的workQueue的算法：

        这样就能获取workQueues的偶数位置的workQueue。m保证m & r & SQMASK这整个运算结果不会超出workQueues的下标，SQMASK保证取到的是偶数位置的workQueue。这里有一个有趣的现象，假设0到workQueues.length-1之间有n个偶数，m & r & SQMASK每次都能取到其中一个偶数，而且连续n次取到的偶数不会出现重复值，散列性非常好。而且是循环的，即1到n次取n个不同偶数，n+1到2n也是取n次不同偶数，此时n个偶数每个都被重新取一次。下面分析下r值有什么秘密，为何能保证这样的散列性

        ThreadLocalRandom内有一常量PROBE_INCREMENT = 0x9eb9，以及一个静态的probeGenerator =new AtomicInteger() ，然后每个线程的probe= probeGenerator.addAndGet(PROBE_INCREMENT)所以第一个线程的probe值是0x9eb9，第二个线程的值就是0x9eb9+0x9eb9，第三个线程的值就是0x9eb9+0x9eb9+0x9eb9以此类推，整个值是线性的，可以用y=kx表示，其中k=0x9eb9，x表示第几个线程。这样每个线程的probe可以保证不一样，而且具有很好的离散性。

        实际上，可以不用0x9eb9这个值，用任意一个奇数都是可以的，比如1。如果用1的话，probe+=1，这样每个线程的probe就都是不同的，而且具有很好的离散性。也就是说，假设有限制条件probe<n，超过n则产生溢出。则probe自加n次后才会开始出现重复值，n次前probe每次自加的值都不同。实际上用任意一个奇数，都可以保证probe自加n次后才会开始出现重复值，有兴趣可看本文最后附录部分。由于奇数的离散性，所以只要线程数小于m或者SQMASK两者中的最小值，则每个线程都能唯一地占据一个ws中的一个位置

        当一个操作是在非ForkjoinThread的线程中进行的，则称该操作为外部操作。比如我们前面执行pool.invoke，invoke内又执行externalPush。由于invoke是在非ForkjoinThread线程中进行的（这里是在main线程中进行），所以是一个外部操作，调用的是externalPush。之后task的执行是通过ForkJoinThread来执行的，所以task中的fork就是内部操作，调用的是push，把任务提交到工作队列。其实fork的实现是类似下面这样的：

        即fork会根据执行自身的线程是否是ForkJoinThread的实例来判断是处于外部还是内部。那为何要区分内外部？

        任何线程都可以使用ForkJoin框架，但是对于非ForkJoinThread的线程，它到底是怎样的，ForkJoin无法控制，也无法对其优化。因此区分出内外部，这样方便ForkJoin框架对任务的执行进行控制和优化

        forkJoinPool.invoke(task)是把任务放入工作队列，并等待任务执行。源码如下

        这里externalPush负责任务提交，externalPush源码如下：