【庄家出货公式源码】【改单源码】【idc主页源码】eworker源码

2024-11-19 05:48:19 来源:弹幕直播源码 分类:探索

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

eworker源码

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数组,idc主页源码*第一个元素(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:});};});}

       这个设计是不是发现挺灵活的,真是文华ac源码个人才哇

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

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

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

文件分片的方式

       这里可能大家会说,文件分片方式不就是等分吗,其实还可以根据网速上传的python 白化源码速度来实时调整分片的大小哦!

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源码如下:

本文地址:http://581.net.cn/news/84a376196154.html 欢迎转发