首页 文章 精选 留言 我的

精选列表

搜索[手游加固],共7832篇文章
优秀的个人博客,低调大师

敲,Ascend算子开发入门笔记分享

本文分享自华为云社区《Ascend算子开发入门笔记》,作者: JeffDing 。 基础概念 什么是Ascend C Ascend C是CANN针对算子开发场景推出的编程语言,原生支持C和C++标准规范,最大化匹配用户开发习惯;通过多层接口抽象、自动并行计算、孪生调试等关键技术,极大提高算子开发效率,助力AI开发者低成本完成算子开发和模型调优部署。 使用Ascend C开发自定义算子的优势 C/C++原语编程,最大化匹配用户的开发习惯 编程模型屏蔽硬件差异,编程范式提高开发效率 多层级API封装,从简单到灵活,兼顾易用与高效 孪生调试,CPU侧模拟NPU侧的行为,可优化在CPU侧调试 昇腾计算架构CANN CANN 介绍网站:https://www.hiascend.com/software/cann AI Core是NPU卡的计算核心,NPU内部有多个AI Core。每个AI Core相当于多核CPU中的一个核心 SIMD SIMD,也就是单指令多数据计算,一条指令可以处理多个数据:Ascend C编程API主要是向量计算API和矩阵运算API,计算API都是SIMD样式 并行计算之SPMD数据并行与流线型并行 SPMD数据并行原理 启动一组进程,他们运行的相同程序 把待处理数据切分,把切分后数据分片分发给不同进程处理 每个进程对自己的数据分片进行3个任务T1、T2、T3的处理 流水线并行原理 启动一组进程 对数据进行切分 每个进程都处理所有的数据切片,对输入数据分片只做一个任务的处理 Ascend C编程模型与范式 并行计算架构抽象 使用Ascend C编程语言开发的算子运行在AI Core上,AI Core是昇腾AI处理器中的计算核心 一个AI处理器内部有多个AI Core,AI Core中包含计算单元、存储单元、搬运单元等核心组件 计算单元包括了三种基础计算资源 Scalar计算单元:执行地址计算、循环控制等标量计算工作,并把向量计算、矩阵计算、数据半圆、同步指令发射给对应单元执行 Cube计算单元:负责执行矩阵运算 Vector计算单元:负责执行向量计算 搬运单元负责在Global Memory和Local Memory之间搬运数据,包含搬运单元MTE(Memory Transfer Engine,数据搬入单元),MTE3(数据搬出单元) 存储单元为AI Core的内部存储,统称为Local Memory与此相对应,AI Core的外部存储称之为Global Memory 异步指令流 Scalar计算单元读取指令序列,并把向量计算、矩阵计算、数据搬运指令发射给对应单元的指令队列,向量计算单元、矩阵计算单元、数据搬运单元异步的并行执行接收到的指令 同步信号流 指令间可能存在依赖关系,为了保证不同指令队列间的指令按照正确的逻辑关系执行,Scalar计算单元也会给对应单元下发同步指令 计算数据流 DMA搬入单元把数据搬运到Local Memory,Vector/Cube计算单元完成数据计算,并把计算结构写回Local Memory,DMA搬出单元把处理好的数据搬运回Global Memory SPMD编程模型介绍 Ascend C算子编程是SPMD的编程,将需要处理的数据拆分并行分布在多个计算核心上运行多个AI Core共享相同的指令代码,每个核上的运行实例唯一的区别是block_idx不同block的类似于进程,block_idx就是标识进程唯一性的进程ID,编程中使用函数GetBlockIdx()获取ID 核函数编写及调用 核函数(Kernel Function)是Acend C算子设备侧的入口。Ascend C允许用户使用核函数这种C/C++函数的语法扩展来管理设备侧的运行代码,用户在核函数中实现算子逻辑的编写,例如自定义算子类及其成员函数以实现该算子的所有功能。核函数是主机侧和设备侧连接的桥梁 核函数是直接在设备侧执行的代码。在核函数中,需要为在一个核上执行的代码规定要进行的数据访问和计算操作,SPMD编程模型允许核函数调用时,多个核并行地执行同一个计算任务。 使用函数类型限定符 除了需要按照C/C++函数声明的方式定义核函数之外,还要为核函数加上额外的函数类型限定符,包含__global__和__aicore__ 使用__global__函数类型限定符来标识它是一个核函数,可以被<<<…>>>调用;使用__aicore__函数类型限定符来标识该函数在设备侧AI Core上执行 __gloabl__ __aircore__ void kernel_name(argument list); 使用变量类型限定符 为了方便:指针入参变量统一的类型定义为__gm__uint8_t* 用户可统一使用uint8_t类型的指针,并在使用时转化为实际的指针类型;亦可直接传入实际的指针类型 规则或建议 核函数必须具有void返回类型 仅支持入参为指针类型或C/C++内置数据类型(Primitive Data Types),如:half* s0、flat* s1、int32_t c 提供了一个封装的宏GM_ADDR来避免过长的函数入参列表 #define GM_ADDR __gm__ unit8_t* __restrict__ 调用核函数 核函数的调用语句是C/C++函数调用语句的一种扩展 常见的C/C++函数调用方式是如下的形式: function_name(argument list); 核函数使用内部调用符<<<…>>>这种语法形式,来规定核函数的执行配置: kernel_name<<<blockDim, l2ctrl, stream>>>(argument list); 注:内核调用符仅可在NPU模式下编译时调用,CPU模式下编译无法识别该符号 blocakdim,规定了核函数将会在几个核上执行,每个执行该核函数的核会被分配一个逻辑ID,表现为内置变量block_idx,编号从0开始,可为不同的逻辑核定义不同的行为,可以在算子实现中使用GetBlockIDX()函数来获得。 l2ctl,保留函数,展示设置为固定值nullptr。 stream:类型为aclrtStream,stream是一个任务队列,应用程序通过stream来管理任务的并行 使用内核调用符<<<…>>>调用核函数: HelloWorld<<<8, nullptr, stream>>>(fooDevice)); blockDim设置为8,表示在8个核上调用了HelloWorld核函数,每个核都会独立且并行地执行该核函数Stream可以通过aclrtCreateStream来创建,它的作用是在当前进程或线程中显式创建一个aclrtStream argument list设置为cooDevice这1个入参。 核函数的调用是异步的,核函数的调用结束后,控制权立刻返回给主机侧。 强制主机侧程序等待所有核函数执行完毕的API(阻塞应用程序运行,直到指定Stream中的所有任务都完成,同步接口)为aclrtSynchronizeStream aclError aclrtSynchronizeStream(aclrtStream stream); 编程API介绍 Ascend C算子采用标准C++语法和一组类库API进行编程 计算类API:标量计算API、向量计算API、矩阵计算API、分别实现调用Scalar计算单元、Vector计算单元、Cube计算单元 数据搬运API:基于Local Memory数据进行计算、数据需要先从Gloabl Memory搬运至Local Memory,再使用计算接口完成计算,最后从Local Memory搬出至Gloabl Memory。比如DataCopy接口 内存管理API:用于分配管理内存,比如AllocTensor、FreeTensor接口 任务同步API:完成任务间的通信和同步,比如EnQue、DeQue接口。不同的指令异步并行执行,为了保证不同指令队列间的指令按照正确的逻辑关系执行,需要向不同的组件发送同步指令 Ascend C API用于计算的基本数据类型都是Tensor:GlobalTensor和LocalTensor 4级API定义 4级API定义:API根据用户使用的场景分为4级 3级API,运算符重载,支持+, - ,* ,/ ,= ,| ,& ,^ ,> ,< ,>- ,<= 实现计算的简单表述,类似dst=src1+src2 2级连续计算API,类似Add(dst,src1,src2,count),针对源操作数的连续COUNT个数据进行计算连续写入目的操作数,解决一维tensor的连续count个数据的计算问题 1级slice计算API,解决多维数据中的切片计算问题(开发中) 0级丰富功能计算API,可以完整发挥硬件优势的计算API,该功能可以充分发挥CANN系列芯片的强大指令,支持对每个操作数的repeattimes,repetstride,MASK的操作。调用类似:Add(dst,src1,src2,repeatTimes,repeatParams); 流水编程范式介绍 Ascend C编程范式把算子内部的处理程序,分成多个流水任务(Stage),以张量(Tensor)为数据载体,以队列(Queue)进行任务之间的通信与同步,以内存管理模块(Pipe)管理任务间的通信内存。 快速开发编程的固定步骤 统一代码框架的开发捷径 使用者总结出的开发经验 面向特定场景的编程思想 定制化的方法论开发体验 抽象编程模型“TPIPE并行计算" 针对各代Davinci芯片的复杂数据流,根据实际计算需求,抽象出并行编程范式,简化流水并行 Ascend C的并行编程式范式核心要素 一组并行计算任务 通过队列实现任务之间的通信和同步 程序员自主表达对并行计算任务和资源的调度 典型的计算范式 基本的矢量编程范式:计算任务分为CopyIn,Compute,CopyOut 基本的矩阵编程范式:计算任务分为CopyIn,Compute,Aggregate,CopyOut 复杂的矢量/矩阵编程范式,通过将矢量/矩阵的Out/ln组合在一起的方式来实现复杂计算数据流 流水任务 流水任务(Stage)指的是单核处理程序中主程序调度的并行任务。 在核函数内部,可以通过流水任务实现数据的并行处理来提升性能 举例来说,单核处理程序的功能可以拆分为3个流水任务:Stage1、Stage2、Stage3,每个任务专注数据切片的处理。Stage间的剪头表达数据间的依赖,比如Stage1处理完Progress1之后,Stage2才能对Proress1进行处理。 若Progres的n=3,待处理的数据被切分成3片,对于同一片数据,Stage1、Stage2、Stage3之间的处理具有依赖关系,需要串行处理;不同的数据切片,同一时间点,可以有多个流水任务Stage在并行处理,由此达到任务并行、提升性能的目的 任务间通信和同步 数据通信与同步的管理者 Ascend C中使用Queue队列完成任务之间的数据通信和同步,Queue提供了EnQue、DeQue等基础API。 Queue队列管理NPU上不同层级的物理内存时,用一种抽象的逻辑位置(QuePosition)来表达各个级别的存储(Storage Scope),代替了片上物理存储的概念,开发者无需感知硬件架构。 矢量编程中Queue类型(逻辑位置)包括:VECIN、VECOUT 数据的载体 Ascend C使用GlobalTensor和LocalTensor作为数据的基本操作单元,它是各种指令API直接调用的对象,也是数据的载体 矢量编程任务间通信和任务 矢量编程中的逻辑位置(QuePosition):搬入数据的存放位置:VECIN、搬出数据的存放位置:VECOUT。 矢量编程主要分为CopyIn、Compute、CopyOut三个任务 CopyIn任务中将输入数据从GlobalTensor搬运至LocalTensor后,需要使用EnQue将LocalTensor放入VECIN的Queue中 Compute任务等待VECIN的Queue中LocalTensor出队之后才可以进行矢量计算,计算完成后使用EnQue将计算结果LocalTensor放入VECOUT的Queue中 CopyOut任务等待VECOUT的Queue中Localtensor出队,再将其拷贝至GlobalTensor Stage1:CopyIn任务 使用DataCopy接口将GlobalTensor拷贝纸LocalTensor 使用EnQue将LocalTensor放入VECIN的Queue中 Stage2:Compute任务 使用DeQue从VECIN中取出LocalTensor 使用Ascend C指令API完成矢量计算:Add 使用EnQue将结果LocalTensor放入VECOUT的Queue中 Stage3:CopyOut任务 使用DeQue接口从VECOUT的Queue中取出LocalTensor 使用DataCopy接口将LocalTensor拷贝至GlobalTensor 内存管理 任务见数据传递使用到的内存统一由内存管理模块Pipe进行管理。 Pipe作为片上内存管理者,通过InitBuffer接口对外提供Queue内存初始化功能,开发者可以通过该接口为指定的Queue分配内存。 Queue队列内存初始化完成后,需要使用内存时,通过调用AllocTensor来为LocalTensor分配内存给Tensor,当创建的LocalTensor完成相关计算无需再使用时,再调用FreeTensor来回收LocalTensor的内存 临时变量内存管理 编程过程中使用到的临时变量内存同样通过Pipe进行管理。临时变量可以使用TBuf数据结构来申请指定QuePosition上的存储空间,并使用Get()来将分配到的存储空间分配给新的LocalTensor从TBuf上获取全部长度,或者获取指定长度的LocalTensor LocalTensor<T> Get<T>(); LocalTensor<T> Get<T>(uint32_t len); Tbuf及Get接口的示例 //为TBuf初始化分配内存,分配内存长度为1024字节 TPipe pipe; TBuf<TPosition::VECIN> calcBuf; //模板参数为QuePosition中的VECIN类型 uint32_t byteLen = 1024; pipe.InitBuffer(calcBuf,byteLen); //从calcBuf获取Tensor,Tensor为pipe分配的所有内存大小,为1024字节 LocalTensor<int32_t> tempTensor1 = calcBuf.Get<int32_t>(); //从calcBuf获取Tensor,Tensor为128个int32_t类型元素的内存大小,为512字节 LocalTensro<int32_t> tempTensor1 = calcBuf.Get<int32_t>(128); 使用TBuf申请的内存空间只能参与计算,无法执行Queue队列的入队出队操作 Ascend C矢量编程 算子分析 开发流程 算子分析:分析算子的数学表达式、输入、输出以及计算逻辑的实现,明确需要调用的Ascend接口 核函数定义:定义Ascend算子入口函数 根据矢量编程范式实现算子类:完成核函数的内部实现 以ElemWise(ADD)算子为,数学公式 为简单起见,设定张量x,y,z为固定shape(8,2048),数据类型dtype为half类型,数据排布类型format为ND,核函数名称为add_custom 算子分析 明确算子的数学表达式及计算逻辑 Add算子的数学表达式为 计算逻辑:输入数据需要先搬入到片上存储,然后使用计算接口完成两个加法运算,得到最终结果,再搬出到外部存储 明确输入输出 Add算子有两个: 输入数据类型为half,输出数据类型与输入数据类型相同。输入支持固定shape(8,2048),输出shape与输入shape相同,输入数据排布类型为ND 确定核函数名称和参数 自定义核函数明,如add_custom,根据输入输出,确定核函数有3个入参x,y,z x,y为输入在GlobalMemory上的内存地址,z为输出在globalMemory上的内存地址 确定算子实现所需接口 涉及内外部存储间的数据搬运,使用数据搬移接口:DataCopy实现 涉及矢量计算的加法操作,使用矢量双目指令:Add实现 使用到LocalTensor,使用Queue队列管理,会使用到Enque,Deque等接口。 算子实现 核函数定义 在add_custom核函数的实现中示例化KernelAdd算子类,调用Init()函数完成内存初始化,调用Process()函数完成核心逻辑。 注:算子类和成员函数名无特殊要求,开发者可根据自身的C/C++编码习惯,决定核函数中的具体实现。 // implementation of kenel function extern "C" __global__ __aicore__ void add_custom(__gm__ uint8_t* x, __gm__ uint8_t* y, __gm__ uint8_t* z) { kernelAdd op; op.Init(x,y,z); op.Process(); } 对于核函数的调用,使用内置宏__CCE_KT_TEST__来标识<<<…>>>仅在NPU模式下才会编译到(CPU模式g++没有<<<…>>>的表达),对核函数的调用进行封装,可以在封装函数中补充其他逻辑,这里仅展示对于核函数的调用。 #ifndef __CCE_KT_TEST__ // call of kernel function void add_custom_do(uint32_t blockDim, void* l2ctrl, void* stream, uint8_t* x, uint8_t* y, uint8_t* z) { add_custom<<<blockDim, l2ctrl, stream>>>(x,y,z); } 算子类实现 CopyIn任务:将Global Memory上的输入Tensor xGm和yGm搬运至Local Memory,分别存储在xlocal,ylocal。 Compute任务:对xLocal,yLocal执行加法操作,计算结果存储在zlocal中。 CopyOut任务:将输出数据从zlocal搬运至Global Memory上的输出tensor zGm中。 CopyIn.Compute任务间通过VECIN队列和inQueueX,inQueueY进行通信和同步。 Compute,CopyOut任务间通过VECOUT和outQueueZ进行通信和同步。 pipe内存管理对象对任务间交互使用到的内存、临时变量是用到的内存进行统一管理。 向量加法z=x+y 代码样例 TPIPE流水式编程范式 算子类实现 算子类类名: KernelAdd 初始化函数Init()和核心处理函数Process() 三个流水任务:CopyIn(),Compute(),CopyOut() Process的含义 TQue模板的BUFFER)NUM的含义: 该Queue的深度,double buffer优化技巧 class KernelAdd{ public: __aicore__ inline KernelAdd() //初始化函数,完成内存初始化相关操作 __aicore__ inline voide Init(__gm__ uint8_t* x, __gm__ uint8_t* y, __gm__ uint8_t* z){} // 核心处理函数,实现算子逻辑,调用私有成员函数CopyIn,Compute,CopyOut完成算子逻辑 __aicore__ inline void Process(){} private: // 搬入函数,完成CopyIn阶段的处理,被Process函数调用 __aicore__ inline void CopyIn(int32_t process){} // 计算函数,完成Compute阶段的处理,被Process函数调用 __aicore__ inline void Compute(int32_t process){} // 搬出函数,完成CopyOut阶段的处理,被Process函数调用 __aicore__ inline void CopyOut(int32_t process){} private: // pipe内存管理对象 TPipe pipe; // 输入数据Queue队列管理对象,QuePosition为VECIN TQue<QuePosition::VECIN, BUFFER_NUM> inQueueX, inQueueY; // 输出数据Queue队列管理对象,QuePosition为VECOUT TQue<QuePosition::VECOUT, BUFFER_NUM> outQueueZ; // 管理输入输出的Global Memory内存地址的对象,其中xGm,yGm为输入,zGm为输出 GlobalTensor<half> xGm, yGm ,zGm; }; Init()函数实现 使用多核并行计算,需要将数据切片,获取到每个核实际需要处理的在Global Memory上的内存偏移地址。 数据整体长度TOTAL_LENGTH为8 * 2048,平均分配到8个核上运行,每个核上处理的数据大小BLOCK_LENGTH为2048,block_idx为核的逻辑ID,(gmhalf*)x + GetBlockIdx() * BLOCK_LENGTH即索引为block_idx的核的输入数据在Global Memory上的内存偏移地址 对于单核处理数据,可以进行数据切块(Tiling),将数据切分成8快,切分后的每个数据块再次切分成BUFFER_NUM=2块,可开启double buffer,实现流水线之间的并行。 单核需要处理的2048个数据切分成16块,每块TILE_LENGTH=128个数据,Pipe为inQueueX分配了BUFFER_NUM块大小为TITLE_LENGTH * sizeof(half)个字节的内存块,每个内存块能容纳TILE_LENGTH=128个half类型数据 代码示例 constexpr int32_t TOTAL_LENGTH = 8 * 2048; //total length of data constexpr int32_t USE_CORE_NUM = 8; //num of core used constexpr int32_t BLOCK_LENGTH = TOTAL_LENGTH / USE_CORE_NUM; //length computed of each ccore constexpr int32_t TILE_NUM = 8; //split data into 8 tiles constexpr int32_t BUFFER_NUM = 2; //tensor num for each queue constexpr int32_t TILE_LENGTH = BLOCK_LENGTH / TILE_NUM / BUFFER_NUM; //seperate to 2 parts, due to double buffer __aicore__ inline void Init(GM_ADDR x, GM_ADDR y, GM_ADDR z) { //get start index for current core,core parallel xGm,SetGlobalBuffer((__gm__ half*)x * BLOCK_LENGTH * GetBlockIdx(), BLOCK_LENGTH); yGm,SetGlobalBuffer((__gm__ half*)y * BLOCK_LENGTH * GetBlockIdx(), BLOCK_LENGTH); zGm,SetGlobalBuffer((__gm__ half*)z * BLOCK_LENGTH * GetBlockIdx(), BLOCK_LENGTH); //pipe alloc memory to queue,the unit is Bytes pipe.InitBuffer(inQueueX, BUFFER_NUM, TILE_LENGTH * sizeof(half)); pipe.InitBuffer(inQueueY, BUFFER_NUM, TILE_LENGTH * sizeof(half)); pipe.InitBuffer(outQueueZ, BUFFER_NUM, TILE_LENGTH * sizeof(half)); } Process()函数实现 代码示例 __aicore__ inline void Process() { // loop count need to be doubled, due to double buffer constexpr int32_t loopCount = TILE_NUM * BUFFER_BUM; // tiling strategy, pipeline prallel for (int32_t i = 0; i < loopCount; i++) { CopyIn(i); Compute(i); CopyOut(i); } } __aicore__ inline void CopyIn(int32_t progress) { // alloc tensor from queue memory LocalTensor<half> xLocal = inQueueX.AllocTensor<half>(); LocalTensor<half> yLocal = inQueueY.AllocTensor<half>(); // copy progress_th tile from global tensor to local tensor DataCopy(xLocal,xGm[progress * TILE_LENGTH], TILE_LENGTH); DataCopy(xLocal,yGm[progress * TILE_LENGTH], TILE_LENGTH); // enque input tensors to VECIN queue inQueueX.EnQue(xLocal); inQueueY.EnQue(yLocal); } __aicore__ inline void Compute(int32_t progress) { //dque input tensors from VECIN queue LocalTensor<half> xLocal = inQueueX.DeQue<half>(); LocalTensor<half> yLocal = inQueueY.DeQue<half>(); LocalTensor<half> zLocal = outQueueZ.AllocTensor<half>(); // call Add instr for computation Add(zLocal, xLocal, yLocal, TILE_LENGTH); // enque the output tensor to VECOUT queue outQueueZ.EnQue<half>(zLocal)l // free input tensors for reuse inQueueX.FreeTensor(xLocal); inQueueY.FreeTensor(yLocal); } __aicore__ inline void CopyOut(int32_t progress) { //deque output tensor form VECOUT queue LocalTensor<half> zLocal = outQueueZ.Deque<half>(); // copy progress_th tile form local tensor to global tensor DataCopy(zGm[progress * TILE_LENGTH), zlocal, TILE_LENGTH); // free outpupt tensor for reuse outQueueZ.freeTensor(zLocal); } double buffer机制 double buffer通过将数据搬运与矢量计算并执行以隐藏数据搬运时间并降低矢量指令的等待时间,最终提高矢量计算单元的利用效率1个Tensor同一时间只能进行搬入、计算和搬出三个流水任务中的一个,其他两个流水任务涉及的硬件但愿则处于Idle状态。 如果将待处理的数据一分为而,比如Tensor1、Tensor2。 当矢量计算单元对于Tensor1进行Compute时,Tensor2可以进行CopyIn的任务 当矢量计算单元对于Tensor2进行Compute时,Tensor1可以进行CopyOut的任务 当矢量计算单元对于Tensor2进行CopyOut时,Tensor2可以进行CopyIn的任务 由此,数据的进出搬运和矢量计算之间实现你并行,硬件单元闲置问题得以有效缓解 Ascend C 算子调用 HelloWorld样例 运行CPU模式包含的头文件 运行NPU模式包含的头文件 核函数的定义 内置宏__CE_KT_TEST__:区分运行CPU模式或NPU模式逻辑的标志 主机侧执行逻辑:负责数据在主机侧内存的申请,主机到设备的拷贝,核函数执行同步和回收资源的工作 设备侧执行逻辑 主机侧执行CPU模式逻辑:使用封装的执行宏ICPU_RUN_KF 主要包括: gMAlloc(…):申请CPU模式下的内存空间 ICPU_RUN_KF:使用封装的执行宏 GmFree:释放CPU模式下的内存空间 流程 AscendCL初始化—>运行管理资源申请—>Host数据传输至Device—>执行任务并等待—>Device数据传输至Host—>运行资源释放—>AscendCL去初始化 主机侧执行NPU模式逻辑:使用内核调用符<<<…>>>重要接口 aclInit aclCreateStream aclMallocHost aclMalloc aclMemcpy <<<…>>> aclrtSynchronizeStream aclrtFree aclrtfreeHost aclrtDestoryStream aclFinalize AddCustom样例 Ascend C矢量算子样例代码 核函数源文件:add_custom.app 真值数据生成脚本:add_custom.py CmakeLists.txt:方便对多个源文件进行编译 读写数据文件辅助函数:data_utils.h 主机侧源文件:main.cpp 一键执行脚本:run.sh 组织CPU模式和NPU模式下编译的cmake脚本 点击关注,第一时间了解华为云新鲜技术~

