首页 文章 精选 留言 我的

精选列表

搜索[源码分析]，共10010篇文章
优秀的个人博客,低调大师

ConcurrentHashMap 源码分析

和 HashMap 不同的是，ConcurrentHashMap 采用分段加锁的方式保障线程安全，JDK 1.8 之后，ConcurrentHashMap 的底层数据结构从 1.8 开始跟 HashMap 差不多。 HashTable 也是线程安全的，存储 Key-Value 键值对的数据结构，Key 和 Value 都不能为空，但不推荐使用，因为其所有的方法采用 synchronized 修饰，效率低。 Key 和 Value 都不能为 Null 的原因是：如果 map.get(key) 返回 null，可以认为是 value 的值本来就是 null，也可以认为 map 中不存在 key 的存储数据，因此具有二义性，但 HashMap 在单线程环境，可以通过 map.containsKey(key) 判断，消除而已性。 但在多线程环境中，map.get(key) 和 map.containsKey(key) 是非原子的操作，可能在线程 A 的两个语句运行之间，其他线程 B 运行 map.put(key,value)，导致线程 A 无法消除上面的二义性。 参考 https://www.cnblogs.com/thisiswhy/p/12059240.html 下图是 ConcurrentHashMap 的 UML 关系图。 1、底层存储结构 1.1、JDK 1.7 的存储结构，了解即可 在 JDK 1.7 ，ConcurrentHashMap 通过对 Segment 的分段加锁实现线程安全。一个 Segment 里面就是 HashMap 的存储结构，可以扩容。Segment 的数据量初始化以后不可以更改，默认值 16，因此默认支持 16 个线程同时操作 ConcurrentHashMap。 1.2 JDK 1.8 的存储结构 JDK 1.8 之后，存储结构变化比较大，跟 HashMap 类似。红黑树节点小于某个数(默认值 6) 又会转换为链表。 [ ] ConcurrentHashMap的主要成员变量，类似 HashMap，补上注释 2、ConcurrentHashMap 的构造方法 ConcurrentHashMap 的默认构造容量为 16，在初始化的时候并不会初始化 table 数组。同 HashMap 一样，在 put 第一个元素的时候才会 initTable() 初始化数组。 /** Creates a new, empty map with the default initial table size (16). */ public ConcurrentHashMap() { } // 设置初始化大小的构造函数 public ConcurrentHashMap(int initialCapacity) { this(initialCapacity, LOAD_FACTOR, 1); } // 根据传入的 map 初始化 public ConcurrentHashMap(Map<? extends K, ? extends V> m) { this.sizeCtl = DEFAULT_CAPACITY; putAll(m); } // 设置初始容量和加载因子的大小 public ConcurrentHashMap(int initialCapacity, float loadFactor) { this(initialCapacity, loadFactor, 1); } // 初始容量、加载因子、并发级别 public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { // 数据校验 if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException(); // 如果初始容量小于并发级别 if (initialCapacity < concurrencyLevel) // Use at least as many bins initialCapacity = concurrencyLevel; // as estimated threads // 一些比较 long size = (long)(1.0 + (long)initialCapacity / loadFactor); int cap = (size >= (long)MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : tableSizeFor((int)size); this.sizeCtl = cap; } 3、get、put 方法 3.1 get 方法，根据 key 找 value，没有返回 null get 的流程总体和 HashMap 差不多，只不过是通过头结点的 hash 值判断是红黑树还是链表。 static final int MOVED = -1; // 转发节点？ TODO 作用？ static final int TREEBIN = -2; // 跟节点 static final int RESERVED = -3; // 临时保留的节点？ TODO 作用？ static final int HASH_BITS = 0x7fffffff; // hash 的扰动函数 spread() 计算用的 // 根据 key 获取 value 值 public V get(Object key) { Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek; // 计算 hash 值 int h = spread(key.hashCode()); // 集散所在的 hash 桶 if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { if ((eh = e.hash) == h) { // 头结点，刚好是要找的节点 if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } else if (eh < 0) // 头结点 hash 值小于 0，说明正在扩容或者是红黑树，find 查找 return (p = e.find(h, key)) != null ? p.val : null; while ((e = e.next) != null) { // 链表遍历查找 if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; } 3.2、put 方法 put 方法的流程跟 HashMap 的流程差不多，不同点在于线程安全，自旋，CAS，synchronized onlyIfAbsent 如果为 true ，如果已经存在了 key，不会替换旧的值。 public V put(K key, V value) { return putVal(key, value, false); } /** Implementation for put and putIfAbsent */ final V putVal(K key, V value, boolean onlyIfAbsent) { // key 和 value 都不能为 null if (key == null || value == null) throw new NullPointerException(); // 计算 hash(key) 的扰动函数 int hash = spread(key.hashCode()); // 离岸边的长度 int binCount = 0; for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; K fk; V fv; // 如果 table 还没有初始化，就初始化 table （自旋+CAS) if (tab == null || (n = tab.length) == 0) tab = initTable(); else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 如果当前 hash 桶为 null，直接放入，CAS 加入，成功了就直接 break if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value))) break; // no lock when adding to empty bin } // TODO : else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else if (onlyIfAbsent // check first node without acquiring lock && fh == hash && ((fk = f.key) == key || (fk != null && key.equals(fk))) && (fv = f.val) != null) return fv; else { // 旧的值 V oldVal = null; // 加锁 synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { binCount = 1; for (Node<K,V> e = f;; ++binCount) { K ek; // 如果存在 hash(key) 和 key 对应的节点，直接更改 value 值 if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; } Node<K,V> pred = e; if ((e = e.next) == null) { // 不存在直接放入，因为前面加锁了 pred.next = new Node<K,V>(hash, key, value); break; } } } // 如果是红黑树，红黑树插入 else if (f instanceof TreeBin) { Node<K,V> p; binCount = 2; if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } else if (f instanceof ReservationNode) throw new IllegalStateException("Recursive update"); } } if (binCount != 0) { // 是否要转为红黑树 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); // 旧的值 if (oldVal != null) return oldVal; break; } } } addCount(1L, binCount); return null; } 4、TODO ConcurrentHashMap 的扩容方法 ConcurrentHashMap 也是默认扩容 2 倍，扩容的方法 transfer() Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; 5、总结 ConcurrentHashMap 在 JDK 1.7 和 1.8 变化很大，在 JDK 1.7 中，采用 Segment 分段存储数据，也通过 Segment 分段加锁。 而在 JDK 1.8 中，使用 synchronized 锁定 hash 桶的链表的首节点/红黑树的根节点，只要 hash(key) 不冲突，就不会影响其他线程。
2020-08-11
优秀的个人博客,低调大师

