tydhot-CSDN博客

原创 Golang sync.pool对象池

概览Goalng中通过sync.pool提供了对象池的实现来达到对象复用的目的。在netty中，也通过Recycle类实现了类似的对象池实现。在netty的对象池Recycle中，当A线程需要将B线程申请的对象回收到对象池中的时候，会专门开辟一个专门由A线程回收到B线程的队列，以避免回收对象的时候所发生的资源竞争。类似的，在golang的对象池sync.pool中也是通过类似的思想来实现所要达到的目的。sync.pool的结构type Pool struct { noCopy noCopy //

2021-06-29 16:04:54 650 2

原创 golang sync.map

在golang中，线程安全的map实现为sync.Map，相较于java中线程安全的map ConcurrentHashMap，在设计与实现上都有巨大的差别。java中的ConcurrentHashMapjava中的ConcurrentHashMap为了实现线程安全，在1.7当中，通过分段锁的实现达到了这一目的，区别于HashTable的全部阻塞操作，分段锁的设计在一定程度上提升了在并发场景下的访问性能。在1.8的过程中，锁的粒度被进一步降低，被缩小到了一个HashEntry首节点的地步，并通过在一定长

2021-06-29 16:04:22 604

原创 Golang heap源码简单走读

golang heap小根堆源码走读heap概览在golang中，通过heap给出了一个实现小根堆的接口。type Interface interface { sort.Interface Push(x interface{}) Pop() interface{} }由于小根堆中，需要根据容器中的元素大小来进行比较以确定元素在堆中的位置。因此也需要实现sort的接口。type Interface interface { Len() int Less(i, j int) bool

2021-06-10 22:48:35 429 1

原创 Golang list双向链表源码走读

list概览list为golang中的双向链表实现，存入list中的元素都会被封装成list中的节点放到双向链表中进行存储。简单的list使用代码如下：func main() { list := list.New() fmt.Println("list length is " + strconv.Itoa(list.Len())) list.PushBack(1) list.PushBack(2) list.PushBack(3) fmt.Println("list length is

2021-06-08 00:34:12 227 1

原创 Sentinel 中限流算法的实现

本文的github地址点击这里Sentinel 中漏桶算法的实现Sentinel 中漏桶算法通过 RateLimiterController 来实现，在漏桶算法中，会记录上一个请求的到达时间，如果新到达的请求与上一次到达的请求之间的时间差小于限流配置所规定的最小时间，新到达的请求将会排队等待规定的最小间隔到达，或是直接失败。@Overridepublic boolean canPass(Node node, int acquireCount, boolean prioritized) { i

2021-01-24 21:54:53 1239

原创 Sentinel底层LongAdder的计数实现

本文的github地址LongAdder 的原理在 LongAdder 中，底层通过多个数值进行累加来得到最后的结果。当多个线程对同一个 LongAdder 进行更新的时候，将会对这一些列的集合进行动态更新，以避免多线程之间的资源竞争。当需要得到 LongAdder 的具体的值的时候，将会将一系列的值进行求和作为最后的结果。在高并发的竞争下进行类似指标数据的收集的时候，LongAdder 通常会和 AtomicLong 进行比较，在低竞争的场景下，两者有着相似的性能表现。而当在高并发竞争的场景下，Lo

2020-12-31 16:43:36 329

原创 sentinel 时间窗口的实现

本文的github地址点击这里获取时间窗口的主要流程在 Sentinel 中，主要是通过 LeapArray 类来实现滑动时间窗口的实现和选择。在 sentinel 的这个获取时间窗口并为时间窗口添加指标的过程中，主要的流程为：根据当前时间选择当前时间应该定位当前时间应该属于的时间窗口 id。根据时间窗口 id 获取时间窗口。这里可能会存在三种情况：时间窗口还未建立，那么将会为此次流量的进入建立一个新的时间窗口返回，并且接下来这个时间窗口内的获取请求都将返回该窗口。时间窗口已经建立的情况下

2020-12-10 10:07:20 662

原创 Netty技术细节源码分析-内存池之PoolChunk设计与实现

该文所涉及的netty源码版本为4.1.16。在一开始需要明确的几个概念在Netty的内存池的PoolChunk中，先要明确以下几个概念。page: page是chunk中所能申请到的最小内存单位。chunk: 一个chunk是一组page的集合在PoolChunk中，chunkSize的大小是2maxOrder*pageSize，其中2maxOrder是PoolChunk中的完全二叉树叶子结点的数量，pageSize则是单个page的大小。综合如上所述，举一个数字上的例子，默认情况下，单个

2020-10-18 14:15:05 287 2

原创 Netty技术细节源码分析-HashedWheelTimer时间轮原理分析

本文是该篇的修正版本文的github地址：点此该文所涉及的netty源码版本为4.1.6。Netty时间轮HashedWheelTimer是什么Netty的时间轮HashedWheelTimer给出了一个粗略的定时器实现，之所以称之为粗略的实现是因为该时间轮并没有严格的准时执行定时任务，而是在每隔一个时间间隔之后的时间节点执行，并执行当前时间节点之前到期的定时任务。当然具体的定时任务的时间执行精度可以通过调节HashedWheelTimer构造方法的时间间隔的大小来进行调节，在大多数网络应用的情况下

2020-10-08 23:19:59 459

原创 Netty技术细节源码分析-ByteBuf的内存泄漏原因与检测

该文所涉及的netty源码版本为4.1.6。Netty中的ByteBuf为什么会发生内存泄漏在Netty中，ByetBuf并不是只采用可达性分析来对ByteBuf底层的byte[]数组来进行垃圾回收，而同时采用引用计数法来进行回收，来保证堆外内存的准确时机的释放。在每个ByteBuf中都维护着一个refCnt用来对ByteBuf的被引用数进行记录，当ByteBuf的retain()方法被调用时，将会增加refCnt的计数，而其release()方法被调用时将会减少其被引用数计数。private bo

2020-10-08 23:16:13 905

原创 Netty技术细节源码分析-MpscLinkedQueue队列原理分析

本文的github地址：点此该文所涉及的netty源码版本为4.1.6。MpscLinkedQueue是什么在Netty的核心中的核心成员NioEventLoop中，其中任务队列的实现taskQueue便是MpscLinkedQueue。MpscLinkedQueue是Netty所实现的一个基于多生产者单消费者的无锁队列，针对NioEventLoop中任务队列的特点，其单消费者的场景在一开始就避免了从队列中取数据时加锁的必要，而其最精妙的地方便是在多生产者并发从队列中添加数据的时候也没有加锁，达到Ne

2020-09-13 22:26:53 385

原创 Netty技术细节源码分析-Recycler对象池原理分析

本文是该篇的修正版本文的github地址：点此该文所涉及的netty源码版本为4.1.6。Netty的对象池Recycler是什么Recycler是Netty中基于ThreadLocal的轻量化的对象池实现。既然是基于ThreadLocal，那么就可以将其理解为当前线程在通过对象池Recycler得到一个对象之后，在回收对象的时候，不需要将其销毁，而是放回到该线程的对象池中即可，在该线程下一次用到该对象的时候，不需要重新申请空间创建，而是直接重新从对象池中获取。Recycler在netty中被如何

2020-09-13 22:10:58 295 2

原创 Netty技术细节源码分析-FastThreadLocal源码分析

本文是该篇的修正版本文的github地址：点此Netty 的 FastThreadLocal 源码解析该文中涉及到的 Netty 源码版本为 4.1.6。Netty 的 FastThreadLocal 是什么A special variant of ThreadLocal that yields higher access performance when accessed from a FastThreadLocalThread.Internally, a FastThreadLocal u

2020-08-31 21:15:31 159

原创 netty 水位线与oom

Netty版本4.1.6。当channel被调用到writAndFlush()的时候，如同字面意思，实现了两次操作，write和flush，其中write的时候并没有将消息直接写入到socket中，而是封装为ChannelOutboundBuffer中的等待发送消息链表中的一个节点，只有等到flush操作发生的时候才会将链表中的消息全都写入到socket中。这样做的目的可以通过缓存消息的方式，减少flush到的套接字缓冲区的次数。因此，当write写入消息过快，而没来得及进行flush的时候将会导致链表

2020-08-24 02:18:04 837

原创 netty时间轮HashedWheelTimer文档翻译及简单说明

Netty版本4.1.6。Netty的时间轮HashedWheelTimer给出了一个粗略的定时器实现，之所以称之为粗略的实现是因为该时间轮并没有严格的准时执行定时任务，而是在每隔一个时间间隔之后的时间节点执行，并执行当前时间节点之前到期的定时任务。当然具体的定时任务的时间执行精度可以通过调节HashedWheelTimer构造方法的时间间隔的大小来进行调节，在大多数网络应用的情况下，由于IO延迟的存在，并不会严格要求具体的时间执行精度，所以默认的100ms时间间隔可以满足大多数的情况，不需要.

2020-08-14 23:05:28 625

原创关于netty的FastThreadLocal的思考

Netty版本4.1.6。与jdk的ThreadLocal相比，netty的FastThreadLoacl具有更高的读写性能，如何针对原生的ThreadLocal进行优化。准备采用netty的FastThreadLoacl的线程都需要继承自netty实现的FastThreadLocalThread，在FastThreadLocalThread中，直接给出了如下字段。private InternalThreadLocalMap threadLocalMap;在这个前提下，当需要用到线程的Fa

2020-08-08 00:11:13 391

原创 netty ByteBuf对象池和内存泄漏检测实现走读

ByteBuf存放在堆外内存中，采用引用计数法的方式进行内存回收，具体的实现在AbstractReferenceCountByteBuf中。private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> refCntUpdater;static { AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> up

2020-07-29 02:15:00 300

原创算法题存档20200627（树）

给你一棵以 root 为根的二叉树和一个 head 为第一个节点的链表。如果在二叉树中，存在一条一直向下的路径，且每个点的数值恰好一一对应以 head 为首的链表中每个节点的值，那么请你返回 True ，否则返回 False 。一直向下的路径的意思是：从树中某个节点开始，一直连续向下的路径。class Solution { public boolean isSubPath(ListNode head, TreeNode root) { if(head == null)

2020-06-27 12:05:57 203

原创 caffeine 读操作源码走读为什么读这么快

Caffeine通过get()方法获取缓存中的数据。Node<K, V> node = data.get(nodeFactory.newLookupKey(key));if (node == null) { if (recordStats) { statsCounter().recordMisses(1); } return null;}long now = expirationTicker().read();if (hasExpired(node, now))

2020-05-30 16:59:21 1020 1

原创 caffeine 弱引用key的实现

Caffeine底层的存储容器还是ConcurrentHashMap，如果需要存储的key是一个弱引用，如果直接把key存储到ConcurrentHashMap中时，将无可避免的导致一个强引用出现，导致弱引用key的目的失败。当在构造caffeine的时候如果选择了弱引用的key，最后保存到ConcurrentHashMap中的key将不再是put()时候的key，而是一个封装的弱引用对象WeakReference，保证map中对key的强引用实际指向的是封装的对象引用，缓存中真正指向key的还是一个

2020-05-29 00:05:00 1413 1

原创 caffeine 线程私有的ReadBuffer实现

Caffeine中，读操作后的afterRead操作都会异步操作，不会阻塞到正常的读取操作。在高并发读取的前提下，为每个读取操作的线程建立了专属的buffer来存放afterRead事件由消费者统一处理afterRead事件，避免高并发读取下对于事件写入buffer的时候的资源竞争。transient volatile Buffer<E>[] table;StrippedBuffer中维护了一个Buffer[]数组table，用来存放各个线程专属的buffer。如何界定某.

2020-05-25 00:07:03 575

原创 caffeine 时间轮的实现

Caffeine内部实现了一个时间轮。时间轮的具体实现为一个二维数组，其数组的具体位置存放的则为一个待执行节点的链表。时间轮的二维数组的第一个维度则是具体的时间间隔，分别是秒，分钟，小时，天，4天，但但并没有严格按照时间单位来区分单位，而是根据以上单位最接近的2的整数次幂作为时间间隔，因此在其第一个维度的时间间隔分别是1.07s，1.14m，1.22h，1.63d，6.5d。static final long[] SPANS = { ceilingPowerOfTwo(TimeUnit

2020-05-18 01:10:54 1321 1

原创 caffeine 淘汰策略

Caffeine在开启了淘汰策略的时候，维护了三条队列，分别是eden，protected，probation队列。具体的三条队列大小分配方案如下所示，long max = Math.min(maximum, MAXIMUM_CAPACITY);long eden = max - (long) (max * PERCENT_MAIN);long mainProtected = (long) ((max - eden) * PERCENT_MAIN_PROTECTED);其中，eden为总大小的

2020-05-10 23:08:13 4277

原创算法题存档20200505

给出两个非空的链表用来表示两个非负的整数。其中，它们各自的位数是按照逆序的方式存储的，并且它们的每个节点只能存储一位数字。如果，我们将这两个数相加起来，则会返回一个新的链表来表示它们的和。您可以假设除了数字 0 之外，这两个数都不会以 0 开头。示例：输入：(2 -> 4 -> 3) + (5 -> 6 -> 4)输出：7 -> 0 -&...

2020-05-06 00:17:04 149

原创 caffeine 4hash lfu频度记录

在caffeine中的lfu通过一个long array来记录各个数据的被访问次数，每个被访问到的元素都将被通过4次不同的hash seed来计算位置，分别记录四次，以便达到最小的hash冲突可能性。因此，long array的每个槽位中的64位空间，出于频度记录需要与空间的平衡，64位被分为4个16位的空间，而16位的空间又将都将被分为4个4位空间区域分别记录4个hash seed的结果。...

2020-05-05 00:58:47 383

原创 threadlocal的set()方法中的内存回收

ThreadLocal在执行set()方法的时候，实际执行set()逻辑的是其内部类ThreadLocalMap。private void set(ThreadLocal<?> key, Object value) { Entry[] tab = table; int len = tab.length; int i = key.threadLocalHa...

2020-05-02 01:25:42 1570

原创算法题存档2020425

给定一个没有重复数字的序列，返回其所有可能的全排列。示例:输入: [1,2,3]输出:[ [1,2,3], [1,3,2], [2,1,3], [2,3,1], [3,1,2], [3,2,1]]class Solution { public List<List<Integer>> permute(int[] nums...

2020-04-25 02:21:07 169

原创 flink CompactingHashTable源码解析

CompactingHashTable是使用flink管理内存的hash表。这个table被设计分为两个部分，一部分是hash索引，用来定位数据的具体位置，而另一部分则是被分区的内存buffer用来存储具体的内存数据。private final ArrayList<MemorySegment> availableMemory;availableMemory为当前fl...

2020-03-15 11:42:40 230

原创 Arrays.asList()的坑

Arrays.asList()产生的ArrayList并不是集合包下的的ArrayList，而是实现在Arrays类中的内部类ArrayList，该类并未重写remove()方法，因此Arrays.asList()产生的ArrayList一旦调用remove()方法将会报错。如果要调用remove()，可以如下new ArrayList<>(Arrays.asList(...

2020-03-08 01:30:16 259

原创 spark两种kafka偏移量维护方式

1.spark可以通过checkpoint的方式来维护kafka的偏移量，配置简单，只需要配置checkpoint的路径就可以完成偏移量的维护，如果本身spark业务就采用了state状态，那么既不需要额外配置即可确保偏移量的维护。原理：spark会将kafka spark straming处理的topic以及对应消费偏移量持久化到文件当中，当spark任务崩溃后，保存在持久化文件的偏移量将会...

2020-02-27 01:23:15 1417

原创 spark 如何用netty实现akka的actor模型

Spark的中，通过netty实现了类似akka的actor机制。在spark中，一个EndPointData就类似一个akka中的actor。private class EndpointData( val name: String, val endpoint: RpcEndpoint, val ref: NettyRpcEndpointRef) { v...

2020-01-30 11:31:40 2410 1

原创 spark 上游rdd的缓存

Rdd的缓存有两种意义上的缓存。当在SparkContext中常创建输入流的时候，将会注册一个InputDStream流到DStreamGraph当中。当对该流进行transform操作，比如map，flatmap等操作的时候，将会以一开始的InputDStream生成MappedDStream和FlatMappedDStream。在所有stream的超类DStream实现了ma...

2020-01-29 09:24:44 300

原创 spark BlockManager如何实现Broadcast广播

当通过SparkContext调用broadcast()方法的时候，将会直接尝试调用BroadcastFactory的newBroadcast()方法，BroadcastFactory的默认实现是TorrentBroadcastFactory，在其newBroadcast()方法中，实际上就是new了一个TorrentBroadcast。在TorrentBroadcast的初试化流程中，将会通...

2020-01-28 10:46:09 371

原创 spark BlockManager如何实现shuffle

当上游的算子完成了shuffle操作，下游的rdd如何获取shuffle所需要的数据。首先在上游进行写入操作时，写入到的是executor本地的BlockManager。val blockId = ShuffleBlockId(dep.shuffleId, mapId, IndexShuffleBlockResolver.NOOP_REDUCE_ID)val partitionLeng...

2020-01-27 10:40:35 1768 1

原创 spark TaskMemoryManager shuffle 内存管理

在spark中，在shuffle过程中内存放的数据将会通过TaskMemoryManager来进行管理，以ShuffleExternalSorter为例子。final long recordPointer = sortedRecords.packedRecordPointer.getRecordPointer();final Object recordPage = taskMemoryMa...

2020-01-26 09:34:07 997

原创 spark 上下游shuffle结果的存放获取

当一个job在DAGScheduler中被分隔为stage，将会根据其内部的shuffle关系将整个job整理出ShuffleMapStage，而最后结果的ResultStage在提交时，将会不断遍历其parent stage，而本身被加入DAGScheduler的等待集合，只在所有parent的stage执行完毕之后才会执行任务流程中的child stage。private def sub...

2020-01-25 09:50:59 3559

原创 spark executor task执行

Executor执行任务的起点是Executor的launchTask()方法。val executorData = executorDataMap(task.executorId)executorData.freeCores -= scheduler.CPUS_PER_TASKlogDebug(s"Launching task ${task.taskId} on executor i...

2020-01-24 01:56:51 5041

原创 spark KafkaRDD的理解

Spark版本 2.4.0先从0-8版本的kafka说起。当jobGenerator根据时间准备生成相应的job的时候，会依次在graph中调用各个输入流的getOrCompute()方法来获取得到rdd，在这里DirectKafkaInputDStream的compute()方法将会被调用，在这里将会在driver端生成一个时间批次的rdd，也就是KafkaRDD。KafkaR...

2020-01-19 00:41:47 1648

原创 spark ui 上schedulingDelay理解

Spark UI页面上的schedulingDelay具体为一批job submit到第一个job开始执行的这段时间。其中submitTime在JobGenerate中根据定时生成job的事件而触发的generateJobs()方法中，当job被封装成JobSet的时候准备提交给JobScheduler执行的时候，在JobSet的构造方法被记录。Try { jobSchedul...

2020-01-06 23:30:58 2097

原创 spark eventLoop模型

Sprak中，线程之前的交互采用eventLoop模型。当JobGenerate中的clock达到触发新一次job生成的时间后，并不会直接驱动graph去生成job，而是通过往eventLoop中发送一个JobGenerate事件以触发job生成事件的产生。private val eventQueue: BlockingQueue[E] = new LinkedBlockingDeque...

2020-01-06 01:03:59 524

空空如也

空空如也