优秀的个人博客,低调大师

撕源码!线程池核心组件源码剖析

本文选自 Doocs 开源社区旗下“源码猎人”项目,作者 AmyliaY。项目将会持续更新,欢迎 Star 关注。项目地址:https://github.com/doocs/source-code-hunter 线程池核心组件图解 看源码之前,先了解一下该组件 最主要的几个 接口、抽象类和实现类的结构关系。 该组件中,Executor 和 ExecutorService 接口 定义了线程池最核心的几个方法,提交任务 submit ()、关闭线程池 shutdown()。抽象类 AbstractExecutorService 主要对公共行为 submit()系列方法进行了实现,这些 submit()方法 的实现使用了 模板方法模式,其中调用的 execute()方法 是未实现的 来自 Executor 接口 的方法。实现类 ThreadPoolExecutor 则对线程池进行了具体而复杂的实现。 另外还有一个常见的工具类 Executors,里面为开发者封装了一些可以直接拿来用的线程池。 源码赏析 话不多说,直接上源码。(这里只看最主要的代码部分) Executor 和 ExecutorService 接口 public interface Executor { /** * 在将来的某个时间执行给定的 Runnable。该 Runnable 可以在新线程、池线程或调用线程中执行。 */ void execute(Runnable command);}public interface ExecutorService extends Executor { /** * 优雅关闭,该关闭会继续执行完以前提交的任务,但不再接受新任务。 */ void shutdown(); /** * 提交一个有返回值的任务,并返回该任务的 未来执行完成后的结果。 * Future的 get()方法 将在成功完成后返回任务的结果。 */ <T> Future<T> submit(Callable<T> task); <T> Future<T> submit(Runnable task, T result); Future<?> submit(Runnable task);} AbstractExecutorService 抽象类 /** * 该抽象类最主要的内容就是,实现了 ExecutorService 中的 submit()系列方法 */public abstract class AbstractExecutorService implements ExecutorService { /** * 提交任务 进行执行,返回获取未来结果的 Future对象。 * 这里使用了 “模板方法模式”,execute()方法来自 Executor接口,该抽象类中并未进行实现, * 而是交由子类具体实现。 */ public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null); execute(ftask); return ftask; } public <T> Future<T> submit(Runnable task, T result) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task, result); execute(ftask); return ftask; } public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); execute(ftask); return ftask; }} ThreadPoolExecutor public class ThreadPoolExecutor extends AbstractExecutorService { /** * ************** * ** 主要属性 ** * ************** */ /** 阻塞队列 */ private final BlockingQueue<Runnable> workQueue; /** 用于创建线程的 线程工厂 */ private volatile ThreadFactory threadFactory; /** 核心线程数 */ private volatile int corePoolSize; /** 最大线程数 */ private volatile int maximumPoolSize; /** * ************** * ** 构造方法 ** * ************** */ /** 最后都使用了最后一个构造方法的实现 */ public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); } public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler); } public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler); } public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null) throw new NullPointerException(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; } /** * ************** * ** 主要实现 ** * ************** */ /** 执行 Runnable任务 */ public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * 分三步进行: * * 1、如果运行的线程少于 corePoolSize,尝试开启一个新的线程;否则尝试进入工作队列 * * 2. 如果工作队列没满,则进入工作队列;否则 判断是否超出最大线程数 * * 3. 如果未超出最大线程数,则尝试开启一个新的线程;否则 按饱和策略处理无法执行的任务 */ int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false)) reject(command); } /** * 优雅关闭,在其中执行以前提交的任务,但不接受新任务。如果已关闭,则调用没有其他效果。 */ public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(SHUTDOWN); interruptIdleWorkers(); onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); } tryTerminate(); }} ThreadPoolExecutor 中的 execute()方法 执行 Runnable 任务 的流程逻辑可以用下图表示。 工具类 Executors 看类名也知道,它最主要的作用就是提供 static 的工具方法,为开发者提供各种封装好的 具有各自特性的线程池。 public class Executors { /** * 创建一个固定线程数量的线程池 */ public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); } /** * 创建一个单线程的线程池 */ public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); } /** * 创建一个缓存的,可动态伸缩的线程池。 * 可以看出来:核心线程数为0,最大线程数为Integer.MAX_VALUE,如果任务数在某一瞬间暴涨, * 这个线程池很可能会把 服务器撑爆。 * 另外需要注意的是,它们底层都是使用了 ThreadPoolExecutor,只不过帮我们配好了参数 */ public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }} 全文完! 希望本文对大家有所帮助。如果感觉本文有帮助,有劳转发或点一下“在看”!让更多人收获知识! 长按识别下图二维码,关注公众号「Doocs 开源社区」,第一时间跟你们分享好玩、实用的技术文章与业内最新资讯。 本文分享自微信公众号 - Doocs开源社区(idoocs)。如有侵权,请联系 support@oschina.cn 删除。本文参与“OSC源创计划”,欢迎正在阅读的你也加入,一起分享。