ArrayList源码分析

一、核心变量 // 序列化ID private static final long serialVersionUID = 8683452581122892189L; // 默认初始化容量 private static final int DEFAULT_CAPACITY = 10; // 空数组 private static final Object[] EMPTY_ELEMENTDATA = {}; // 空数组 private static final Object[] DEFAULTCAPACITY_EMPTY_ELEMENTDATA = {}; // 存储数据元素的数组 transient Object[] elementData; // non-private to simplify nested class access // 当前arraylist集合的大小，也就是elementData数组中数据元素的个数 private int size; 二、构造函数 /** * 一：无参构造方法 */ public ArrayList() { this.elementData = DEFAULTCAPACITY_EMPTY_ELEMENTDATA; } /** * 二：携带一个int类型的参数，指定arraylist的初始容量 */ public ArrayList(int initialCapacity) { if (initialCapacity > 0) { this.elementData = new Object[initialCapacity]; } else if (initialCapacity == 0) { this.elementData = EMPTY_ELEMENTDATA; } else { throw new IllegalArgumentException("Illegal Capacity: "+ initialCapacity); } } 1、无参构造 可以看到，在构造方法中直接将 elementData 指向 DEFAULTCAPACITY_EMPTY_ELEMENTDATA空数组，这个时候该ArrayList的size为初始值0。 1、有参构造 进行参数校验： 参数大于0，则指定数组长度； 参数等于0，则为空数组； 参数小于0，则抛异常。 三、add方法 /** * 一：直接添加数据元素到arraylist的尾部 */ public boolean add(E e) { //是否扩容、记录modCount ensureCapacityInternal(size + 1); // Increments modCount!! //把值添加到数组尾部 elementData[size++] = e; return true; } ---------------------------------------------------------------------- private void ensureCapacityInternal(int minCapacity) { //minCapacity=size+1； //minCapacity表示如果添加成功后，数组的最小长度 //如果为无参构造 if (elementData == DEFAULTCAPACITY_EMPTY_ELEMENTDATA) { //取默认长度和minCapacity的最大值，即10 minCapacity = Math.max(DEFAULT_CAPACITY, minCapacity); } //是否扩容 ensureExplicitCapacity(minCapacity); } ---------------------------------------------------------------------- private void ensureExplicitCapacity(int minCapacity) { modCount++; // 如果添加后最小长度大于 数组长度 if (minCapacity - elementData.length > 0) //扩容 grow(minCapacity); } ---------------------------------------------------------------------- private void grow(int minCapacity) { //获取数组长度 int oldCapacity = elementData.length; //1.5倍扩容 int newCapacity = oldCapacity + (oldCapacity >> 1); //1.5倍扩容也不够用 if (newCapacity - minCapacity < 0) //扩容后长度=minCapacity newCapacity = minCapacity; //简直最大长度 if (newCapacity - MAX_ARRAY_SIZE > 0) newCapacity = hugeCapacity(minCapacity); // 复制 elementData = Arrays.copyOf(elementData, newCapacity); } 代码中已经有注释了，很清晰。 四、remove方法 /* * 一：根据角标进行remove操作 */ public E remove(int index) { // 1. 对角标越界进行判断 if (index >= size) throw new IndexOutOfBoundsException(outOfBoundsMsg(index)); // 2.modCount自增1 modCount++; // 3.获取到指定下角标位置的数据 E oldValue = (E) elementData[index]; // 4.计算需要移动的元素个数 int numMoved = size - index - 1; if (numMoved > 0) // 5. 指定角标位置后的元素前移一位，效率低 System.arraycopy(elementData, index+1, elementData, index, numMoved); // 6.将size自减1，并将数组末尾置为null，便于垃圾回收 elementData[--size] = null; // clear to let GC do its work // 7.最后将所要删除的数据元素return掉 return oldValue; } /* * 二：根据数据元素进行remove操作 */ public boolean remove(Object o) { if (o == null) { for (int index = 0; index < size; index++) if (elementData[index] == null) { fastRemove(index); return true; } } else { for (int index = 0; index < size; index++) if (o.equals(elementData[index])) { fastRemove(index); return true; } } return false; } ---------------------------------------------------------------------- private void fastRemove(int index) { // 1.modCount的值自增1 modCount++; // 2.计算需要移动的元素个数 int numMoved = size - index - 1; if (numMoved > 0) // 3. 指定角标位置后的元素前移一位 System.arraycopy(elementData, index+1, elementData, index, numMoved); // 4.将size自减1，并将数组末尾置为null，便于垃圾回收 elementData[--size] = null; // clear to let GC do its work } 五、set方法 public void set(E e) { if (lastRet < 0) throw new IllegalStateException(); //检查modCount checkForComodification(); try { ArrayList.this.set(lastRet, e); } catch (IndexOutOfBoundsException ex) { throw new ConcurrentModificationException(); } } ---------------------------------------------------------------------- public E set(int index, E element) { rangeCheck(index); E oldValue = elementData(index); elementData[index] = element; return oldValue; } ---------------------------------------------------------------------- final void checkForComodification() { if (expectedModCount != ArrayList.this.modCount) throw new ConcurrentModificationException(); } 六、get方法 public E get(int index) { rangeCheck(index); checkForComodification(); return ArrayList.this.elementData(offset + index); } ---------------------------------------------------------------------- E elementData(int index) { return (E) elementData[index]; } 七、clear方法 public void clear() { modCount++; // clear to let GC do its work for (int i = 0; i < size; i++) elementData[i] = null; size = 0; } 八、contains方法 public boolean contains(Object o) { return indexOf(o) >= 0; } ---------------------------------------------------------------------- public int indexOf(Object o) { if (o == null) { for (int i = 0; i < size; i++) if (elementData[i]==null) return i; } else { for (int i = 0; i < size; i++) if (o.equals(elementData[i])) return i; } return -1; } 未命名文件.jpg 九、fail-fast机制 ail-fast机制是集合中的一种错误检测机制，我们在操作集合中经常会遇到 java.util.ConcurrentModificationException异常，产生该异常的原因就是fail-fast机制。 实现：如果在迭代期间计数器被修改，那么hasNext或next将抛出concurrentModificationException 缺点：这种检查是没有同步的情况下进行的，因此可能会看到失效的计数值，而迭代器可能并没有意识到已经发生了修改。这是一种设计上的权衡，从而降低了并发修改操作的检测代码对程序性能带来的影响。 十、可能的并发问题 add() EX：a、100个元素，可能最后数组长度不到100。 两个线程并发add，对索引位置5的地方几乎同时赋值，第二个线程会覆盖第一个线程的值，并且size少了1个。 b、假设有两个线程在操作同一个ArrayList，线程一执行step1(容量足够)后被挂起，线程二执行add()方法后，线程一被唤醒，这时线程一因为已经不再判断容量是否足够（已经判断过），执行step2就会出现数组越界 数组容量检测的并发问题 remove 两个线程有可能会想要删除同一个内容，一个线程先完成的时候第二个线程再删，就找不到这个内容了
2019-04-23
优秀的个人博客,低调大师