优秀的个人博客,低调大师

亚马逊要求员工删除TikTok又撤回,滑?

当地时间上周五,纽约时报报道称,亚马逊以安全风险为由要求员工将TikTok从他们的移动设备上删除。 据报道,亚马逊在上周五早上给员工的第一封电子邮件中表示:“由于安全隐患,不再允许在访问亚马逊电子邮件的移动设备上使用TikTok应用程序。 “如果您的设备上装有TikTok,则必须在7月10日之前将其删除,以保留对Amazon电子邮件的移动访问权限。目前,允许通过您的Amazon笔记本电脑浏览器使用TikTok。” 纽时记者Taylor Lorenz在推特上公布了这封电子邮件的截图。 然而反转来了,据报道,当地时间7月11日,亚马逊的一位发言人表示,电子邮件“发错了”。 亚马逊发言人说:“此前给员工的电子邮件是错误的,亚马逊给现在对TikTok的政策没有改变。” 据报道,近期,由于iOS 14中的一项新功能,TikTok被发现在后台运行时访问用户剪贴板数据,从而可能暴露密码或其他敏感数据,目前尚不清楚该行为已存在于应用程序中多久。 当地时间7月7日,美国副总统彭斯表示,因国家安全和数据隐私问题,美国将继续对TikTok等公司采取“强硬立场”。 而在前一日,美国国务卿迈克·庞培告诉福克斯新闻社 ,特朗普政府正考虑采取行动禁用TikTok等社交媒体应用。他还称,美国民众对于使用TikTok应提高警觉,不过他并没有说明美国具体将采取何种措施。 此后,TikTok否认了美国政府对其威胁国家安全和违反数据隐私的指控。 据报道,尚不完全清楚政府当局如何着手禁止该应用程序,尽管如此,故障和没安全感使TikTok社区陷入混乱状态。 TikTok发言人回应了亚马逊下令员工删除该应用程序的决定,并在声明中说:“虽然亚马逊在发送电子邮件之前未与我们通信,但我们仍然不理解他们的担忧,但我们欢迎两个公司进行对话,以便我们可以解决他们可能遇到的任何问题,并使其团队能够继续参与我们的社区。” TikTok发言人还称:“成千上万的美国人在TikTok里寻求娱乐、灵感和交流,其中包括许多奋斗在抗疫前线的亚马逊员工和承包商。我们对此感到自豪。” 据报道,TikTok是字节跳动旗下的一款短视频软件,在2016年作为国际版抖音正式进入美国市场。 市场调查公司SensorTower的数据显示,截至今年4月底,美国是TikTok在全球的第二大市场。目前,在美国,TikTok的下载量已经超过1.65亿次,约占美国总人口的50%。而美国用户平均每天打开该软件8次,每人每次平均使用的时间约为5分钟。 这么多用户中,青少年是TikTok在美国的主要用户群体。2019年,TikTok曾表示,在美国2650万的活跃用户中,约有60%的用户年龄在16-24岁。 并且,TikTok作为字节跳动独立的子公司并没有在中国运营。 2019年,字节跳动将TikTok和抖音完全分割,除了中国用户无法使用TikTok外,字节跳动在内部的技术团队也设置了防火墙,并且,工作地点在中国、工作内容为中国市场的员工,则无法获得访问TikTok关键数据的权限。 尽管如此,TikTok面临的麻烦不小,在美国已经受到了额外严格的监管,而且最近确实因为一些举动引起了很大争议。 例如,硅星人报道TikTok滥用iOS剪贴板权限,在用户不知情的情况下频繁访问剪切板,因此也成为了认为TikTok在侵犯隐私的证据之一。不过对此,TikTok表示这是为了“反垃圾账号”,并且在事情曝光后停止了这一做法。 不仅如此,美国军方已经禁止士兵安装这款应用,而美国监督机构一直“不信任”TikTok。 自2019年起,美国监督机构加大了对 TikTok的审查力度,审查的内容主要涉及威胁国家安全和违反数据隐私两方面。 例如2019年2月,美国联邦贸易委员会(FTC)起诉TikTok,FTC表示,TikTok未经父母同意非法收集了13岁以下儿童的个人信息。对此,FTC给TikTok开出了570万美元的罚单。 针对美国政府的多次指控,TikTok通过设立透明中心、分离国内外市场、聘请本土高管等行为尝试从多方面树立维护数据隐私的形象。 并且2020年3月,TikTok在美国洛杉矶设立了透明中心,还计划在华盛顿开设第二个透明中心。 不光美国,当印度禁止使用TikTok后,通过该平台工作的印度网红们损失惨重,例如“鸡腿哥”。 据华盛顿邮报当地时间7月11日报道,来自印度的41岁健身房老板卡玛什,此前在TikTok上分享吃鸡腿而闻名,短短几个月内,他就吸引了近全球700万粉丝。 印度是TikTok最大的用户市场。如今,不少印度大V用户郁闷地表示,禁令让他们失去了主要的收入来源,现在,他们必须找到一个新的平台。 本文转自雷锋网,如需转载请至雷锋网官网申请授权。

优秀的个人博客,低调大师

集成支撑平台:智慧城市隐患的“扫雷

21世纪以来的15年是我国城镇化快速发展的15年,快速发展中也出现了一些矛盾和冲突,如资源短缺、交通拥堵、环境污染、安全隐患等,而发展智慧城市是解决这些新问题的一个重要手段。 近年来,我国智慧城市在国家政策的推动下有了较为长足的发展,如智慧交通、智慧医疗、互联网+、大数据、云计算等。现如今,人们只要在家里动动手,上网一查询,就可以实时了解道路及公交信息;只要手机一操作,就可以实现看病在线预约挂号,甚至在线问诊或远程诊疗;即便是政务问题,也可以通过网络在“家里”解决。这样类似的还有很多便民利民的举措均得益于智慧城市的推行。 不过,你可曾知道,就这随手一点的便捷,都是基于对各个领域海量数据信息的收集、分析及处理等环节的支撑。而因智慧城市产生的海量信息在处理分析过程中却存在“信息孤岛”等现象,因此急需一个大的平台来打通各领域的信息通道,为社会提供更加高效、便捷、多样化的服务,扫除城市隐患。正元地理信息有限责任公司智慧城市集成支撑平台即致力于探索解决相关问题。 集成支撑平台或为解决“信息孤岛”一剂良药 智慧城市建设的目标是感知、物联、智能。行业人士对智慧城市的定义认为,在城市发展中,数字城市是智慧城市的萌芽阶段,主要通过对城市各领域数据的采集为未来奠定基础;第二阶段为智能城市,该阶段主要致力于政务智能,使为民服务与监管更加便捷;而智慧城市则是前两个阶段的升级,在大数据收集基础上,打通各部门间通道,使信息数据交互融合,从而更好地服务于民。 例如某人想要申请公司注册,通过智慧政务,一个部门审批完后通过电子方式流转到下一部门继续审批。在智慧城市中,不仅可以进行电子审批,并且通过机构流程再造,还可将同时审批的部门设置为并行,进一步提高审批效率。一些可以网上申请审批的事项,可通过手机申请,也可通过社区门口的政务屏申请。需要提交资料的,可提前在市民大厅预订一个资料柜,通过二维码进行资料投放,免去舟车劳顿排队良久只为递件的麻烦。 当然,这只是一个美好的愿景,虽然一些城市或地区已着手开始尝试,但在一些基础性工作中,还有不少问题需要解决。例如智慧城市发展中出现的“信息孤岛”现象。 正元地理信息有限责任公司研究开发中心基础研发部副经理胡杰在接受中国经济导报记者采访时表示,“智慧城市建设发展迅速,交通、市政、社区等领域都取得了一定成绩,但智慧应用之间业务、数据不共享问题日益突出,在处理问题时经常难以形成闭环。而这些智慧应用中往往存在着一些可以共享共用的数据、软件或硬件设施,这就导致了使用的不便和重复建设的浪费。”城市信息化的发展对城市信息化的网络基础设施建设、信息资源数据库建设和共享、城市管理与运行相关系统功能提升等都提出了新的要求,迫切需要解决如下问题:城市“感知”节点远远不够,无法满足精细化管理需要;城市 各部门业务系统呈“信息孤岛”态势,跨部门协同能力较弱;城市管理海量数据处理和分析能力不足,无法满足城市管理综合监控和智能化决策的需要等。因此,需要通过新的视角、新的思路、新的技术手段和更加全面系统的方法来加以解决和实现。 于是,去年12月24日,国务院发布《关于深入推进城市执法体制改革改进城市管理工作的指导意见》,其中指出,要综合设置机构、整合信息平台。“在这一文件发布之前,正元已于去年7月开始了智慧城市集成支撑平台的研发。现在看来,正元建设智慧城市的思路很好地响应了这一文件的要求。”胡杰自豪地说道,通过智慧城市集成支撑平台,可对数据、业务等进行整合、利用,从而用以支撑智慧城市的建设。 胡杰向中国经济导报记者介绍,智慧应用建设过程中出现的“信息孤岛”现象,无法将信息资源统一起来做相关应用、分析,无法为智慧城市整体的决策进行支持。而集成支撑平台可整合智慧城市各方面信息,进而可提供更为智能的决策,且因为集成的功能,可避免重复建设的浪费。“我们首先会建立大数据中心,将人口基础信息、法人单位基础信息、自然资源和空间地理基础信息、宏观经济信息以及物联网数据、市政、人防、国土、公安等行业数据统一融合纳入。进而,能够根据上层应用的需要来使用这些数据。平台提供了完备的接口,以服务的形式对外共享数据,并将外部数据集成进来,通过抽取、清洗等一系列的步骤保存在数据中心。”胡杰进一步解释道。 此外,以服务形式提供的除了数据还有功能,如地理信息服务、商业智能服务、数据挖掘服务、物联网服务、业务专项服务等应用服务,日志、缓存、配置、身份认证等通用服务都包含在内。 平台可在消除城市隐患等更为广泛的场景中发挥作用 胡杰告诉中国经济导报记者,集成支撑平台的意义在于整合了较为全面的数据,并结合“云”的思想,以平台层的身份对外提供服务支撑。“只有掌握了更全面的数据,才能从全局进行决策分析,比如在处理应急事件时,需要人防、安检、消防等各部门的数据,而目前各部门间信息数据的通道仍不是很畅通。另一方面,整合数据、统一对外提供服务避免了重复建设。” 可以说,对于在地理信息领域浸淫已久的正元而言,早已建成多年的地理信息共享服务平台是智慧城市集成支撑平台的一个重要基础,亦或说现在的支撑平台是其升华。“不但在业务、数据方面不再局限于地理信息行业,更是通过云计算、大数据等一系列新技术的运用满足了性能、功能上的需求。平台层应用的清晰定位决定了它能够在更广泛的场景中得到使用。当然,地理信息尤其是管线方面的优势,在我们这个平台中也得到了继承和发扬。” 据悉,目前已有数家公司建成了智慧城市整体运营类产品。这些产品均能对市政、交通、环保等行业进行一定程度的支撑,但以单独系统存在的形式使其在集成和共享数据、服务,以及整合第三方应用方面有着天生的短板;此外,在被称为“城市生命线”的地下管线管理工作中,因为缺少技术储备和行业积累,很难实现有效的监管。而正元智慧城市集成支撑平台在履行为其他系统提供数据、功能服务使命的同时,很好地支持了公司在数十年来逐渐形成的一套地下管线“全生命周期”管理办法,在满足常规规划设计、建设施工、运行维护、报废处置工作需要的基础上,能够有效预防、及时发现、合理处置地下管线隐患、事故,最大限度避免人身、财产损失,可更好地保障公共安全。 据胡杰透露,借力集团内事业部、各分子公司的鼎力支持,目前如此“包容”的一个平台主体功能已研发完毕,该平台已在智慧城市项目中成功实施。“它集中了原本分散在不同部门的各类数据,为城市安全预警、城市应急指挥、城市领导指挥决策等城市整体管理工作,以及供水科学调度、防汛排涝、绿色照明等专项业务应用提供了强有力的支撑。”胡杰说,“以城市应急来说,在集成支撑平台的帮助下,我们能够及时获取事故周边的危险源、人口密集场所、医疗力量、紧急避难场所、救援人员物资等各类信息,通过路径分析等服务可获取人员撤离、救援的最佳路径,通过大数据手段预判可能造成的次生事故,从而进行预防和监控,避免事态扩大。” 不过,胡杰强调,在实施过程中,集成整合数据其实是一件十分困难的工作。由于历史原因,各部门间存在着数据壁垒,不仅各自的数据储存于不同场所,这些数据的种类、结构、管理方法也不尽相同。因此,建立标准规范对于数据整合的意义非同小可。在此之后,只有经过数据抽取、清洗和转换,这一系列多样繁杂的数据才能为我所用。 数据的挖掘及提升运算性能是持续性任务 在智慧城市中,大数据的角色愈发重要。如何有效挖掘、分析大数据,是准确高效解决城市运营问题的关键。“例如城市安全预警需建立模型,设计人员只能先通过经验或推理来假设某一城市领域在何种条件下将会产生哪种问题,但实际上还需要挖掘大量数据来分析验证推理条件的合理性并提高其准确性。但由于受数据积累时间不够长,且一些数据不易协调等客观因素影响,暂时无法在短期内做到十分完善。”胡杰坦言。 下一步,正元将继续深化大数据挖掘,充分利用云计算的能力提升性能。“平台虽然已经建立,但部分功能在技术方面仍有改进的空间。”胡杰解释,“如管线安全评价服务,目前仍通过传统的方式实现,如需对较大区域进行分析耗时会比较长,但运用大数据技术,则能够大大缩短时间。”有时城市数据分析需要做到秒级反应,如遇到应急情况,需在最短时间内迅速计算并反馈,此时传统运算方式的速度远不能及,这就需要大数据、云计算来发挥作用。 值得一提的是,数据已成为国家基础性战略资源。今年9月,《促进大数据发展行动纲要》明确指出,在未来5~10年内,要打造精准治理、多方协作的社会治理新模式,要加快政府数据开放、共享,要推进数据资源的整合及治理能力的提升。胡杰希望,利国利民的政策一定要快速落地实现。 本文转自d1net(转载)

资源下载

更多资源
腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题,腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构,目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring

Spring框架(Spring Framework)是由Rod Johnson于2002年提出的开源Java企业级应用框架,旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念,提供核心容器、应用上下文、数据访问集成等模块,支持整合Hibernate、Struts等第三方框架,其适用范围不仅限于服务器端开发,绝大多数Java应用均可从中受益。

Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能,例如代码缩略图,Python的插件,代码段等。还可自定义键绑定,菜单和工具栏。Sublime Text 的主要功能包括:拼写检查,书签,完整的 Python API , Goto 功能,即时项目切换,多选择,多窗口等等。Sublime Text 是一个跨平台的编辑器,同时支持Windows、Linux、Mac OS X等操作系统。

WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源,继承了IntelliJ IDEA强大的JS部分的功能。

用户登录
用户注册