openstack 源码分析

Nova对于底层Hypervisor（如KVM/QEMU等）的调用与管理主要通过LibvirtDriver类，nova/virt/libvirt/driver.py Libvirt对Hypervisor的连接有两种方式：一种是只读式，用于管理；另一种是认证式，用于操作； 创建实例过程：/nova/api/ec2/cloud.py/L1193, run_instances()，获取创建数目，kernal_id,ramdisk_id，镜像uuid等通过self.compute_api.create（） 发送实例的信息和要运行实例的请求消息到远程调度器scheduler. 从用户发送请求（创建新实例）的工作流程： 上图是一个全局的流程图，图中每个服务是一个单独的进程实例，他们之间通过rpc调用（广播或者调用）另一个服务。 nova-api服务是一个wsgi服务实例，创建新instance的入口代码是在nova /api/openstack/compute/servers.py，处理函数为create(self, req, body); 参数验证之后，调用compute api的create 函数（代码在nova/compute/api.py中）： http://www.infoq.com/cn/articles/openstack-access-request-calling-process eclipse快捷键： 注释：ctrl+/ 跳转函数定义：F3 显示outline:ctrl+o 自己用c++实现c++反射机制；类比java反射机制
2015-10-22
优秀的个人博客,低调大师

LevelDB源码分析-Compact

Compaction compact由背景线程完成，代码中触发背景线程的函数为： void DBImpl::MaybeScheduleCompaction() { mutex_.AssertHeld(); if (background_compaction_scheduled_) { // Already scheduled } else if (shutting_down_.Acquire_Load()) { // DB is being deleted; no more background compactions } else if (!bg_error_.ok()) { // Already got an error; no more changes } else if (imm_ == nullptr && manual_compaction_ == nullptr && !versions_->NeedsCompaction()) { // No work to be done } else { background_compaction_scheduled_ = true; env_->Schedule(&DBImpl::BGWork, this); } } 背景线程调用的函数为： void DBImpl::BGWork(void *db) { reinterpret_cast<DBImpl *>(db)->BackgroundCall(); } 这个函数是调用BackgroundCall函数实现的： void DBImpl::BackgroundCall() { MutexLock l(&mutex_); assert(background_compaction_scheduled_); if (shutting_down_.Acquire_Load()) { // No more background work when shutting down. } else if (!bg_error_.ok()) { // No more background work after a background error. } else { BackgroundCompaction(); } background_compaction_scheduled_ = false; // Previous compaction may have produced too many files in a level, // so reschedule another compaction if needed. MaybeScheduleCompaction(); background_work_finished_signal_.SignalAll(); } 这个函数主要调用BackgroundCompaction实现相应逻辑： void DBImpl::BackgroundCompaction() 如果immutable table存在则先调用CompactMemTable函数： mutex_.AssertHeld(); if (imm_ != nullptr) { CompactMemTable(); return; } 然后调用PickCompaction函数，PickCompaction函数选取合适的level中需要compact的文件，然后封装一个Compact对象： else { c = versions_->PickCompaction(); } 通过IsTrivialMove判断选出的sstable是否只有一个且在较低的level，并且该sstable与下下一level不会有太多的overlap，如果是的话，简单的将这个sstable移动到下一level。 Status status; if (c == nullptr) { // Nothing to do } else if (!is_manual && c->IsTrivialMove()) { // Move file to next level assert(c->num_input_files(0) == 1); FileMetaData *f = c->input(0, 0); c->edit()->DeleteFile(c->level(), f->number); c->edit()->AddFile(c->level() + 1, f->number, f->file_size, f->smallest, f->largest); status = versions_->LogAndApply(c->edit(), &mutex_); if (!status.ok()) { RecordBackgroundError(status); } VersionSet::LevelSummaryStorage tmp; Log(options_.info_log, "Moved #%lld to level-%d %lld bytes %s: %s\n", static_cast<unsigned long long>(f->number), c->level() + 1, static_cast<unsigned long long>(f->file_size), status.ToString().c_str(), versions_->LevelSummary(&tmp)); } 如果当前level中需要compact的sstable和level+2中的overlap没有超过阈值时，调用DoCompactionWork进行compact，然后调用CleanupCompaction清除compactstate对象，调用DeleteObsoleteFiles删除旧文件，回收内存和磁盘空间： else { CompactionState *compact = new CompactionState(c); status = DoCompactionWork(compact); if (!status.ok()) { RecordBackgroundError(status); } CleanupCompaction(compact); c->ReleaseInputs(); DeleteObsoleteFiles(); } delete c; 其中调用的PickCompaction函数为： Compaction *VersionSet::PickCompaction() 如果有level达到compact的阈值，优先将这个level进行compact。将最大的key大于上一次compact的最大的key的sstable加入inputs_[0]的数组中作为compact的sstable，如果compact_pointer_[level]中没有记录，则从头开始取sstable，如果不存在最大的key大于上一次compact的最大的key的sstable，则将key最小的sstable作为需要compact的sstable： Compaction *c; int level; // We prefer compactions triggered by too much data in a level over // the compactions triggered by seeks. const bool size_compaction = (current_->compaction_score_ >= 1); const bool seek_compaction = (current_->file_to_compact_ != nullptr); if (size_compaction) { level = current_->compaction_level_; assert(level >= 0); assert(level + 1 < config::kNumLevels); c = new Compaction(options_, level); // Pick the first file that comes after compact_pointer_[level] for (size_t i = 0; i < current_->files_[level].size(); i++) { FileMetaData *f = current_->files_[level][i]; if (compact_pointer_[level].empty() || icmp_.Compare(f->largest.Encode(), compact_pointer_[level]) > 0) { c->inputs_[0].push_back(f); break; } } if (c->inputs_[0].empty()) { // Wrap-around to the beginning of the key space c->inputs_[0].push_back(current_->files_[level][0]); } } 如果没有level达到需要compact的阈值，但是有文件因为seek的命中率不够而达到了compact的阈值，则将这个文件compact： else if (seek_compaction) { level = current_->file_to_compact_level_; c = new Compaction(options_, level); c->inputs_[0].push_back(current_->file_to_compact_); } 如果compact的是level0，则还需要调用GetOverlappingInputs函数求出有overlap的文件，然后再调用SetupOtherInputs函数完成compact对象的构建： c->input_version_ = current_; c->input_version_->Ref(); // Files in level 0 may overlap each other, so pick up all overlapping ones if (level == 0) { InternalKey smallest, largest; GetRange(c->inputs_[0], &smallest, &largest); // Note that the next call will discard the file we placed in // c->inputs_[0] earlier and replace it with an overlapping set // which will include the picked file. current_->GetOverlappingInputs(0, &smallest, &largest, &c->inputs_[0]); assert(!c->inputs_[0].empty()); } SetupOtherInputs(c); return c; CompactMemTable函数 void DBImpl::CompactMemTable() 调用WriteLevel0Table将immutable mmtable写入sstable： mutex_.AssertHeld(); assert(imm_ != nullptr); // Save the contents of the memtable as a new Table VersionEdit edit; Version *base = versions_->current(); base->Ref(); Status s = WriteLevel0Table(imm_, &edit, base); base->Unref(); if (s.ok() && shutting_down_.Acquire_Load()) { s = Status::IOError("Deleting DB during memtable compaction"); } 更新当前lognumber，旧日志不再需要，生效新的version： // Replace immutable memtable with the generated Table if (s.ok()) { edit.SetPrevLogNumber(0); edit.SetLogNumber(logfile_number_); // Earlier logs no longer needed s = versions_->LogAndApply(&edit, &mutex_); } 删除旧文件： if (s.ok()) { // Commit to the new state imm_->Unref(); imm_ = nullptr; has_imm_.Release_Store(nullptr); DeleteObsoleteFiles(); } else { RecordBackgroundError(s); } WriteLevel0Table函数为： Status DBImpl::WriteLevel0Table(MemTable *mem, VersionEdit *edit, Version *base) 调用BuildTable接口将immutable memtable写入sstable： mutex_.AssertHeld(); const uint64_t start_micros = env_->NowMicros(); FileMetaData meta; meta.number = versions_->NewFileNumber(); pending_outputs_.insert(meta.number); Iterator *iter = mem->NewIterator(); Log(options_.info_log, "Level-0 table #%llu: started", (unsigned long long)meta.number); Status s; { mutex_.Unlock(); s = BuildTable(dbname_, env_, options_, table_cache_, iter, &meta); mutex_.Lock(); } Log(options_.info_log, "Level-0 table #%llu: %lld bytes %s", (unsigned long long)meta.number, (unsigned long long)meta.file_size, s.ToString().c_str()); delete iter; pending_outputs_.erase(meta.number); 调用PickLevelForMemTableOutput函数选择合适的level将新的sstable加入： // Note that if file_size is zero, the file has been deleted and // should not be added to the manifest. int level = 0; if (s.ok() && meta.file_size > 0) { const Slice min_user_key = meta.smallest.user_key(); const Slice max_user_key = meta.largest.user_key(); if (base != nullptr) { level = base->PickLevelForMemTableOutput(min_user_key, max_user_key); } edit->AddFile(level, meta.number, meta.file_size, meta.smallest, meta.largest); } CompactionStats stats; stats.micros = env_->NowMicros() - start_micros; stats.bytes_written = meta.file_size; stats_[level].Add(stats); return s; PickLevelForMemTableOutput函数： int Version::PickLevelForMemTableOutput( const Slice &smallest_user_key, const Slice &largest_user_key) { int level = 0; if (!OverlapInLevel(0, &smallest_user_key, &largest_user_key)) { // Push to next level if there is no overlap in next level, // and the #bytes overlapping in the level after that are limited. InternalKey start(smallest_user_key, kMaxSequenceNumber, kValueTypeForSeek); InternalKey limit(largest_user_key, 0, static_cast<ValueType>(0)); std::vector<FileMetaData *> overlaps; while (level < config::kMaxMemCompactLevel) { if (OverlapInLevel(level + 1, &smallest_user_key, &largest_user_key)) { break; } if (level + 2 < config::kNumLevels) { // Check that file does not overlap too many grandparent bytes. GetOverlappingInputs(level + 2, &start, &limit, &overlaps); const int64_t sum = TotalFileSize(overlaps); if (sum > MaxGrandParentOverlapBytes(vset_->options_)) { break; } } level++; } } return level; } 选择level的策略为：如果level0和sstable的key有重合，则插入level0，否则循环搜索之后的level，如果下一level和sstable的key有重合，则直接返回，如果没有重合，考虑下下个level和sstable的重合的key的量，如果超过了设定的阈值，则返回（为了保证之后选取该sstable与下一level进行compact的时候，涉及下一level的文件不会太多，减小开销）。 DoCompactionWork函数 Status DBImpl::DoCompactionWork(CompactionState *compact) 将smallest_snapshot赋值为最老的snapshots的sequencenumber，如果不存在snapshots则赋值为当前数据库的sequencenumber： const uint64_t start_micros = env_->NowMicros(); int64_t imm_micros = 0; // Micros spent doing imm_ compactions Log(options_.info_log, "Compacting %d@%d + %d@%d files", compact->compaction->num_input_files(0), compact->compaction->level(), compact->compaction->num_input_files(1), compact->compaction->level() + 1); assert(versions_->NumLevelFiles(compact->compaction->level()) > 0); assert(compact->builder == nullptr); assert(compact->outfile == nullptr); if (snapshots_.empty()) { compact->smallest_snapshot = versions_->LastSequence(); } else { compact->smallest_snapshot = snapshots_.oldest()->sequence_number(); } 调用MakeInputIterator获取需要compact的文件上的迭代器，可以通过这个迭代器直接遍历所有需要compact的文件。 // Release mutex while we're actually doing the compaction work mutex_.Unlock(); Iterator *input = versions_->MakeInputIterator(compact->compaction); input->SeekToFirst(); Status status; ParsedInternalKey ikey; std::string current_user_key; bool has_current_user_key = false; SequenceNumber last_sequence_for_key = kMaxSequenceNumber; 开始通过迭代器遍历需要compact的文件。首先，如果有immutable memtable存在，首先调用CompactMemTable函数进行compact。 for (; input->Valid() && !shutting_down_.Acquire_Load();) { // Prioritize immutable compaction work if (has_imm_.NoBarrier_Load() != nullptr) { const uint64_t imm_start = env_->NowMicros(); mutex_.Lock(); if (imm_ != nullptr) { CompactMemTable(); // Wake up MakeRoomForWrite() if necessary. background_work_finished_signal_.SignalAll(); } mutex_.Unlock(); imm_micros += (env_->NowMicros() - imm_start); } 获取迭代器当前指向的entry的key值，调用ShouldStopBefore函数判断是否需要停止当前sstable的构建，如果需要，则调用FinishCompactionOutputFile函数将包含当前sstable的文件输出。ShouldStopBefore函数主要是判断当前的sstable中的key是否与下下一level中的key的overlap过大，如果过大则停止当前sstable的构建： Slice key = input->key(); if (compact->compaction->ShouldStopBefore(key) && compact->builder != nullptr) { status = FinishCompactionOutputFile(compact, input); if (!status.ok()) { break; } } 将key解码，key和之前的key重复且之前的key的sequencenumber比smallest_snapshot要小则丢弃这个key，因为既然之前重复的key的sequencenumber都比smallest_snapshot要小，当前key也肯定小，key标记为删除且之后的level中不存在这个key（也就是说这个key的删除不会影响到之后）也丢弃这个key。 // Handle key/value, add to state, etc. bool drop = false; if (!ParseInternalKey(key, &ikey)) { // Do not hide error keys current_user_key.clear(); has_current_user_key = false; last_sequence_for_key = kMaxSequenceNumber; } else { if (!has_current_user_key || user_comparator()->Compare(ikey.user_key, Slice(current_user_key)) != 0) { // First occurrence of this user key current_user_key.assign(ikey.user_key.data(), ikey.user_key.size()); has_current_user_key = true; last_sequence_for_key = kMaxSequenceNumber; } if (last_sequence_for_key <= compact->smallest_snapshot) { // Hidden by an newer entry for same user key drop = true; // (A) } else if (ikey.type == kTypeDeletion && ikey.sequence <= compact->smallest_snapshot && compact->compaction->IsBaseLevelForKey(ikey.user_key)) { // For this user key: // (1) there is no data in higher levels // (2) data in lower levels will have larger sequence numbers // (3) data in layers that are being compacted here and have // smaller sequence numbers will be dropped in the next // few iterations of this loop (by rule (A) above). // Therefore this deletion marker is obsolete and can be dropped. drop = true; } last_sequence_for_key = ikey.sequence; } 如果不丢弃key的话，当前的写入sstable的文件不存在则调用OpenCompactionOutputFile函数创建一个文件，然后将KV写入，如果写入后sstable大小达到阈值则调用FinishCompactionOutputFile函数写出这个文件。 if (!drop) { // Open output file if necessary if (compact->builder == nullptr) { status = OpenCompactionOutputFile(compact); if (!status.ok()) { break; } } if (compact->builder->NumEntries() == 0) { compact->current_output()->smallest.DecodeFrom(key); } compact->current_output()->largest.DecodeFrom(key); compact->builder->Add(key, input->value()); // Close output file if it is big enough if (compact->builder->FileSize() >= compact->compaction->MaxOutputFileSize()) { status = FinishCompactionOutputFile(compact, input); if (!status.ok()) { break; } } } 循环以上过程： input->Next(); } 结束遍历之后，将没有写出的文件写出： if (status.ok() && shutting_down_.Acquire_Load()) { status = Status::IOError("Deleting DB during compaction"); } if (status.ok() && compact->builder != nullptr) { status = FinishCompactionOutputFile(compact, input); } if (status.ok()) { status = input->status(); } delete input; input = nullptr; 记录统计信息： CompactionStats stats; stats.micros = env_->NowMicros() - start_micros - imm_micros; for (int which = 0; which < 2; which++) { for (int i = 0; i < compact->compaction->num_input_files(which); i++) { stats.bytes_read += compact->compaction->input(which, i)->file_size; } } for (size_t i = 0; i < compact->outputs.size(); i++) { stats.bytes_written += compact->outputs[i].file_size; } mutex_.Lock(); stats_[compact->compaction->level() + 1].Add(stats); 将此次compaction生效： if (status.ok()) { status = InstallCompactionResults(compact); } if (!status.ok()) { RecordBackgroundError(status); } VersionSet::LevelSummaryStorage tmp; Log(options_.info_log, "compacted to: %s", versions_->LevelSummary(&tmp)); return status; 230 Love u
2019-04-05
优秀的个人博客,低调大师

LevelDB源码分析-编码

编码（util/coding.h util/coding.cc） LevelDB将整型编码为二进制字符串的形式，同时又能够和ASCII字符区分。 首先是定长编码： void EncodeFixed32(char *buf, uint32_t value) { if (port::kLittleEndian) { memcpy(buf, &value, sizeof(value)); } else { buf[0] = value & 0xff; buf[1] = (value >> 8) & 0xff; buf[2] = (value >> 16) & 0xff; buf[3] = (value >> 24) & 0xff; } } void EncodeFixed64(char *buf, uint64_t value) { if (port::kLittleEndian) { memcpy(buf, &value, sizeof(value)); } else { buf[0] = value & 0xff; buf[1] = (value >> 8) & 0xff; buf[2] = (value >> 16) & 0xff; buf[3] = (value >> 24) & 0xff; buf[4] = (value >> 32) & 0xff; buf[5] = (value >> 40) & 0xff; buf[6] = (value >> 48) & 0xff; buf[7] = (value >> 56) & 0xff; } } 这里根据机器区分大端和小端，LevelDB编码后的字符串为小端存储。在编码时，只是简单的将8位二进制码存储在一个char字符的位置上。因为定长，所以可以和ASCII字符区分。 接下来是定长编码的一些接口函数： void PutFixed32(std::string *dst, uint32_t value) // 将一个32位整型值定长编码后存入dst void PutFixed64(std::string *dst, uint64_t value) // 将一个64位整型值定长编码后存入dst 然后是变长编码： char *EncodeVarint32(char *dst, uint32_t v) { // Operate on characters as unsigneds unsigned char *ptr = reinterpret_cast<unsigned char *>(dst); static const int B = 128; if (v < (1 << 7)) { *(ptr++) = v; } else if (v < (1 << 14)) { *(ptr++) = v | B; *(ptr++) = v >> 7; } else if (v < (1 << 21)) { *(ptr++) = v | B; *(ptr++) = (v >> 7) | B; *(ptr++) = v >> 14; } else if (v < (1 << 28)) { *(ptr++) = v | B; *(ptr++) = (v >> 7) | B; *(ptr++) = (v >> 14) | B; *(ptr++) = v >> 21; } else { *(ptr++) = v | B; *(ptr++) = (v >> 7) | B; *(ptr++) = (v >> 14) | B; *(ptr++) = (v >> 21) | B; *(ptr++) = v >> 28; } return reinterpret_cast<char *>(ptr); } char *EncodeVarint64(char *dst, uint64_t v) { static const int B = 128; unsigned char *ptr = reinterpret_cast<unsigned char *>(dst); while (v >= B) { *(ptr++) = (v & (B - 1)) | B; v >>= 7; } *(ptr++) = static_cast<unsigned char>(v); return reinterpret_cast<char *>(ptr); } LevelDB的变长编码设计的十分巧妙，它以7个二进制bit为一个单位，存入一个char中，同时为了和ASCII码进行区分，将char的最高位设为1（ASCII码为0-127），同样采用小端存储的形式。但是变长编码的最后一个char的最高位是0，以此作为变长编码后的字符串的结束标志。 例如，11001011101111001会被编码为11111001 10101110 00000110。 接下来是一些变长编码的接口函数： void PutVarint32(std::string *dst, uint32_t v) // 将一个32位整型值变长编码后存入dst void PutVarint64(std::string *dst, uint64_t v) // 将一个64位整型值变长编码后存入dst int VarintLength(uint64_t v) // 获取变长编码后的字符串长度(以字节计数) const char *GetVarint32PtrFallback(const char *p, const char *limit, uint32_t *value) // 将以p到limit之间的变长编码字符串解码为32位整型值 bool GetVarint32(Slice *input, uint32_t *value) // 将以p到limit之间的变长编码字符串解码为32位整型值并封装入Slice中 const char *GetVarint64Ptr(const char *p, const char *limit, uint64_t *value) // 将以p到limit之间的变长编码字符串解码为64位整型值 bool GetVarint64(Slice *input, uint64_t *value) // 将以p到limit之间的变长编码字符串解码为64位整型值并封装入Slice中 227 Love u
2019-04-05

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

时间: 2026-04-11 大小: 211.28KB 下载: 56次
Spring

Spring

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

时间: 2026-08-21 大小: 5MB 下载: 2次
Sublime Text

Sublime Text

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

时间: 2026-08-21 大小: 9.84MB 下载: 40次
WebStorm

WebStorm

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

时间: 2025-12-13 大小: 249.54MB 下载: 22次
热门标签
DeepSeek Jdk25 OpenAi HarmonyOS6 Nacos3 SpringBoot4 SpringCloud Docker Kubernetes Service Mesh Redis Gemini3 Rocky

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册