how-to Documentation

if (key == null || value == null) throw new NullPointerException();
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;
	if (tab == null || (n = tab.length) == 0)
		tab = initTable();
	else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
		if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
			break;                   // no lock when adding to empty bin
	}
	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;

V oldVal = null;
synchronized (f) {
	if (tabAt(tab, i) == f) {
		if (fh >= 0) {
			binCount = 1;
			for (Node<K,V> e = f;; ++binCount) {
				K ek;
				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");
	}
}

Input：coins = [1, 2, 5], amount = 11
Output: 3
Explain: 11 = 5 + 5 + 1

dp[i - 1][j - k*coins[i]] + k

public int coinChange(int[] coins, int amount) {
	if (coins == null || coins.length == 0 || amount == 0) return 0;
	int[][] arr = new int[coins.length][amount + 1];
	for (int i = 0; i <= amount; i++) {
		if (i % coins[0] == 0) arr[0][i] = i / coins[0];
		else arr[0][i] = Integer.MAX_VALUE;
	}
	int result = arr[0][amount];
	for (int i = 1; i < coins.length; i++) {
		for (int j = 0, k = 0; j <= amount; j++, k++) {
			if (j < coins[i]) {
				arr[i][j] = arr[i - 1][j];
			} else {
				arr[i][j] = Math.min(arr[i - 1][j] - 1, arr[i][j - coins[i]]) + 1;
			}
			if (j == amount) {
				result = Math.min(result, arr[i][j]);
			}
		}
	}
	
	if (amount != 0 && result == Integer.MAX_VALUE) {
		return -1;
	} else {
		return result;
	}
}

break      default       func     interface   select
case       defer         go       map         struct
chan       else          goto     package     switch
const      fallthrough   if       range       type
continue   for           import   return      var

内建常量: true false iota nil

内建类型: int int8 int16 int32 int64
          uint uint8 uint16 uint32 uint64 uintptr
          float32 float64 complex128 complex64
          bool byte rune string error

内建函数: make len cap new append copy close delete
          complex real imag
          panic recover

var 变量名称 类型 = 表达式

变量名 := 表达式

i := 100
i, j := 1, 0 // 多个变量赋值
f, err := os.Open(name) // 利用函数的返回值进行赋值

i := 1
p := &i // 声明p为变量x的指针
fmt.print(*p) // 1
*p = 2
fmt.print(*p) // 2

p := new(int) // 声明一个匿名的int变量,p为匿名变量的指针
*p = 2 // 通过指针p改变匿名变量的值

输入: [1,7,4,9,2,5]
输出: 6 
解释: 整个序列均为摆动序列。

输入: [1,17,5,10,13,15,10,5,16,8]
输出: 7
解释: 这个序列包含几个长度为 7 摆动序列，其中一个可为[1,17,10,13,10,16,8]。

输入: [1,2,3,4,5,6,7,8,9]
输出: 2

static final boolean UP = true;
static final boolean DOWN = false;
public int wiggleMaxLength(int[] nums) {
	if (nums == null || nums.length == 0) return 0;
	if (nums.length == 1) return 1;
	int i = getFirstValidIndex(nums);
	if (i == nums.length - 1) return 1;
	boolean cur = nums[i] > nums[i + 1] ? UP : DOWN;
	int result = 1;
	while (i != nums.length - 1) {
		if (cur == UP) {
			int pre = nums[i++];
			while (i != nums.length - 1 && nums[i] < pre && nums[i] >= nums[i + 1]) {
				i++;
			}
			cur = DOWN;
			result++;
		} else {
			int pre = nums[i++];
			while (i != nums.length - 1 && nums[i] > pre && nums[i] <= nums[i + 1]) {
				i++;
			}
			cur = UP;
			result++;
		}
	}
	return result;
}

public int getFirstValidIndex(int[] nums) {
	int result = 0;
	for (;;) {
		if (result == nums.length - 1 || nums[result] != nums[result + 1]) 
			break;
		result++;
		continue;
	}
	return result;
}

apiVersion: v1 # 遵循v1版本的Kubernetes API
kind: Pod # 这个yaml描述的是一个Pod
metadata:
    name: kubia-manual-v2 # Pod的名称
    labels: 
      creation_method: manual
      env: prod
spec:
    containers:
    - image: docker.io/hengyoush/kubia # 创建容器使用的镜像
      name: kubia #容器的名称
      ports:
      - containerPort: 8080 # 应用监听的端口, 指定端口是展示性的
        protocol: TCP

kubectl create -f kubia-manual-with-labels.yaml

kubectl get po --show-labels

NAME              READY   STATUS    RESTARTS   AGE    LABELS
kubia-manual-v2   1/1     Running   0          47s    creation_method=manual,env=prod

kubectl get po -L env

NAME              READY   STATUS    RESTARTS   AGE    ENV 
kubia-manual-v2   1/1     Running   0          52s    prod

kubectl label po kubia-manual creation_method=manual

kubectl label po kubia-manual-v2 env=debug --overwrite

NAME              READY   STATUS    RESTARTS   AGE     LABELS
kubia-d58dn       1/1     Running   0          146m    run=kubia
kubia-manual      1/1     Running   0          43m     creation_method=manual
kubia-manual-v2   1/1     Running   0          7m13s   creation_method=manual,env=debug

kubectl get po -l creation_method=manual

kubectl get po -l env

kubectl get po -l '!env'

kubectl get po -l 'creation_method!=manual'

kubectl get po -l 'env in (prod,debug)'

kubectl get po -l 'env notin (prod,debug)'

kubectl get po -l 'env notin (prod,debug)', reation_method=manual

kubectl label node minikube gpu=true

apiVersion: v1 # 遵循v1版本的Kubernetes API
kind: Pod # 这个yaml描述的是一个Pod
metadata:
    name: kubia-manual # Pod的名称
spec:
    nodeSelector:
      gpu: "true"
    containers:
    - image: docker.io/hengyoush/kubia # 创建容器使用的镜像
      name: kubia #容器的名称

brew install kubernetes-cli

kubectl version

curl -Lo minikube http://kubernetes.oss-cn-hangzhou.aliyuncs.com/minikube/releases/v0.28.1/minikube-darwin-amd64 && chmod +x minikube && sudo mv minikube /usr/local/bin/

minikube start --registry-mirror=https://registry.docker-cn.com

😄  minikube v1.1.0 on darwin (amd64)
✅  using image repository registry.cn-hangzhou.aliyuncs.com/google_containers
🔥  Creating virtualbox VM (CPUs=2, Memory=2048MB, Disk=20000MB) ...
🐳  Configuring environment for Kubernetes v1.14.2 on Docker 18.09.6
🚜  Pulling images ...
🚀  Launching Kubernetes ... 
⌛  Verifying: apiserver proxy etcd scheduler controller dns
🏄  Done! kubectl is now configured to use "minikube"

kubectl run kubia --image=docker.io/hengyoush/kubia --port=8080 --generator=run/v1

kubectl get pods

NAME          READY     STATUS    RESTARTS   AGE
kubia-d58dn   1/1       Running   0          29s

kubectl expose rc kubia --type=LoadBalancer --name kubia-http

kubectl get svc

NAME         TYPE           CLUSTER-IP    EXTERNAL-IP   PORT(S)          AGE
kubernetes   ClusterIP      10.96.0.1     <none>        443/TCP          18h
kubia-http   LoadBalancer   10.105.25.8   <pending>     8080:32055/TCP   3m

kubectl scale rc kubia --replicas=3

NAME          READY     STATUS              RESTARTS   AGE
kubia-7rvr6   0/1       ContainerCreating   0          23s
kubia-d58dn   1/1       Running             0          22m
kubia-tkqbm   0/1       ContainerCreating   0          23s

➜  ~ curl http://192.168.99.104:32055/
You've hit kubia-7rvr6
➜  ~ curl http://192.168.99.104:32055/
You've hit kubia-tkqbm
➜  ~ curl http://192.168.99.104:32055/
You've hit kubia-tkqbm
➜  ~ curl http://192.168.99.104:32055/
You've hit kubia-d58dn

minikube dashboard

CASE(_monitorenter): {
            oop lockee = STACK_OBJECT(-1);
            // derefing's lockee ought to provoke implicit null check
            CHECK_NULL(lockee);
            // find a free monitor or one already allocated for this object
            // if we find a matching object then we need a new monitor
            // since this is recursive enter
            BasicObjectLock* limit = istate->monitor_base();
            BasicObjectLock* most_recent = (BasicObjectLock*) istate->stack_base();
            BasicObjectLock* entry = NULL;
            while (most_recent != limit ) { // 找到一个地址最高的空闲的lockrecord
                if (most_recent->obj() == NULL) entry = most_recent;
                else if (most_recent->obj() == lockee) break; // 重入
                most_recent++;
            }
        ...

BasicObjectLock
    |
    |--lock
       |--displaced_header
    |--obj

if (entry != NULL) {
          entry->set_obj(lockee); // 设置这个对象
          int success = false;
          uintptr_t epoch_mask_in_place = (uintptr_t)markOopDesc::epoch_mask_in_place;

          markOop mark = lockee->mark(); // 获取markword
          intptr_t hash = (intptr_t) markOopDesc::no_hash;
          ...

// implies UseBiasedLocking
if (mark->has_bias_pattern()) { // 可偏向, 看markword最后是不是101

uintptr_t thread_ident; // 当前线程id
uintptr_t anticipated_bias_locking_value;
thread_ident = (uintptr_t)istate->thread();
anticipated_bias_locking_value =
    (((uintptr_t)lockee->klass()->prototype_header() | thread_ident) ^ (uintptr_t)mark) &
    ~((uintptr_t) markOopDesc::age_mask_in_place);

if  (anticipated_bias_locking_value == 0) { // 已偏向当前线程
   // already biased towards this thread, nothing to do
   if (PrintBiasedLockingStatistics) {
   (* BiasedLocking::biased_lock_entry_count_addr())++;
   }
   success = true;
}

else if ((anticipated_bias_locking_value & markOopDesc::biased_lock_mask_in_place) != 0) {
    // try revoke bias
    markOop header = lockee->klass()->prototype_header(); // Klass中也有一个Markword
    if (hash != markOopDesc::no_hash) {
    header = header->copy_set_hash(hash);
    }
    if (lockee->cas_set_mark(header, mark) == mark) {// 将对象的markword替换为class中的markword
    if (PrintBiasedLockingStatistics)
        (*BiasedLocking::revoked_lock_entry_count_addr())++;
    }
}

// epoch不等于class中的epoch，尝试重偏向
else if ((anticipated_bias_locking_value & epoch_mask_in_place) !=0) {
    // try rebias
    // 构造一个偏向当前线程的markword
    markOop new_header = (markOop) ( (intptr_t) lockee->klass()->prototype_header() | thread_ident);
    if (hash != markOopDesc::no_hash) {
    new_header = new_header->copy_set_hash(hash);
    }
    if (lockee->cas_set_mark(new_header, mark) == mark) {
    if (PrintBiasedLockingStatistics)
        (* BiasedLocking::rebiased_lock_entry_count_addr())++;
    }
    else {// CAS偏向失败，使用monitorenter进入轻量级锁逻辑
    CALL_VM(InterpreterRuntime::monitorenter(THREAD, entry), handle_exception);
    }
    success = true;
}

else {
// 构造一个匿名偏向的markword
// try to bias towards thread in case object is anonymously biased
markOop header = (markOop) ((uintptr_t) mark & ((uintptr_t)markOopDesc::biased_lock_mask_in_place |
                                                (uintptr_t)markOopDesc::age_mask_in_place |
                                                epoch_mask_in_place));
if (hash != markOopDesc::no_hash) {
    header = header->copy_set_hash(hash);
}
markOop new_header = (markOop) ((uintptr_t) header | thread_ident);
// debugging hint
DEBUG_ONLY(entry->lock()->set_displaced_header((markOop) (uintptr_t) 0xdeaddead);)
if (lockee->cas_set_mark(new_header, header) == header) {
    if (PrintBiasedLockingStatistics)
    (* BiasedLocking::anonymously_biased_lock_entry_count_addr())++;
}
else {
    CALL_VM(InterpreterRuntime::monitorenter(THREAD, entry), handle_exception);
}
success = true;

// traditional lightweight locking
    if (!success) {
    markOop displaced = lockee->mark()->set_unlocked();
    entry->lock()->set_displaced_header(displaced);// lockrecord中存放原先的markword
    bool call_vm = UseHeavyMonitors;
    if (call_vm || lockee->cas_set_mark((markOop)entry, displaced) != displaced) {// 将对象的markword指向lockrecord
        // Is it simple recursive case? // 判断是不是重入，如果是，简单将displacedHeader设置为null
        if (!call_vm && THREAD->is_lock_owned((address) displaced->clear_lock_bits())) {
        entry->lock()->set_displaced_header(NULL);
        } else {
        CALL_VM(InterpreterRuntime::monitorenter(THREAD, entry), handle_exception);
        }
    }
    }

if (UseBiasedLocking) {
    // Retry fast entry if bias is revoked to avoid unnecessary inflation
    ObjectSynchronizer::fast_enter(h_obj, elem->lock(), true, CHECK);
} else {
    ObjectSynchronizer::slow_enter(h_obj, elem->lock(), CHECK);
}

void ObjectSynchronizer::fast_enter(Handle obj, BasicLock* lock,
                                    bool attempt_rebias, TRAPS) {
  if (UseBiasedLocking) {
    if (!SafepointSynchronize::is_at_safepoint()) {
      BiasedLocking::Condition cond = BiasedLocking::revoke_and_rebias(obj, attempt_rebias, THREAD);
      if (cond == BiasedLocking::BIAS_REVOKED_AND_REBIASED) {
        return;
      }
    } else {
      assert(!attempt_rebias, "can not rebias toward VM thread");
      BiasedLocking::revoke_at_safepoint(obj);
    }
    assert(!obj->mark()->has_bias_pattern(), "biases should be revoked by now");
  }

  slow_enter(obj, lock, THREAD);
}

if (mark->is_biased_anonymously() && !attempt_rebias) { // hashcode方法
    markOop biased_value       = mark;
    markOop unbiased_prototype = markOopDesc::prototype()->set_age(mark->age()); // prototype是nohash + 1(无锁)
    markOop res_mark = obj->cas_set_mark(unbiased_prototype, mark);
    if (res_mark == biased_value) { // 如果成功设置,说明没有竞争
      return BIAS_REVOKED;
    }
  }

else if (mark->has_bias_pattern()) {
    Klass* k = obj->klass();
    markOop prototype_header = k->prototype_header();
    if (!prototype_header->has_bias_pattern()) { // class关闭了偏向，简单cas即可，可能race codition，但是不要紧
      markOop biased_value       = mark;
      markOop res_mark = obj->cas_set_mark(prototype_header, mark);
      assert(!obj->mark()->has_bias_pattern(), "even if we raced, should still be revoked");
      return BIAS_REVOKED;
    } else if (prototype_header->bias_epoch() != mark->bias_epoch()) { // epoch过时,该对象不是锁
      if (attempt_rebias) {
        assert(THREAD->is_Java_thread(), "");
        markOop biased_value       = mark;
        markOop rebiased_prototype = markOopDesc::encode((JavaThread*) THREAD, mark->age(), prototype_header->bias_epoch());
        markOop res_mark = obj->cas_set_mark(rebiased_prototype, mark); // 尝试重偏向到当前线程
        if (res_mark == biased_value) {
          return BIAS_REVOKED_AND_REBIASED; // 注意这里偏向成功则不会进入slow_enter
        }
      } else {
        markOop biased_value       = mark;
        markOop unbiased_prototype = markOopDesc::prototype()->set_age(mark->age());// 置为无锁
        markOop res_mark = obj->cas_set_mark(unbiased_prototype, mark);
        if (res_mark == biased_value) {
          return BIAS_REVOKED;
        }
      }
    }
  }

else if (heuristics == HR_SINGLE_REVOKE) {
    Klass *k = obj->klass();
    markOop prototype_header = k->prototype_header();
    if (mark->biased_locker() == THREAD &&
        prototype_header->bias_epoch() == mark->bias_epoch()) {// 需要撤销当前线程的偏向
      ResourceMark rm;
      log_info(biasedlocking)("Revoking bias by walking my own stack:");
      EventBiasedLockSelfRevocation event;
      BiasedLocking::Condition cond = revoke_bias(obj(), false, false, (JavaThread*) THREAD, NULL);
      ((JavaThread*) THREAD)->set_cached_monitor_info(NULL);
      assert(cond == BIAS_REVOKED, "why not?");
      if (event.should_commit()) {
        post_self_revocation_event(&event, k);
      }
      return cond;
    } else { // 如果不是偏向当前线程需要到safepoint再撤销
      EventBiasedLockRevocation event;
      VM_RevokeBias revoke(&obj, (JavaThread*) THREAD);
      VMThread::execute(&revoke);
      if (event.should_commit() && revoke.status_code() != NOT_BIASED) {
        post_revocation_event(&event, k, &revoke);
      }
      return revoke.status_code();
    }
  }

EventBiasedLockClassRevocation event;
  VM_BulkRevokeBias bulk_revoke(&obj, (JavaThread*) THREAD,
                                (heuristics == HR_BULK_REBIAS),
                                attempt_rebias);
  VMThread::execute(&bulk_revoke);

uint age = mark->age();
markOop   biased_prototype = markOopDesc::biased_locking_prototype()->set_age(age);
markOop unbiased_prototype = markOopDesc::prototype()->set_age(age);

bool thread_is_alive = false;
  if (requesting_thread == biased_thread) {
    thread_is_alive = true;
  } else {
    ThreadsListHandle tlh;
    thread_is_alive = tlh.includes(biased_thread);
  }
  if (!thread_is_alive) {
    if (allow_rebias) {
      obj->set_mark(biased_prototype);
    } else {
      obj->set_mark(unbiased_prototype);
    }
    return BiasedLocking::BIAS_REVOKED;

GrowableArray<MonitorInfo*>* cached_monitor_info = get_or_compute_monitor_info(biased_thread);
  BasicLock* highest_lock = NULL;
  for (int i = 0; i < cached_monitor_info->length(); i++) {
    MonitorInfo* mon_info = cached_monitor_info->at(i);
    if (oopDesc::equals(mon_info->owner(), obj)) {
      log_trace(biasedlocking)("   mon_info->owner (" PTR_FORMAT ") == obj (" PTR_FORMAT ")",
                               p2i((void *) mon_info->owner()),
                               p2i((void *) obj));
      // Assume recursive case and fix up highest lock later
      markOop mark = markOopDesc::encode((BasicLock*) NULL);
      highest_lock = mon_info->lock();
      highest_lock->set_displaced_header(mark);
    } else {
      log_trace(biasedlocking)("   mon_info->owner (" PTR_FORMAT ") != obj (" PTR_FORMAT ")",
                               p2i((void *) mon_info->owner()),
                               p2i((void *) obj));
    }
  }
  if (highest_lock != NULL) {
    highest_lock->set_displaced_header(unbiased_prototype);
    obj->release_set_mark(markOopDesc::encode(highest_lock));

if (bulk_rebias) {
      if (klass->prototype_header()->has_bias_pattern()) {
        int prev_epoch = klass->prototype_header()->bias_epoch();
        klass->set_prototype_header(klass->prototype_header()->incr_bias_epoch());
        int cur_epoch = klass->prototype_header()->bias_epoch();

        // Now walk all threads' stacks and adjust epochs of any biased
        // and locked objects of this data type we encounter
        for (; JavaThread *thr = jtiwh.next(); ) {
          GrowableArray<MonitorInfo*>* cached_monitor_info = get_or_compute_monitor_info(thr);
          for (int i = 0; i < cached_monitor_info->length(); i++) {
            MonitorInfo* mon_info = cached_monitor_info->at(i);
            oop owner = mon_info->owner();
            markOop mark = owner->mark();
            if ((owner->klass() == k_o) && mark->has_bias_pattern()) {
              // We might have encountered this object already in the case of recursive locking
              assert(mark->bias_epoch() == prev_epoch || mark->bias_epoch() == cur_epoch, "error in bias epoch adjustment");
              owner->set_mark(mark->set_bias_epoch(cur_epoch));
            }
          }
        }
      }

      // At this point we're done. All we have to do is potentially
      // adjust the header of the given object to revoke its bias.
      revoke_bias(o, attempt_rebias_of_object && klass->prototype_header()->has_bias_pattern(), true, requesting_thread, NULL);

klass->set_prototype_header(markOopDesc::prototype());

    // Now walk all threads' stacks and forcibly revoke the biases of
    // any locked and biased objects of this data type we encounter.
    for (; JavaThread *thr = jtiwh.next(); ) {
    GrowableArray<MonitorInfo*>* cached_monitor_info = get_or_compute_monitor_info(thr);
    for (int i = 0; i < cached_monitor_info->length(); i++) {
        MonitorInfo* mon_info = cached_monitor_info->at(i);
        oop owner = mon_info->owner();
        markOop mark = owner->mark();
        if ((owner->klass() == k_o) && mark->has_bias_pattern()) {
        revoke_bias(owner, false, true, requesting_thread, NULL);
        }
    }
    }

    // Must force the bias of the passed object to be forcibly revoked
    // as well to ensure guarantees to callers
    revoke_bias(o, false, true, requesting_thread, NULL);

if (UseBiasedLocking) {
    if (!SafepointSynchronize::is_at_safepoint()) {
      BiasedLocking::Condition cond = BiasedLocking::revoke_and_rebias(obj, attempt_rebias, THREAD);
        if (cond == BiasedLocking::BIAS_REVOKED_AND_REBIASED) {
        return;
      }
    } else {
      assert(!attempt_rebias, "can not rebias toward VM thread");
      BiasedLocking::revoke_at_safepoint(obj);
    }
    assert(!obj->mark()->has_bias_pattern(), "biases should be revoked by now");
  }

  slow_enter(obj, lock, THREAD); // 进入slow-enter

void ObjectSynchronizer::slow_enter(Handle obj, BasicLock* lock, TRAPS) {
  markOop mark = obj->mark();
  assert(!mark->has_bias_pattern(), "should not see bias pattern here");

  if (mark->is_neutral()) {// 如果当前无锁，那么尝试使用lockrecord的地址cas掉对象的mark
    lock->set_displaced_header(mark);
    if (mark == obj()->cas_set_mark((markOop) lock, mark)) {
      TEVENT(slow_enter: release stacklock);
      return;
    }
    // cas失败, 膨胀至重量级锁
  } else if (mark->has_locker() &&
             THREAD->is_lock_owned((address)mark->locker())) { // 处理重入
    assert(lock != mark->locker(), "must not re-lock the same lock");
    assert(lock != (BasicLock*)obj->mark(), "don't relock with same BasicLock");
    lock->set_displaced_header(NULL);
    return;
  }

  // 将displaced——header设置为0(unused)
  lock->set_displaced_header(markOopDesc::unused_mark());
  ObjectSynchronizer::inflate(THREAD, // 膨胀
                              obj(),
                              inflate_cause_monitor_enter)->enter(THREAD);
}

for (;;) {
    const markOop mark = object->mark();
    assert(!mark->has_bias_pattern(), "invariant");

    // 当前对象的markword
    // *  Inflated-已膨胀     - 其他线程将其膨胀了,直接返回
    // *  Stack-locked-轻量级锁 - 尝试膨胀
    // *  INFLATING-膨胀中    - 等待其他线程将其膨胀完成
    // *  Neutral-无锁      - 尝试膨胀
    // *  BIASED-偏向状态       - 非法,如果处于偏向锁模式,说明没有撤销,应该先撤销

    // CASE: inflated--已膨胀,直接返回
    if (mark->has_monitor()) {
      ObjectMonitor * inf = mark->monitor();
      assert(inf->header()->is_neutral(), "invariant");
      assert(oopDesc::equals((oop) inf->object(), object), "invariant");
      assert(ObjectSynchronizer::verify_objmon_isinpool(inf), "monitor is invalid");
      return inf;
    }

    // CASE: 正在膨胀,等待自旋然后返回
    if (mark == markOopDesc::INFLATING()) {
      TEVENT(Inflate: spin while INFLATING); // 自旋
      ReadStableMark(object);
      continue;
    }

    // CASE: stack-locked-轻量级锁,尝试设置INFLATING,如果成功替换掉对象的markword
    // 为当前monitor地址

    if (mark->has_locker()) {
      ObjectMonitor * m = omAlloc(Self);
      // 优化:先准备好monitor,然后再设置INFLATING,防止INFLATING时间过长
      m->Recycle();
      m->_Responsible  = NULL;
      m->_recursions   = 0;
      m->_SpinDuration = ObjectMonitor::Knob_SpinLimit;   // Consider: maintain by type/class

      markOop cmp = object->cas_set_mark(markOopDesc::INFLATING(), mark);
      if (cmp != mark) { // 其他线程成功设置, 重试
        omRelease(Self, m, true);
        continue;       // Interference -- just retry
      }

      // 为什么使用一个INFLATING而不是直接设置monitor呢?
      // 这是因为在轻量级锁解锁的过程中,要将displacedHeader还原到对象头中,这时我们设置一个INFLATING
      // 可以让它cas失败,然后进入重量级锁的释放流程,而不是直接还原,造成hashcode值莫名其妙的改变
      markOop dmw = mark->displaced_mark_helper();
      assert(dmw->is_neutral(), "invariant");

      // Setup monitor fields to proper values -- prepare the monitor
      m->set_header(dmw);

      // 将owner设置为lockrecord
      m->set_owner(mark->locker());
      m->set_object(object);

      guarantee(object->mark() == markOopDesc::INFLATING(), "invariant");
      object->release_set_mark(markOopDesc::encode(m)); // 设置对象的mark为monitor的地址

      ...

      return m;
    }

    // CASE: 无锁,尝试直接替换markword

    assert(mark->is_neutral(), "invariant");
    ObjectMonitor * m = omAlloc(Self);
    // prepare m for installation - set monitor to initial state
    m->Recycle();
    m->set_header(mark);
    m->set_owner(NULL);
    m->set_object(object);
    m->_recursions   = 0;
    m->_Responsible  = NULL;
    m->_SpinDuration = ObjectMonitor::Knob_SpinLimit;       // consider: keep metastats by type/class

    if (object->cas_set_mark(markOopDesc::encode(m), mark) != mark) {
      m->set_object(NULL);
      m->set_owner(NULL);
      m->Recycle();
      omRelease(Self, m, true);
      m = NULL;
      continue;
      // interference - the markword changed - just retry.
      // The state-transitions are one-way, so there's no chance of
      // live-lock -- "Inflated" is an absorbing state.
    }

void ObjectMonitor::enter(TRAPS) {
  Thread * const Self = THREAD;

  void * cur = Atomic::cmpxchg(Self, &_owner, (void*)NULL);// 尝试cas掉owner字段,只有在owner为空的情况下才成功.
  if (cur == NULL) {
    // Either ASSERT _recursions == 0 or explicitly set _recursions = 0.
    assert(_recursions == 0, "invariant");
    assert(_owner == Self, "invariant");
    return;
  }

  if (cur == Self) { // 重入,简单自增_recursions
    // TODO-FIXME: check for integer overflow!  BUGID 6557169.
    _recursions++;
    return;
  }

  if (Self->is_lock_owned ((address)cur)) { // 当前线程就是轻量级锁的拥有者,直接设置_owner
    assert(_recursions == 0, "internal state error");
    _recursions = 1;
    _owner = Self;
    return;
  }

  Self->_Stalled = intptr_t(this);

  // 尝试自旋
  if (Knob_SpinEarly && TrySpin (Self) > 0) {
    assert(_owner == Self, "invariant");
    assert(_recursions == 0, "invariant");
    assert(((oop)(object()))->mark() == markOopDesc::encode(this), "invariant");
    Self->_Stalled = 0;
    return;
  }

  Atomic::inc(&_count);

  EventJavaMonitorEnter event;
  if (event.should_commit()) {
    event.set_monitorClass(((oop)this->object())->klass());
    event.set_address((uintptr_t)(this->object_addr()));
  }

  { // 设置java线程状态
    JavaThreadBlockedOnMonitorEnterState jtbmes(jt, this);

    Self->set_current_pending_monitor(this);

    OSThreadContendState osts(Self->osthread());
    ThreadBlockInVM tbivm(jt);

    for (;;) {
        // 用于suspend支持,这块没看太懂,但是现在基本不会使用suspend方法了,所以没多大用?
      jt->set_suspend_equivalent();

      EnterI(THREAD); // 进入队列_cxq竞争锁

      if (!ExitSuspendEquivalent(jt)) break;

      // We have acquired the contended monitor, but while we were
      // waiting another thread suspended us. We don't want to enter
      // the monitor while suspended because that would surprise the
      // thread that suspended us.
      //
      _recursions = 0;
      _succ = NULL;
      exit(false, Self);

      jt->java_suspend_self();
    }
    Self->set_current_pending_monitor(NULL);
  }

  Atomic::dec(&_count);
  Self->_Stalled = 0;

  ...
}

void ObjectMonitor::EnterI(TRAPS) {
  Thread * const Self = THREAD;
  assert(Self->is_Java_thread(), "invariant");
  assert(((JavaThread *) Self)->thread_state() == _thread_blocked, "invariant");

  // Try the lock - TATAS
  if (TryLock (Self) > 0) {
    assert(_succ != Self, "invariant");
    assert(_owner == Self, "invariant");
    assert(_Responsible != Self, "invariant");
    return;
  }

  DeferredInitialize();

  if (TrySpin (Self) > 0) {
    assert(_owner == Self, "invariant");
    assert(_succ != Self, "invariant");
    assert(_Responsible != Self, "invariant");
    return;
  }

  ObjectWaiter node(Self);
  Self->_ParkEvent->reset();
  node._prev   = (ObjectWaiter *) 0xBAD;
  node.TState  = ObjectWaiter::TS_CXQ;

  ObjectWaiter * nxt;
  for (;;) { // 插入队首
    node._next = nxt = _cxq;
    if (Atomic::cmpxchg(&node, &_cxq, nxt) == nxt) break;

    // 再次尝试TATAS
    if (TryLock (Self) > 0) {
      assert(_succ != Self, "invariant");
      assert(_owner == Self, "invariant");
      assert(_Responsible != Self, "invariant");
      return;
    }
  }
    // 当EntryList为空时,说明没有正在等待的线程,直接将_Responsible设置为自己
  if ((SyncFlags & 16) == 0 && nxt == NULL && _EntryList == NULL) {
    // Try to assume the role of responsible thread for the monitor.
    // CONSIDER:  ST vs CAS vs { if (Responsible==null) Responsible=Self }
    Atomic::replace_if_null(Self, &_Responsible);
  }

  int nWakeups = 0;
  int recheckInterval = 1;

  for (;;) {

    if (TryLock(Self) > 0) break; // 再次尝试获取TATAS
    assert(_owner != Self, "invariant");

    if ((SyncFlags & 2) && _Responsible == NULL) {
      Atomic::replace_if_null(Self, &_Responsible);
    }

    // 获取锁失败，park自己，分为是不是_Responsible，如果是那么是timedpark，
    // 间隔一定时间后尝试获取_owner
    if (_Responsible == Self || (SyncFlags & 1)) {
      TEVENT(Inflated enter - park TIMED);
      Self->_ParkEvent->park((jlong) recheckInterval);
      // Increase the recheckInterval, but clamp the value.
      recheckInterval *= 8;
      if (recheckInterval > MAX_RECHECK_INTERVAL) {
        recheckInterval = MAX_RECHECK_INTERVAL;
      }
    } else {
      TEVENT(Inflated enter - park UNTIMED);
      Self->_ParkEvent->park();
    }

    if (TryLock(Self) > 0) break;

    ...

    ++nWakeups;

    if ((Knob_SpinAfterFutile & 1) && TrySpin(Self) > 0) break;

    ...

    if (_succ == Self) _succ = NULL; // 释放锁时，succ会被选取
  }


  // 从cxq或者entrylist上unlink自己
  assert(_owner == Self, "invariant");
  assert(object() != NULL, "invariant");
  // I'd like to write:
  //   guarantee (((oop)(object()))->mark() == markOopDesc::encode(this), "invariant") ;
  // but as we're at a safepoint that's not safe.

  UnlinkAfterAcquire(Self, &node);
  if (_succ == Self) _succ = NULL;

  ...

  return;
}

for (;;) {
    assert(THREAD == _owner, "invariant");

    if (Knob_ExitPolicy == 0) {// 默认为0
      OrderAccess::release_store(&_owner, (void*)NULL);   // 设置owner为null
      OrderAccess::storeload();                      
      if ((intptr_t(_EntryList)|intptr_t(_cxq)) == 0 || _succ != NULL) { // _succ也会在TrySpin中设置，如果存在的话就不用唤醒其他线程了
        TEVENT(Inflated exit - simple egress);
        return;
      }
      TEVENT(Inflated exit - complex egress);
      
      // 由于之后我们要操作cxq和entrylist, 所以需要重新获取锁,如果此处失败那么返回即可,由新的锁拥有者来处理唤醒线程
      if (!Atomic::replace_if_null(THREAD, &_owner)) {
        return;
      }
      TEVENT(Exit - Reacquired);
    } else {
      ...
    }

    ObjectWaiter * w = NULL;
    int QMode = Knob_QMode;

    // Qmode为2,那么直接从cxq中唤醒
    if (QMode == 2 && _cxq != NULL) {
      w = _cxq;
      assert(w != NULL, "invariant");
      assert(w->TState == ObjectWaiter::TS_CXQ, "Invariant");
      ExitEpilog(Self, w);
      return;
    }

    // QMode为3, 那么将cxq 抽干到EntryList中(放在EntryList末尾)
    if (QMode == 3 && _cxq != NULL) {
      w = _cxq;
      for (;;) {
        assert(w != NULL, "Invariant");
        ObjectWaiter * u = Atomic::cmpxchg((ObjectWaiter*)NULL, &_cxq, w); // 将cxq设置为null
        if (u == w) break;
        w = u;
      }
      assert(w != NULL, "invariant");

      ObjectWaiter * q = NULL;
      ObjectWaiter * p;
      for (p = w; p != NULL; p = p->_next) { // 更改状态为TS_ENTER
        guarantee(p->TState == ObjectWaiter::TS_CXQ, "Invariant");
        p->TState = ObjectWaiter::TS_ENTER;
        p->_prev = q;
        q = p;
      }

      // 将cxq放在EntryList末尾
      ObjectWaiter * Tail;
      for (Tail = _EntryList; Tail != NULL && Tail->_next != NULL;
           Tail = Tail->_next)
        /* empty */;
      if (Tail == NULL) {
        _EntryList = w;
      } else {
        Tail->_next = w;
        w->_prev = Tail;
      }

    }

    // / QMode为3, 那么将cxq 抽干到EntryList中(放在EntryList头部)
    if (QMode == 4 && _cxq != NULL) {
      w = _cxq;
      for (;;) {
        assert(w != NULL, "Invariant");
        ObjectWaiter * u = Atomic::cmpxchg((ObjectWaiter*)NULL, &_cxq, w);
        if (u == w) break;
        w = u;
      }
      assert(w != NULL, "invariant");

      ObjectWaiter * q = NULL;
      ObjectWaiter * p;
      for (p = w; p != NULL; p = p->_next) {
        guarantee(p->TState == ObjectWaiter::TS_CXQ, "Invariant");
        p->TState = ObjectWaiter::TS_ENTER;
        p->_prev = q;
        q = p;
      }

      // Prepend the RATs to the EntryList
      if (_EntryList != NULL) {
        q->_next = _EntryList;
        _EntryList->_prev = q;
      }
      _EntryList = w;

      // Fall thru into code that tries to wake a successor from EntryList
    }

    // _EntryList不为空,唤醒_EntryList的头部线程
    w = _EntryList;
    if (w != NULL) {
      assert(w->TState == ObjectWaiter::TS_ENTER, "invariant");
      ExitEpilog(Self, w);
      return;
    }

    // cxq和_EntryList都为空,那么重新将不空的cxq放在_EntryList中
    w = _cxq;
    if (w == NULL) continue;

    for (;;) {
      assert(w != NULL, "Invariant");
      ObjectWaiter * u = Atomic::cmpxchg((ObjectWaiter*)NULL, &_cxq, w);
      if (u == w) break;
      w = u;
    }

    // QMode为1, 将cxq逆序
    if (QMode == 1) {
      // QMode == 1 : drain cxq to EntryList, reversing order
      // We also reverse the order of the list.
      ObjectWaiter * s = NULL;
      ObjectWaiter * t = w;
      ObjectWaiter * u = NULL;
      while (t != NULL) {
        guarantee(t->TState == ObjectWaiter::TS_CXQ, "invariant");
        t->TState = ObjectWaiter::TS_ENTER;
        u = t->_next;
        t->_prev = u;
        t->_next = s;
        s = t;
        t = u;
      }
      _EntryList  = s;
      assert(s != NULL, "invariant");
    } else {
      // QMode == 0 or QMode == 2, 将cxq正序插入, 最晚的线程在最前面
      _EntryList = w;
      ObjectWaiter * q = NULL;
      ObjectWaiter * p;
      for (p = w; p != NULL; p = p->_next) {
        guarantee(p->TState == ObjectWaiter::TS_CXQ, "Invariant");
        p->TState = ObjectWaiter::TS_ENTER;
        p->_prev = q;
        q = p;
      }
    }

    // context-switch rate.
    if (_succ != NULL) continue;

    w = _EntryList;
    if (w != NULL) {
      guarantee(w->TState == ObjectWaiter::TS_ENTER, "invariant");
      ExitEpilog(Self, w);
      return;
    }
  }

void ObjectMonitor::ExitEpilog(Thread * Self, ObjectWaiter * Wakee) {
  assert(_owner == Self, "invariant");

  // Exit protocol:
  // 1. ST _succ = wakee
  // 2. membar #loadstore|#storestore;
  // 2. ST _owner = NULL
  // 3. unpark(wakee)

  _succ = Knob_SuccEnabled ? Wakee->_thread : NULL;
  ParkEvent * Trigger = Wakee->_event;

  // Hygiene -- once we've set _owner = NULL we can't safely dereference Wakee again.
  // The thread associated with Wakee may have grabbed the lock and "Wakee" may be
  // out-of-scope (non-extant).
  Wakee  = NULL;

  // Drop the lock
  OrderAccess::release_store(&_owner, (void*)NULL);
  OrderAccess::fence();                               // ST _owner vs LD in unpark()

  if (SafepointMechanism::poll(Self)) {
    TEVENT(unpark before SAFEPOINT);
  }

  DTRACE_MONITOR_PROBE(contended__exit, this, object(), Self);
  Trigger->unpark();

  // Maintain stats and report events to JVMTI
  OM_PERFDATA_OP(Parks, inc());
}

private final PoolChunkList<T> q050;
private final PoolChunkList<T> q025;
private final PoolChunkList<T> q000;
private final PoolChunkList<T> qInit;
private final PoolChunkList<T> q075;
private final PoolChunkList<T> q100;

threadCache = new PoolThreadLocalCache(useCacheForAllThreads); // 用于在线程本地分配内存
this.tinyCacheSize = tinyCacheSize;
this.smallCacheSize = smallCacheSize;
this.normalCacheSize = normalCacheSize;
chunkSize = validateAndCalculateChunkSize(pageSize, maxOrder);
...
if (nHeapArena > 0) {
    ...
}

if (nDirectArena > 0) {
    directArenas = newArenaArray(nDirectArena);
    List<PoolArenaMetric> metrics = new ArrayList<PoolArenaMetric>(directArenas.length);
    for (int i = 0; i < directArenas.length; i ++) {
        PoolArena.DirectArena arena = new PoolArena.DirectArena(
                this, pageSize, maxOrder, pageShifts, chunkSize, directMemoryCacheAlignment);
        directArenas[i] = arena;
        metrics.add(arena);
    }
    directArenaMetrics = Collections.unmodifiableList(metrics);
} else {
    directArenas = null;
    directArenaMetrics = Collections.emptyList();
}

protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
    PoolThreadCache cache = threadCache.get();
    PoolArena<ByteBuffer> directArena = cache.directArena;

    final ByteBuf buf;
    if (directArena != null) {
        buf = directArena.allocate(cache, initialCapacity, maxCapacity);
    } else {
        buf = PlatformDependent.hasUnsafe() ?
                UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) :
                new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
    }

    return toLeakAwareBuffer(buf);
}

protected synchronized PoolThreadCache initialValue() {
    final PoolArena<byte[]> heapArena = leastUsedArena(heapArenas);
    final PoolArena<ByteBuffer> directArena = leastUsedArena(directArenas);

    final Thread current = Thread.currentThread();
    if (useCacheForAllThreads || current instanceof FastThreadLocalThread) {
        final PoolThreadCache cache = new PoolThreadCache(
                heapArena, directArena, tinyCacheSize, smallCacheSize, normalCacheSize,
                DEFAULT_MAX_CACHED_BUFFER_CAPACITY, DEFAULT_CACHE_TRIM_INTERVAL);

        if (DEFAULT_CACHE_TRIM_INTERVAL_MILLIS > 0) {
            final EventExecutor executor = ThreadExecutorMap.currentExecutor();
            if (executor != null) {
                executor.scheduleAtFixedRate(trimTask, DEFAULT_CACHE_TRIM_INTERVAL_MILLIS,
                        DEFAULT_CACHE_TRIM_INTERVAL_MILLIS, TimeUnit.MILLISECONDS);
            }
        }
        return cache;
    }
    // No caching so just use 0 as sizes.
    return new PoolThreadCache(heapArena, directArena, 0, 0, 0, 0, 0);
}

private final int maxOrder; // chunk二叉树的高度，默认11
final int pageSize; // 单个page大小，默认8192
final int pageShifts;  //
final int chunkSize; // chunk大小
final int subpageOverflowMask; // 用于计算是否大于tiny/small
final int numSmallSubpagePools; // small请求的链表头个数
final int directMemoryCacheAlignment; // 对齐内存单位
final int directMemoryCacheAlignmentMask; // 用于对齐内存
private final PoolSubpage<T>[] tinySubpagePools; // 元素为tiny请求的Subpage双向链表头，下标表示subpage的大小，下同
private final PoolSubpage<T>[] smallSubpagePools;// 元素为small请求的Subpage双向链表头
// 各个利用率的chunkList
private final PoolChunkList<T> q050;
private final PoolChunkList<T> q025;
private final PoolChunkList<T> q000;
private final PoolChunkList<T> qInit;
private final PoolChunkList<T> q075;
private final PoolChunkList<T> q100;

this.parent = parent; // PooledByteBufAllocator
this.pageSize = pageSize;
this.maxOrder = maxOrder;
this.pageShifts = pageShifts;
this.chunkSize = chunkSize;
directMemoryCacheAlignment = cacheAlignment;
directMemoryCacheAlignmentMask = cacheAlignment - 1;
subpageOverflowMask = ~(pageSize - 1);
tinySubpagePools = newSubpagePoolArray(numTinySubpagePools);
for (int i = 0; i < tinySubpagePools.length; i ++) {
    tinySubpagePools[i] = newSubpagePoolHead(pageSize);
}

numSmallSubpagePools = pageShifts - 9;// pageShifts默认为13，则此处为4，即512，1k，2k，4k
smallSubpagePools = newSubpagePoolArray(numSmallSubpagePools);
for (int i = 0; i < smallSubpagePools.length; i ++) {
    smallSubpagePools[i] = newSubpagePoolHead(pageSize);
}

PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) {
    PooledByteBuf<T> buf = newByteBuf(maxCapacity);
    allocate(cache, buf, reqCapacity);
    return buf;
}

final int normCapacity = normalizeCapacity(reqCapacity); // 规范化请求内存大小
if (isTinyOrSmall(normCapacity)) { // 是否是tiny或者small, 即判断是否小于pageSize
    int tableIdx;
    PoolSubpage<T>[] table;
    boolean tiny = isTiny(normCapacity);
    if (tiny) { // < 512
        if (cache.allocateTiny(this, buf, reqCapacity, normCapacity)) {// 先从线程本地缓存中尝试分配, 如果成功直接返回, 如果没有可供分配的缓存, 那么尝试在Subpage数组中分配, 见如下分析.
            // 如果在线程本地分配成功, 那么直接返回
            return;
        }
        tableIdx = tinyIdx(normCapacity); // tinyIdx方法根据请求大小获得在tinySubpagePools中的数组索引, 即寻找具有合适的Subpage大小的Page链表
        table = tinySubpagePools;
    } else {
        if (cache.allocateSmall(this, buf, reqCapacity, normCapacity)) { // 同上, 只不过分配Small
            return;
        }
        tableIdx = smallIdx(normCapacity);
        table = smallSubpagePools;
    }

    final PoolSubpage<T> head = table[tableIdx]; // 获取链表头节点

    // 在SubpagePool中尝试分配
    // 需要在链表头节点上同步, 因为Chunk中的分配Subpage和free时也会修改链表
    synchronized (head) {
        final PoolSubpage<T> s = head.next;
        if (s != head) { // 如果链表中有除了头节点以外的节点, 即其他Chunk已经分配的Page
            assert s.doNotDestroy && s.elemSize == normCapacity;
            long handle = s.allocate();
            assert handle >= 0;
            s.chunk.initBufWithSubpage(buf, null, handle, reqCapacity);
            incTinySmallAllocation(tiny);
            return;
        }
    }

    // 
    synchronized (this) {
        allocateNormal(buf, reqCapacity, normCapacity);
    }

    incTinySmallAllocation(tiny);
    return;
}// normal大小的请求
if (normCapacity <= chunkSize) {
    if (cache.allocateNormal(this, buf, reqCapacity, normCapacity)) {// 尝试在线程本地缓存中分配
        // was able to allocate out of the cache so move on
        return;
    }
    synchronized (this) {
        allocateNormal(buf, reqCapacity, normCapacity);
        ++allocationsNormal;
    }
} else {
    // Huge allocations are never served via the cache so just call allocateHuge
    allocateHuge(buf, reqCapacity); // huge大小的分配不会缓存
}
}

boolean allocateTiny(PoolArena<?> area, PooledByteBuf<?> buf, int reqCapacity, int normCapacity) {
    return allocate(cacheForTiny(area, normCapacity), buf, reqCapacity);
}

private final MemoryRegionCache<byte[]>[] tinySubPageHeapCaches;
private final MemoryRegionCache<byte[]>[] smallSubPageHeapCaches;
private final MemoryRegionCache<ByteBuffer>[] tinySubPageDirectCaches;
private final MemoryRegionCache<ByteBuffer>[] smallSubPageDirectCaches;
private final MemoryRegionCache<byte[]>[] normalHeapCaches;
private final MemoryRegionCache<ByteBuffer>[] normalDirectCaches;

tinySubPageDirectCaches = createSubPageCaches(
        tinyCacheSize, PoolArena.numTinySubpagePools, SizeClass.Tiny);
smallSubPageDirectCaches = createSubPageCaches(
        smallCacheSize, directArena.numSmallSubpagePools, SizeClass.Small);

private static <T> MemoryRegionCache<T>[] createSubPageCaches(
        int cacheSize, int numCaches, SizeClass sizeClass) {
        MemoryRegionCache<T>[] cache = new MemoryRegionCache[numCaches];
        for (int i = 0; i < cache.length; i++) {
            cache[i] = new SubPageMemoryRegionCache<T>(cacheSize, sizeClass);
        }
        return cache;
}

private final int size; // 控制内部缓存个数
private final Queue<Entry<T>> queue; // 内部的缓存队列
private final SizeClass sizeClass; 
private int allocations; // 分配次数

MemoryRegionCache(int size, SizeClass sizeClass) {
    this.size = MathUtil.safeFindNextPositivePowerOfTwo(size); // 规范化
    queue = PlatformDependent.newFixedMpscQueue(this.size); // 构造线程安全的Mpsc队列
    this.sizeClass = sizeClass;
}

public final boolean add(PoolChunk<T> chunk, ByteBuffer nioBuffer, long handle) {
    Entry<T> entry = newEntry(chunk, nioBuffer, handle);
    boolean queued = queue.offer(entry);
    if (!queued) {
        entry.recycle(); // 队列满了, 直接回收完事
    }

    return queued;
}

static final class Entry<T> {
    final Handle<Entry<?>> recyclerHandle;
    PoolChunk<T> chunk;
    ByteBuffer nioBuffer;
    long handle = -1;

    Entry(Handle<Entry<?>> recyclerHandle) {
        this.recyclerHandle = recyclerHandle;
    }

    void recycle() {
        chunk = null;
        nioBuffer = null;
        handle = -1;
        recyclerHandle.recycle(this);
    }
}

public final boolean allocate(PooledByteBuf<T> buf, int reqCapacity) {
    Entry<T> entry = queue.poll();
    if (entry == null) {
        return false;
    }
    initBuf(entry.chunk, entry.nioBuffer, entry.handle, buf, reqCapacity);
    entry.recycle();
    ++ allocations;
    return true;
}

private MemoryRegionCache<?> cacheForTiny(PoolArena<?> area, int normCapacity) {
    int idx = PoolArena.tinyIdx(normCapacity);
    if (area.isDirect()) {
        return cache(tinySubPageDirectCaches, idx);
    }
    return cache(tinySubPageHeapCaches, idx);
}

/** PoolArena#tinyIdx **/
static int tinyIdx(int normCapacity) {
    return normCapacity >>> 4;
}

/** PoolThreadCache#cache **/
private static <T> MemoryRegionCache<T> cache(MemoryRegionCache<T>[] cache, int idx) {
    if (cache == null || idx > cache.length - 1) {
        return null;
    }
    return cache[idx];
}

private boolean allocate(MemoryRegionCache<?> cache, PooledByteBuf buf, int reqCapacity) {
    boolean allocated = cache.allocate(buf, reqCapacity);
    if (++ allocations >= freeSweepAllocationThreshold) {
        allocations = 0;
        trim();
    }
    return allocated;
}

tableIdx = tinyIdx(normCapacity);
table = tinySubpagePools;

// or

tableIdx = smallIdx(normCapacity);
table = smallSubpagePools;

final PoolSubpage<T> head = table[tableIdx];

synchronized (head) {
    final PoolSubpage<T> s = head.next;
    if (s != head) {
        assert s.doNotDestroy && s.elemSize == normCapacity;
        long handle = s.allocate();
        assert handle >= 0;
        s.chunk.initBufWithSubpage(buf, null, handle, reqCapacity);
        incTinySmallAllocation(tiny);
        return;
    }
}

private void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
    if (q050.allocate(buf, reqCapacity, normCapacity) || q025.allocate(buf, reqCapacity, normCapacity) ||
        q000.allocate(buf, reqCapacity, normCapacity) || qInit.allocate(buf, reqCapacity, normCapacity) ||
        q075.allocate(buf, reqCapacity, normCapacity)) {
        return;
    }

    // Add a new chunk.
    PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize);
    boolean success = c.allocate(buf, reqCapacity, normCapacity);
    assert success;
    qInit.add(c);
}

boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
    if (normCapacity > maxCapacity) {
        return false;
    }

    for (PoolChunk<T> cur = head; cur != null; cur = cur.next) {
        if (cur.allocate(buf, reqCapacity, normCapacity)) {
            if (cur.usage() >= maxUsage) {
                remove(cur);
                nextList.add(cur);
            }
            return true;
        }
    }
    return false;
}

final PoolArena<T> arena; // Chunk所在的Arena
final T memory; // 实际的内存，根据是不是direct的有区分，比如direct的是ByteBuffer，heap的是byte[]
final boolean unpooled; // false
final int offset; // 相对memory的偏移（因为要内存对齐，详见PoolArena#offsetCacheLine）
private final byte[] memoryMap; // 记录可分配信息
private final byte[] depthMap; // 存放每个深度信息
private final PoolSubpage<T>[] subpages; // 存放该Chunk的subpage
private final int subpageOverflowMask; // 判断是否大于pageSize的掩码
private final int pageSize; // 默认8K
private final int pageShifts; // 默认13
private final int maxOrder; // 11
private final int chunkSize; // 16MB
private final int log2ChunkSize; // 24
private final int maxSubpageAllocs;// 1 << maxOrder

PoolChunk(PoolArena<T> arena, T memory, int pageSize, int maxOrder, int pageShifts, int chunkSize, int offset) {
    unpooled = false;
    this.arena = arena;
    this.memory = memory;
    this.pageSize = pageSize;
    this.pageShifts = pageShifts;
    this.maxOrder = maxOrder;
    this.chunkSize = chunkSize;
    this.offset = offset;
    unusable = (byte) (maxOrder + 1); // 12
    log2ChunkSize = log2(chunkSize);
    subpageOverflowMask = ~(pageSize - 1);
    freeBytes = chunkSize;

    maxSubpageAllocs = 1 << maxOrder; // 2^11

    // Generate the memory map.
    memoryMap = new byte[maxSubpageAllocs << 1];// 一共2^12
    depthMap = new byte[memoryMap.length];// 一共2^12
    int memoryMapIndex = 1;
    for (int d = 0; d <= maxOrder; ++ d) {
        int depth = 1 << d;
        for (int p = 0; p < depth; ++ p) {
            memoryMap[memoryMapIndex] = (byte) d;// 全都初始化为对应的深度
            depthMap[memoryMapIndex] = (byte) d;
            memoryMapIndex ++;
        }
    }

    subpages = newSubpageArray(maxSubpageAllocs);
    cachedNioBuffers = new ArrayDeque<ByteBuffer>(8);
}

boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
    final long handle;
    if ((normCapacity & subpageOverflowMask) != 0) { // >= pageSize
        handle =  allocateRun(normCapacity);// 分配多个Page
    } else {
        handle = allocateSubpage(normCapacity);// 在一个Page内分配
    }

    if (handle < 0) {
        return false;
    }
    ByteBuffer nioBuffer = cachedNioBuffers != null ? cachedNioBuffers.pollLast() : null;
    initBuf(buf, nioBuffer, handle, reqCapacity);
    return true;
}

private long allocateSubpage(int normCapacity) {
    // 获取对应arena中的Subpage链表中的head
    PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity);
    int d = maxOrder; // d为要分配的深度, 此处是Subpage,故一定在叶节点, d = 11;
    synchronized (head) {
        int id = allocateNode(d);// 重要逻辑在这
        if (id < 0) {
            return id;
        }

        final PoolSubpage<T>[] subpages = this.subpages;
        final int pageSize = this.pageSize;

        freeBytes -= pageSize;

        int subpageIdx = subpageIdx(id);// 获取subpage在Chunk中的位置
        PoolSubpage<T> subpage = subpages[subpageIdx];
        if (subpage == null) {
            subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity);
            subpages[subpageIdx] = subpage;
        } else {
            subpage.init(head, normCapacity);
        }
        return subpage.allocate();
    }
}

private int allocateNode(int d) {
    int id = 1;
    int initial = - (1 << d); // 最后d位为0, 其他为1
    byte val = value(id); // 获取memoryMap的值, 这里id=1, 二叉树的root, 从root开始往下
    if (val > d) { // 当memoryMap的值大于某个值说明该节点下面没有足够的空间分配了
        return -1;
    }
    while (val < d || (id & initial) == 0) { // 只要当前id对应的节点还能分配(有余) || id还在d层的上面但节点无余(val==d) -> 继续向下
        id <<= 1; // 继续向下
        val = value(id);
        if (val > d) { // 如果左节点不行,那么尝试右节点(右节点一定可以, 因为已经通过了while条件的判断了)
            id ^= 1;
            val = value(id);
        }
    }
    byte value = value(id);
    assert value == d && (id & initial) == 1 << d : String.format("val = %d, id & initial = %d, d = %d",
            value, id & initial, d);
    setValue(id, unusable); // 将获取到的节点的memoryMap设置为unusable
    updateParentsAlloc(id); // 更新祖先节点
    return id;
}

/** Chunk#value **/
private byte value(int id) {
    return memoryMap[id];
}
/** Chunk#updateParentsAlloc **/
private void updateParentsAlloc(int id) {
    while (id > 1) { // 没到达root, root没有父节点
        int parentId = id >>> 1; // 父节点
        byte val1 = value(id);
        byte val2 = value(id ^ 1);
        byte val = val1 < val2 ? val1 : val2; // 获取自己和兄弟节点的value, 选取小的作为parent的value
        setValue(parentId, val);
        id = parentId;
    }
}

int id = allocateNode(d);// 我们这里返回了在二叉树中的索引
if (id < 0) {
    return id;
}

final PoolSubpage<T>[] subpages = this.subpages;
final int pageSize = this.pageSize;

freeBytes -= pageSize;

int subpageIdx = subpageIdx(id);// 获取subpage在Chunk中的位置
PoolSubpage<T> subpage = subpages[subpageIdx];
if (subpage == null) {
    subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity); // runOffset(id)获取的是在Chunk中的内存偏移量
    subpages[subpageIdx] = subpage;
} else {
    subpage.init(head, normCapacity);
}
return subpage.allocate();

/** subpageIdx **/
private int subpageIdx(int memoryMapIdx) {
    return memoryMapIdx ^ maxSubpageAllocs; // remove highest set bit, to get offset
}

final PoolChunk<T> chunk; // 所属chunk
private final int memoryMapIdx; // memoryMap数组的索引
private final int runOffset; // 在chunk中的内存偏移量
private final int pageSize; 
private final long[] bitmap; // 表示Subpage的分配信息，长度为pageSize / 16 / 64，64代表是long，16代表每个位16B(对应最小16B), 默认为8

PoolSubpage(PoolSubpage<T> head, PoolChunk<T> chunk, int memoryMapIdx, int runOffset, int pageSize, int elemSize) {
    this.chunk = chunk;
    this.memoryMapIdx = memoryMapIdx;
    this.runOffset = runOffset;
    this.pageSize = pageSize;
    bitmap = new long[pageSize >>> 10]; // pageSize / 16 / 64
    init(head, elemSize);
}

void init(PoolSubpage<T> head, int elemSize) {
    doNotDestroy = true;
    this.elemSize = elemSize;
    if (elemSize != 0) {
        maxNumElems = numAvail = pageSize / elemSize; // 可分配个数
        nextAvail = 0;
        bitmapLength = maxNumElems >>> 6; // 可分配个数除以64, 计算bitmap的长度
        if ((maxNumElems & 63) != 0) { // 此处补上一步整除的余数, 假设maxNumElems = 100, 那么上一步得出的结果为1, 此处再加一
            bitmapLength ++;
        }

        for (int i = 0; i < bitmapLength; i ++) {// 初始化bitmap
            bitmap[i] = 0;
        }
    }
    addToPool(head);
}

private void addToPool(PoolSubpage<T> head) { // 简单插入head的next中
    assert prev == null && next == null;
    prev = head;
    next = head.next;
    next.prev = this;
    head.next = this;
}

long allocate() {
    if (elemSize == 0) {
        return toHandle(0);
    }

    if (numAvail == 0 || !doNotDestroy) { // 分配大小过大或者doNotDestroy为false,返回分配失败
        return -1;
    }

    final int bitmapIdx = getNextAvail(); // 获取可用的bitmap索引,第一次分配时为0
    int q = bitmapIdx >>> 6; // 第几个long?
    int r = bitmapIdx & 63; // long中的第几个?
    assert (bitmap[q] >>> r & 1) == 0; // 一定没分配
    bitmap[q] |= 1L << r; // 把对应的位置为1

    if (-- numAvail == 0) {
        removeFromPool(); // 从链表中unlink
    }
    // 0x4000000000000000L | (long) bitmapIdx << 32 | memoryMapIdx;
    return toHandle(bitmapIdx);  // 即前32位放bitmap, 后32位放memoryMap的索引即Page在Chunk中的位置
}
/**---------------------------**/
private int getNextAvail() {
    int nextAvail = this.nextAvail;
    if (nextAvail >= 0) { // 初始时为0
        this.nextAvail = -1;
        return nextAvail;
    }
    return findNextAvail(); // 如果不是第一次分配
}
/**---------------------------**/
private int findNextAvail() {
    final long[] bitmap = this.bitmap;
    final int bitmapLength = this.bitmapLength;
    for (int i = 0; i < bitmapLength; i ++) { // 遍历bitmap,寻找没有被占用的空间
        long bits = bitmap[i];
        if (~bits != 0) { // bits的反不为0说明bits存在0, 即Subpage没有分配的位置
            return findNextAvail0(i, bits); // i为当前第几个long, bits为当前long
        }
    }
    return -1;
}
/**---------------------------**/
private int findNextAvail0(int i, long bits) {
    final int maxNumElems = this.maxNumElems;
    final int baseVal = i << 6; 

    for (int j = 0; j < 64; j ++) {// 遍历bits的每一位, 寻找0, 在分配时1是从右向左分配的
        if ((bits & 1) == 0) { // 找到0了
            int val = baseVal | j; 
            if (val < maxNumElems) {
                return val;
            } else {
                break;
            }
        }
        bits >>>= 1; // 这一位是1,继续向左寻找
    }
    return -1;
}

boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
    final long handle;
    if ((normCapacity & subpageOverflowMask) != 0) { // >= pageSize
        handle =  allocateRun(normCapacity);
    } else {
        handle = allocateSubpage(normCapacity); // 我们刚刚分析的是这个方法
    }

    if (handle < 0) {
        return false;
    }
    ByteBuffer nioBuffer = cachedNioBuffers != null ? cachedNioBuffers.pollLast() : null;
    initBuf(buf, nioBuffer, handle, reqCapacity); 
    return true;
}

private long allocateRun(int normCapacity) {
    int d = maxOrder - (log2(normCapacity) - pageShifts); // 找到第几层
    int id = allocateNode(d); // 在该层上分配, 获取id, id为memoryMap的索引
    if (id < 0) {
        return id;
    }
    freeBytes -= runLength(id); // freeBytes减去id节点分配的大小
    return id;
}

private int runLength(int id) {
    return 1 << log2ChunkSize - depth(id); // id代表的节点有多大
}

void initBuf(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int reqCapacity) {
    int memoryMapIdx = memoryMapIdx(handle);// 获取handle的低32位
    int bitmapIdx = bitmapIdx(handle);// 获取高32位
    if (bitmapIdx == 0) {
        byte val = value(memoryMapIdx); // memoryMap[memoryMapIdx]
        assert val == unusable : String.valueOf(val);
        buf.init(this, nioBuffer, handle, runOffset(memoryMapIdx) + offset,
                reqCapacity, runLength(memoryMapIdx), arena.parent.threadCache());
    } else {
      int memoryMapIdx = memoryMapIdx(handle);
        PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
        buf.init(
            this, nioBuffer, handle,
            runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize + offset,
                reqCapacity, subpage.elemSize, arena.parent.threadCache());
    }
}

private int runOffset(int id) {
    // represents the 0-based offset in #bytes from start of the byte-array chunk
    int shift = id ^ 1 << depth(id);
    return shift * runLength(id);
}

private int runLength(int id) {
    // represents the size in #bytes supported by node 'id' in the tree
    return 1 << log2ChunkSize - depth(id);
}

private void init0(PoolChunk<T> chunk, ByteBuffer nioBuffer,
                    long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
    assert handle >= 0;
    assert chunk != null;

    this.chunk = chunk;
    memory = chunk.memory;
    tmpNioBuf = nioBuffer;
    allocator = chunk.arena.parent;
    this.cache = cache;
    this.handle = handle;
    this.offset = offset;
    this.length = length;
    this.maxLength = maxLength;
}

boolean free(PoolSubpage<T> head, int bitmapIdx) {
    if (elemSize == 0) {
        return true;
    }
    int q = bitmapIdx >>> 6;
    int r = bitmapIdx & 63;
    assert (bitmap[q] >>> r & 1) != 0;
    bitmap[q] ^= 1L << r; // 将bitmap中的位置为0

    setNextAvail(bitmapIdx); // 设置当前bitmapIdx为nextAvail

    if (numAvail ++ == 0) {// 又可以用了， 加入到SubpagePool中
        addToPool(head);
        return true;
    }

    if (numAvail != maxNumElems) { // Subpage还存在被分配的
        return true;
    } else {
        if (prev == next) { // 链表至少要保证一个存在
            return true;
        }

        doNotDestroy = false; // 可以释放了
        removeFromPool();
        return false;
    }
}

void free(long handle, ByteBuffer nioBuffer) {
    int memoryMapIdx = memoryMapIdx(handle);
    int bitmapIdx = bitmapIdx(handle);

    if (bitmapIdx != 0) { // 释放Subpage
        PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
        assert subpage != null && subpage.doNotDestroy;

        PoolSubpage<T> head = arena.findSubpagePoolHead(subpage.elemSize);
        synchronized (head) {
            if (subpage.free(head, bitmapIdx & 0x3FFFFFFF)) { // Subpage还在使用直接返回否则释放Subpage并且从pool中移除
                return;
            }
        }
    }
    freeBytes += runLength(memoryMapIdx);
    setValue(memoryMapIdx, depth(memoryMapIdx)); // 还原memoryMap节点值
    updateParentsFree(memoryMapIdx); // 更新祖先节点memoryMap

    if (nioBuffer != null && cachedNioBuffers != null &&
            cachedNioBuffers.size() < PooledByteBufAllocator.DEFAULT_MAX_CACHED_BYTEBUFFERS_PER_CHUNK) {
        cachedNioBuffers.offer(nioBuffer);
    }
}

boolean free(PoolChunk<T> chunk, long handle, ByteBuffer nioBuffer) {
    chunk.free(handle, nioBuffer);
    if (chunk.usage() < minUsage) {
        remove(chunk);
        return move0(chunk);
    }
    return true;
}

private boolean move0(PoolChunk<T> chunk) {
    if (prevList == null) {
        // There is no previous PoolChunkList so return false which result in having the PoolChunk destroyed and
        // all memory associated with the PoolChunk will be released.
        assert chunk.usage() == 0;
        return false;
    }
    return prevList.move(chunk);
}

private boolean move(PoolChunk<T> chunk) {
    assert chunk.usage() < maxUsage;

    if (chunk.usage() < minUsage) {
        // Move the PoolChunk down the PoolChunkList linked-list.
        return move0(chunk);
    }

    // PoolChunk fits into this PoolChunkList, adding it here.
    add0(chunk);
    return true;
}

void freeChunk(PoolChunk<T> chunk, long handle, SizeClass sizeClass, ByteBuffer nioBuffer, boolean finalizer) {
    final boolean destroyChunk;
    ...
    if (destroyChunk) {
        // 调用unsafe相关方法释放Chunk内存
        destroyChunk(chunk);
    }
}

void free(PoolChunk<T> chunk, ByteBuffer nioBuffer, long handle, int normCapacity, PoolThreadCache cache) {
    if (chunk.unpooled) {
        ...
    } else {
        SizeClass sizeClass = sizeClass(normCapacity);
        if (cache != null && cache.add(this, chunk, nioBuffer, handle, normCapacity, sizeClass)) { 
          // 调用线程本地缓存.将对应Chunk中handle对应的内存缓存起来不释放
            return;
        }

        freeChunk(chunk, handle, sizeClass, nioBuffer, false);
    }
}

/** PoolThreadCache#add **/
boolean add(PoolArena<?> area, PoolChunk chunk, ByteBuffer nioBuffer,
            long handle, int normCapacity, SizeClass sizeClass) {
    MemoryRegionCache<?> cache = cache(area, normCapacity, sizeClass);
    if (cache == null) {
        return false;
    }
    return cache.add(chunk, nioBuffer, handle);
}

void init(Channel channel) throws Exception {
    // 省略部分代码
    ... 

    p.addLast(new ChannelInitializer<Channel>() {
    @Override
    public void initChannel(final Channel ch) throws Exception {
        final ChannelPipeline pipeline = ch.pipeline();
        ChannelHandler handler = config.handler();
        if (handler != null) {
            pipeline.addLast(handler);
        }

        ch.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                pipeline.addLast(new ServerBootstrapAcceptor(
                        ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
            }
        });
    }
});
}

// ServerBootstrapAcceptor#channelRead
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    final Channel child = (Channel) msg;
    // childHandler
    child.pipeline().addLast(childHandler);

    setChannelOptions(child, childOptions, logger);

    for (Entry<AttributeKey<?>, Object> e: childAttrs) {
        child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
    }

    try {
        childGroup.register(child).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    forceClose(child, future.cause());
                }
            }
        });
    } catch (Throwable t) {
        forceClose(child, t);
    }
}

// MultithreadEventLoopGroup#register
@Override
public ChannelFuture register(Channel channel) {
    return next().register(channel);
}

public ChannelFuture register(Channel channel) {
    return register(new DefaultChannelPromise(channel, this));
}

public ChannelFuture register(final ChannelPromise promise) {
    promise.channel().unsafe().register(this, promise);
    return promise;
}

private void register0(ChannelPromise promise) {
    try {
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return;
        }
        boolean firstRegistration = neverRegistered;
        doRegister();
        neverRegistered = false;
        registered = true;

        pipeline.invokeHandlerAddedIfNeeded();

        safeSetSuccess(promise);
        pipeline.fireChannelRegistered();
        if (isActive()) {
            if (firstRegistration) {
                pipeline.fireChannelActive();
            } else if (config().isAutoRead()) {
                beginRead();
            }
        }
    } catch (Throwable t) {
        closeForcibly();
        closeFuture.setClosed();
        safeSetFailure(promise, t);
    }
}

protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                eventLoop().selectNow();
                selected = true;
            } else {
                throw e;
            }
        }
    }
}

ServerBootstrap childHandler =
    this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
        // ...省略TCP socket配置
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline()
                    .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, handshakeHandler)
                    .addLast(defaultEventExecutorGroup,
                        encoder,
                        new NettyDecoder(), // 注意这里是请求解码器
                        new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                        connectionManageHandler,
                        serverHandler
                    );
            }
        });

public Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
    ByteBuf frame = null;
    try {
        frame = (ByteBuf) super.decode(ctx, in);

        ByteBuffer byteBuffer = frame.nioBuffer();

        return RemotingCommand.decode(byteBuffer);
    } catch (Exception e) {
        RemotingUtil.closeChannel(ctx.channel());
    } finally {
        if (null != frame) {
            frame.release();
        }
    }

    return null;
}

public static RemotingCommand decode(final ByteBuffer byteBuffer) {
    int length = byteBuffer.limit();
    int oriHeaderLen = byteBuffer.getInt();
    int headerLength = getHeaderLength(oriHeaderLen);

    byte[] headerData = new byte[headerLength];
    byteBuffer.get(headerData);

    RemotingCommand cmd = headerDecode(headerData, getProtocolType(oriHeaderLen));

    int bodyLength = length - 4 - headerLength;
    byte[] bodyData = null;
    if (bodyLength > 0) {
        bodyData = new byte[bodyLength];
        byteBuffer.get(bodyData);
    }
    cmd.body = bodyData;

    return cmd;
}

this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);
this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);

public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) {
    final RemotingCommand cmd = msg;
    if (cmd != null) {
        switch (cmd.getType()) {
            case REQUEST_COMMAND:
                processRequestCommand(ctx, cmd);
                break;
            case RESPONSE_COMMAND:
                processResponseCommand(ctx, cmd);
                break;
            default:
                break;
        }
    }
}

final Pair<NettyRequestProcessor, ExecutorService> pair = this.processorTable.get(cmd.getCode());
Runnable run = new Runnable() {
            @Override
            public void run() {
                final RemotingCommand response = pair.getObject1().processRequest(ctx, cmd);
                ctx.writeAndFlush(response);
        };
final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);
pair.getObject2().submit(requestTask);

private RemotingCommand sendMessage(final ChannelHandlerContext ctx, // 用于获取channel相关信息
                                        final RemotingCommand request, // 消息请求, 主要使用里面的body
                                        final SendMessageContext sendMessageContext, // 开启消息跟踪时的消息上下文信息
                                        final SendMessageRequestHeader requestHeader) { // 用于获取消息的相关信息,比如topic等
    // 检查topic是否可写, 如果topic可自动且topic不存在的情况下则创建topic, 否则返回TOPIC_NOT_EXIST异常
    // 进行queueId的范围校验等
    super.msgCheck(ctx, requestHeader, response);
    // MessageExtBrokerInner是消息在broker内部的表现形式, 其中不仅包含body也包含topic ,queueId等信息
    MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
    msgInner.setTopic(requestHeader.getTopic());
    msgInner.setQueueId(queueIdInt);
    msgInner.setBody(body);
    // 消费端消费失败时, 会将重试消息发到所属consumerGroup的retry topic
    // 这里handleRetryAndDLQ就是用来控制重试次数的, 如果重试次数超过最大重试次数时,
    // 会进入DLQ死信队列
    if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig)) {
        return response;
    }
    // 将消息存储到MessageStore中
    putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
    // 对存储结果进行处理 TODO
    return handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt);
}

long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
this.beginTimeInLock = beginLockTimestamp;

// Here settings are stored timestamp, in order to ensure an orderly
// global
msg.setStoreTimestamp(beginLockTimestamp);

if (null == mappedFile || mappedFile.isFull()) {
    mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
}
if (null == mappedFile) {
    log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
    beginTimeInLock = 0;
    return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
}

result = mappedFile.appendMessage(msg, this.appendMessageCallback);
switch (result.getStatus()) {
    case PUT_OK:
        break;
    case END_OF_FILE:
        unlockMappedFile = mappedFile;
        // Create a new file, re-write the message
        mappedFile = this.mappedFileQueue.getLastMappedFile(0);
        if (null == mappedFile) {
            // XXX: warn and notify me
            log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
            beginTimeInLock = 0;
            return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result);
        }
        result = mappedFile.appendMessage(msg, this.appendMessageCallback);
        break;
        ...
}

if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
    this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
}
// MappedFile.java#munlock
public void munlock() {
    final long beginTime = System.currentTimeMillis();
    final long address = ((DirectBuffer) (this.mappedByteBuffer)).address();
    Pointer pointer = new Pointer(address);
    int ret = LibC.INSTANCE.munlock(pointer, new NativeLong(this.fileSize));
    log.info("munlock {} {} {} ret = {} time consuming = {}", address, this.fileName, this.fileSize, ret, System.currentTimeMillis() - beginTime);
}

PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
return putMessageResult;

public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) {
    long createOffset = -1;
    MappedFile mappedFileLast = getLastMappedFile();

    if (mappedFileLast == null) {
        createOffset = startOffset - (startOffset % this.mappedFileSize);
    }

    if (mappedFileLast != null && mappedFileLast.isFull()) {
        createOffset = mappedFileLast.getFileFromOffset() + this.mappedFileSize;
    }

    if (createOffset != -1 && needCreate) {
        String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset);
        String nextNextFilePath = this.storePath + File.separator
            + UtilAll.offset2FileName(createOffset + this.mappedFileSize);
        MappedFile mappedFile = null;

        if (this.allocateMappedFileService != null) {
            mappedFile = this.allocateMappedFileService.putRequestAndReturnMappedFile(nextFilePath,
                nextNextFilePath, this.mappedFileSize);
        } else {
            try {
                mappedFile = new MappedFile(nextFilePath, this.mappedFileSize);
            } catch (IOException e) {
                log.error("create mappedFile exception", e);
            }
        }

        if (mappedFile != null) {
            if (this.mappedFiles.isEmpty()) {
                mappedFile.setFirstCreateInQueue(true);
            }
            this.mappedFiles.add(mappedFile);
        }

        return mappedFile;
    }

    return mappedFileLast;
}

MappedFile mappedFile;
if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
    try {
        mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
        mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
    } catch (RuntimeException e) {
        log.warn("Use default implementation.");
        mappedFile = new MappedFile(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
    }
} else {
    mappedFile = new MappedFile(req.getFilePath(), req.getFileSize());
}

public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb) {
    int currentPos = this.wrotePosition.get();

    if (currentPos < this.fileSize) {
        // 首先我们要确定消息首先写入哪里? 如果我们开启了TransientStorePoolEnable那么就需要写入到writeBuffer——这是我们之前
        // 从TransientStorePool中borrow的buffer, 否则就写入到文件映射buffer中
        ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
        byteBuffer.position(currentPos);
        AppendMessageResult result;
        if (messageExt instanceof MessageExtBrokerInner) {
            // 这里是消息append的核心实现
            result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBrokerInner) messageExt);
        } else if (messageExt instanceof MessageExtBatch) {
            result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt);
        } else {
            return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
        }
        this.wrotePosition.addAndGet(result.getWroteBytes());
        this.storeTimestamp = result.getStoreTimestamp();
        return result;
    }
    log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
    return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}

private final FlushCommitLogService flushCommitLogService;
private final FlushCommitLogService commitLogService = new CommitRealTimeService();

if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
    this.flushCommitLogService = new GroupCommitService();
} else {
    this.flushCommitLogService = new FlushRealTimeService();
}

public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
    // 同步刷盘, flushCommitLogService使用GroupCommitService实现, 调用GroupCommitRequest的waitForFlush方法
    // 同步等待刷盘
    if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
        if (messageExt.isWaitStoreMsgOK()) {
            GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
            service.putRequest(request);
            boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
            ...
        } else {
            service.wakeup();
        }
    }
    // 异步刷盘, 如果开启了TransientStorePoolEnable的话会先commit
    else {
        if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
            flushCommitLogService.wakeup();
        } else {
            commitLogService.wakeup();
        }
    }
}

boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages);

public boolean commit(final int commitLeastPages) {
    boolean result = true;
    MappedFile mappedFile = this.findMappedFileByOffset(this.committedWhere, this.committedWhere == 0);
    if (mappedFile != null) {
        int offset = mappedFile.commit(commitLeastPages);
        long where = mappedFile.getFileFromOffset() + offset;
        result = where == this.committedWhere;
        this.committedWhere = where;
    }

    return result;
}

public int commit(final int commitLeastPages) {
    if (writeBuffer == null) {
        // 如果writeBuffer为空说明没有开启TransientPool, 不需要commit, 直接返回即可
        return this.wrotePosition.get();
    }
    // isAbleToCommit方法用来判断当前脏数据是否满足commitLeastPages的需求
    if (this.isAbleToCommit(commitLeastPages)) {
        if (this.hold()) {
            commit0(commitLeastPages);
            this.release();
        } else {
            log.warn("in commit, hold failed, commit offset = " + this.committedPosition.get());
        }
    }

    // 当前writeBuffer的所有脏数据都写到FileChannel上了, 那么需要新建mappedFile, 重新向TransientPool申请一个
    // writeBuffer, 此时将writeBuffer归还给pool
    if (writeBuffer != null && this.transientStorePool != null && this.fileSize == this.committedPosition.get()) {
        this.transientStorePool.returnBuffer(writeBuffer);
        this.writeBuffer = null;
    }

    return this.committedPosition.get();
}

protected void commit0(final int commitLeastPages) {
    int writePos = this.wrotePosition.get();
    int lastCommittedPosition = this.committedPosition.get();

    if (writePos - this.committedPosition.get() > 0) {
        try {
            ByteBuffer byteBuffer = writeBuffer.slice();
            // 此处确实如我所料, 将上次commit的pos到wrotePos之间的数据写入到FileChannel中
            byteBuffer.position(lastCommittedPosition);
            byteBuffer.limit(writePos);
            // 这里使用FileChannel的write写入,实际上是
            // 系统调用, 只不过由于刷盘的条件以满足脏数据大小大于4页
            // 为什么要与读的mappedByteBuffer分离? 这是为了实现一种读写分离的机制:
            // 即读使用PageCache, 写先写入堆外内存, 然后再以一定间隔频率写入PageCache, 而不是
            // 每次刷盘都写入PageCache, 这样会造成PageCache繁忙(对PageCache具体造成什么影响暂时还不清楚)
            this.fileChannel.position(lastCommittedPosition);
            this.fileChannel.write(byteBuffer);
            this.committedPosition.set(writePos);
        } catch (Throwable e) {
            log.error("Error occurred when commit data to FileChannel.", e);
        }
    }
}

public int flush(final int flushLeastPages) {
    // 判断当前数据是否符合flushLeastPages的要求
    if (this.isAbleToFlush(flushLeastPages)) {
        if (this.hold()) {
            int value = getReadPosition();

            try {
                // 使用TransientPool是直接写入到FileChannel中的, 所以使用fileChannel的force方法
                if (writeBuffer != null || this.fileChannel.position() != 0) {
                    this.fileChannel.force(false);
                } else {
                    this.mappedByteBuffer.force();
                }
            } catch (Throwable e) {
                log.error("Error occurred when force data to disk.", e);
            }
            // 刷新flushedPos, 这里的value指的是当前commitPos, 
            // 如果使用TransientPool, 那么value就是commitPos
            // 否则就是当前的wrotePos
            this.flushedPosition.set(value);
            this.release();
        } else {
            log.warn("in flush, hold failed, flush offset = " + this.flushedPosition.get());
            this.flushedPosition.set(getReadPosition());
        }
    }
    return this.getFlushedPosition();
}

commitlog offset   +    size    +    tag hashcode
-----8bytes-----    ---4bytes--- ------8bytes-----

SelectMappedBufferResult result = DefaultMessageStore.this.commitLog.getData(reputFromOffset);
for (int readSize = 0; readSize < result.getSize() && doNext; ) {
// 构造DispatchRequest对象
 DispatchRequest dispatchRequest =
    DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false);
int size = dispatchRequest.getBufferSize() == -1 ? dispatchRequest.getMsgSize() : dispatchRequest.getBufferSize();

if (dispatchRequest.isSuccess()) {
    if (size > 0) {
        // 真正进行分发的地方
        DefaultMessageStore.this.doDispatch(dispatchRequest);
        ...
        this.reputFromOffset += size;
        readSize += size;
        ...
    } else if (size == 0) {
        // 还记得当commitlog的mappedFile最后空间不够了怎么办吗? 👆(在之前有相关解释)
        this.reputFromOffset = DefaultMessageStore.this.commitLog.rollNextFile(this.reputFromOffset);
        readSize = result.getSize();
    }
}
}

private final String topic;
private final int queueId;
private final long commitLogOffset;
private int msgSize;
private final long tagsCode;
private final long storeTimestamp;
private final long consumeQueueOffset;
private final String keys;
private final boolean success;
private final String uniqKey;

private final int sysFlag;
private final long preparedTransactionOffset;
private final Map<String, String> propertiesMap;
private byte[] bitMap;

public void doDispatch(DispatchRequest req) {
    for (CommitLogDispatcher dispatcher : this.dispatcherList) {
        dispatcher.dispatch(req);
    }
}

this.dispatcherList = new LinkedList<>();
this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());
this.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());

public void putMessagePositionInfo(DispatchRequest dispatchRequest) {
    ConsumeQueue cq = this.findConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId());
    cq.putMessagePositionInfoWrapper(dispatchRequest);
}

this.byteBufferIndex.flip();
this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE);
this.byteBufferIndex.putLong(offset);
this.byteBufferIndex.putInt(size);
this.byteBufferIndex.putLong(tagsCode);

MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(expectLogicOffset);
this.maxPhysicOffset = offset + size;
return mappedFile.appendMessage(this.byteBufferIndex.array());

int keyHash = indexKeyHashMethod(key);
int slotPos = keyHash % this.hashSlotNum;
int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;

int absIndexPos =
      IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
      + this.indexHeader.getIndexCount() * indexSize;

// hash
this.mappedByteBuffer.putInt(absIndexPos, keyHash);
// 物理位移
this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
// 与beginTimestamp相对的时间位移
this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
// 记录“链表”的上一个indexItem的位置
this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);

// 更新hashslot的值, 将其值设置为本次插入的位置(位置指第几个indexItem)
this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());

int keyHash = indexKeyHashMethod(key);
int slotPos = keyHash % this.hashSlotNum;
int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;

int absIndexPos =
      IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
            + nextIndexToRead * indexSize;

int keyHashRead = this.mappedByteBuffer.getInt(absIndexPos);
long phyOffsetRead = this.mappedByteBuffer.getLong(absIndexPos + 4);

long timeDiff = (long) this.mappedByteBuffer.getInt(absIndexPos + 4 + 8);
int prevIndexRead = this.mappedByteBuffer.getInt(absIndexPos + 4 + 8 + 4);

if (timeDiff < 0) {
      break;
}

timeDiff *= 1000L;

long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff;
boolean timeMatched = (timeRead >= begin) && (timeRead <= end);

if (keyHash == keyHashRead && timeMatched) {
      phyOffsets.add(phyOffsetRead);
}

if (prevIndexRead <= invalidIndex
      || prevIndexRead > this.indexHeader.getIndexCount()
      || prevIndexRead == nextIndexToRead || timeRead < begin) {
      break;
}

yum remove -y docker \
docker-client \
docker-client-latest \
docker-common \
docker-latest \
docker-latest-logrotate \
docker-logrotate \
docker-selinux \
docker-engine-selinux \
docker-engine

yum install -y yum-utils \
device-mapper-persistent-data \
lvm2
yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo

yum install -y docker-ce-18.09.7 docker-ce-cli-18.09.7 containerd.io
systemctl enable docker
systemctl start docker

yum install -y nfs-utils

systemctl stop firewalld
systemctl disable firewalld

setenforce 0
sed -i "s/SELINUX=enforcing/SELINUX=disabled/g" /etc/selinux/config

swapoff -a
yes | cp /etc/fstab /etc/fstab_bak
cat /etc/fstab_bak |grep -v swap > /etc/fstab

sed -i "s#^net.ipv4.ip_forward.*#net.ipv4.ip_forward=1#g"  /etc/sysctl.conf
sed -i "s#^net.bridge.bridge-nf-call-ip6tables.*#net.bridge.bridge-nf-call-ip6tables=1#g"  /etc/sysctl.conf
sed -i "s#^net.bridge.bridge-nf-call-iptables.*#net.bridge.bridge-nf-call-iptables=1#g"  /etc/sysctl.conf

echo "net.ipv4.ip_forward = 1" >> /etc/sysctl.conf
echo "net.bridge.bridge-nf-call-ip6tables = 1" >> /etc/sysctl.conf
echo "net.bridge.bridge-nf-call-iptables = 1" >> /etc/sysctl.conf

sysctl -p

cat <<EOF > /etc/yum.repos.d/kubernetes.repo
[kubernetes]
name=Kubernetes
baseurl=http://mirrors.aliyun.com/kubernetes/yum/repos/kubernetes-el7-x86_64
enabled=1
gpgcheck=0
repo_gpgcheck=0
gpgkey=http://mirrors.aliyun.com/kubernetes/yum/doc/yum-key.gpg
       http://mirrors.aliyun.com/kubernetes/yum/doc/rpm-package-key.gpg
EOF

yum install -y kubelet-1.15.3 kubeadm-1.15.3 kubectl-1.15.3

# 修改docker Cgroup Driver为systemd
# # 将/usr/lib/systemd/system/docker.service文件中的这一行 ExecStart=/usr/bin/dockerd -H fd:// --containerd=/run/containerd/containerd.sock
# # 修改为 ExecStart=/usr/bin/dockerd -H fd:// --containerd=/run/containerd/containerd.sock --exec-opt native.cgroupdriver=systemd
# 如果不修改，在添加 worker 节点时可能会碰到如下错误
# [WARNING IsDockerSystemdCheck]: detected "cgroupfs" as the Docker cgroup driver. The recommended driver is "systemd". 
# Please follow the guide at https://kubernetes.io/docs/setup/cri/
sed -i "s#^ExecStart=/usr/bin/dockerd.*#ExecStart=/usr/bin/dockerd -H fd:// --containerd=/run/containerd/containerd.sock --exec-opt native.cgroupdriver=systemd#g" /usr/lib/systemd/system/docker.service

# 设置 docker 镜像，提高 docker 镜像下载速度和稳定性
# 如果您访问 https://hub.docker.io 速度非常稳定，亦可以跳过这个步骤
curl -sSL https://get.daocloud.io/daotools/set_mirror.sh | sh -s http://f1361db2.m.daocloud.io

# 重启 docker，并启动 kubelet
systemctl daemon-reload
systemctl restart docker
systemctl enable kubelet && systemctl start kubelet

docker version
2. 使用kubeadm在master机器上初始化集群
使用kubeadm初始化集群  
apiVersion: kubeadm.k8s.io/v1beta2
kind: ClusterConfiguration
kubernetesVersion: v1.15.3
imageRepository: registry.cn-hangzhou.aliyuncs.com/google_containers
controlPlaneEndpoint: "172.31.129.129:6443"
networking:
  serviceSubnet: "10.96.0.0/12"
  podSubnet: "10.244.0.0/16"
  dnsDomain: "cluster.local"
其中要注意：

controlPlaneEndpoint填写master的ip。
podSubnet不能和master节点/worker节点重叠。（podSubnet之后还要使用）
初始化结束后，在master上执行如下命令，让kubectl可用：
配置kubectl  
mkdir -p $HOME/.kube
sudo cp -i /etc/kubernetes/admin.conf $HOME/.kube/config
sudo chown $(id -u):$(id -g) $HOME/.kube/config
配置好kubectl后，执行：
kubectl get pod -n kube-system
查看master上pod的状态，正常应该是全部为Running。
使用kubectl get node查看机器节点的状况：
 
目前只有一个节点即master节点。
3. 配置Pod网络
我们使用calico部署k8s集群网络。
先获取calico.yaml: wget https://docs.projectcalico.org/v3.8/manifests/calico.yaml
执行命令： sed -i "s#192\.168\.0\.0/16#${POD_SUBNET}#" calico.yaml
其中${POD_SUBNET}替换为上面配置master时的podSubnet。
然后执行：kubectl apply -f calico.yaml

4. 将worker节点加入集群
worker节点安装好必要的软件之后，我们就可以开始加入到集群中了。
首先获取加入到集群中的命令，在master上执行：kubeadm token create --print-join-command
得到类似如下的命令：
配置kubectl  
kubeadm join 172.31.129.129:6443 --token 5h5stg.rmjosu2a3wbhz2ql     --discovery-token-ca-cert-hash sha256:ad066deabfd7c55482f8d02f3bc35bdaf95060fca021ec6b68f551009195a582
在worker上执行即可。
然后在master上执行kubectl get node命令观察节点的状态是否都是Ready。
之后再执行get pod命令观察calico是否是Running状态，如果出现CrashBackoff状态那么可能是calico获取的网络接口不对，
整calicao 网络插件的网卡发现机制，修改IP_AUTODETECTION_METHOD对应的value值。官方提供的yaml文件中，ip识别策略（IPDETECTMETHOD）没有配置，即默认为first-found，这会导致一个网络异常的ip作为nodeIP被注册，从而影响node-to-node mesh。我们可以修改成can-reach或者interface的策略，尝试连接某一个Ready的node的IP，以此选择出正确的IP。
配置kubectl  
// calico.yaml 文件添加以下二行
- name: IP_AUTODETECTION_METHOD
value: "interface=eth.*"  #  根据实际网卡开头配置
 
 // 配置如下             
- name: CLUSTER_TYPE
  value: "k8s,bgp"
# Auto-detect the BGP IP address.
- name: IP
  value: "autodetect"
- name: IP_AUTODETECTION_METHOD
  value: "interface=ens.*"
# Enable IPIP
- name: CALICO_IPV4POOL_IPIP
  value: "Always"
Optional可选步骤：
安装dashboard
首先获取dashboard的yaml文件，然后对其进行改动使其可以由外部根据端口直接访问，否则还需要启动一个kube-proxy进行代理比较麻烦。
改动如下：
配置kubectl  
ports:
- containerPort: 8443
protocol: TCP
name: https
args:
- --auto-generate-certificates
- --token-ttl=43200


...


kind: Service
apiVersion: v1
metadata:
  labels:
    k8s-app: kubernetes-dashboard
  name: kubernetes-dashboard
  namespace: kube-system
spec:
  ports:
    - port: 443
      targetPort: 8443
      nodePort: 30001
      name: https
  type: NodePort
  selector:
    k8s-app: kubernetes-dashboard

apiVersion: v1
kind: ServiceAccount
metadata:
name: admin-user
namespace: kube-system
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: admin-user
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: ClusterRole
name: cluster-admin
subjects:
- kind: ServiceAccount
name: admin-user
namespace: kube-system

if (System.currentTimeMillis() < nextTimeToRequestVote && !needIncreaseTermImmediately) {
    return;
}

long term;
long ledgerEndTerm;
long ledgerEndIndex;

synchronized (memberState) {
    if (!memberState.isCandidate()) {
        return;
    }
    if (lastParseResult == VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT || needIncreaseTermImmediately) {
        long prevTerm = memberState.currTerm();
        term = memberState.nextTerm();
        logger.info("{}_[INCREASE_TERM] from {} to {}", memberState.getSelfId(), prevTerm, term);
        lastParseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
    } else {
        term = memberState.currTerm();
    }
    ledgerEndIndex = memberState.getLedgerEndIndex();
    ledgerEndTerm = memberState.getLedgerEndTerm();
}

long startVoteTimeMs = System.currentTimeMillis();
final List<CompletableFuture<VoteResponse>> quorumVoteResponses = voteForQuorumResponses(term, ledgerEndTerm, ledgerEndIndex);

private List<CompletableFuture<VoteResponse>> voteForQuorumResponses(long term, long ledgerEndTerm,
    long ledgerEndIndex) throws Exception {
    List<CompletableFuture<VoteResponse>> responses = new ArrayList<>();
    for (String id : memberState.getPeerMap().keySet()) {
        VoteRequest voteRequest = new VoteRequest();
        voteRequest.setGroup(memberState.getGroup());
        voteRequest.setLedgerEndIndex(ledgerEndIndex);
        voteRequest.setLedgerEndTerm(ledgerEndTerm);
        voteRequest.setLeaderId(memberState.getSelfId());
        voteRequest.setTerm(term);
        voteRequest.setRemoteId(id);
        CompletableFuture<VoteResponse> voteResponse;
        if (memberState.getSelfId().equals(id)) {
            voteResponse = handleVote(voteRequest, true);
        } else {
            //async
            voteResponse = dLedgerRpcService.vote(voteRequest);
        }
        responses.add(voteResponse);

    }
    return responses;
}

if (request.getTerm() < memberState.currTerm()) {
    return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_EXPIRED_VOTE_TERM));
} else if (request.getTerm() == memberState.currTerm()) {
    if (memberState.currVoteFor() == null) {
        //let it go
    } else if (memberState.currVoteFor().equals(request.getLeaderId())) {
        //repeat just let it go
    } else {
        if (memberState.getLeaderId() != null) {
            return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_ALREADY__HAS_LEADER));
        } else {
            return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_ALREADY_VOTED));
        }
    }
} else {
    //stepped down by larger term
    changeRoleToCandidate(request.getTerm());
    needIncreaseTermImmediately = true;
    //only can handleVote when the term is consistent
    return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_TERM_NOT_READY));
}

if (request.getLedgerEndTerm() < memberState.getLedgerEndTerm()) {
    return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_EXPIRED_LEDGER_TERM));
} else if (request.getLedgerEndTerm() == memberState.getLedgerEndTerm() && request.getLedgerEndIndex() < memberState.getLedgerEndIndex()) {
    return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_SMALL_LEDGER_END_INDEX));
}

if (request.getTerm() < memberState.getLedgerEndTerm()) {
    return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.getLedgerEndTerm()).voteResult(VoteResponse.RESULT.REJECT_TERM_SMALL_THAN_LEDGER));
}

memberState.setCurrVoteFor(request.getLeaderId());
return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.ACCEPT);

CountDownLatch voteLatch = new CountDownLatch(1);
        for (CompletableFuture<VoteResponse> future : quorumVoteResponses) {
    // ...
}

public enum RESULT {
    UNKNOWN,
    ACCEPT, // 接受投票
    REJECT_UNKNOWN_LEADER, // 发起投票的机器不在集群中
    REJECT_UNEXPECTED_LEADER, // 
    REJECT_EXPIRED_VOTE_TERM, // 请求的term过期
    REJECT_ALREADY_VOTED, // 已投票给某个节点
    REJECT_ALREADY__HAS_LEADER, // 已经存在有效的leader，且该leader是其他机器
    REJECT_TERM_NOT_READY, // 本节点term过小（接下来转为Candidate）
    REJECT_TERM_SMALL_THAN_LEDGER, // 请求的term小于本机log的term
    REJECT_EXPIRED_LEDGER_TERM, // 请求log的term小于本机log的term
    REJECT_SMALL_LEDGER_END_INDEX;// 请求log的index小于本机log的index
}

switch (x.getVoteResult()) {
    case ACCEPT:
        acceptedNum.incrementAndGet();
        break;
    case REJECT_ALREADY_VOTED:
        break;
    case REJECT_ALREADY__HAS_LEADER:
        alreadyHasLeader.compareAndSet(false, true);
        break;
    case REJECT_TERM_SMALL_THAN_LEDGER:
    case REJECT_EXPIRED_VOTE_TERM:
        if (x.getTerm() > knownMaxTermInGroup.get()) {
            knownMaxTermInGroup.set(x.getTerm());
        }
        break;
    case REJECT_EXPIRED_LEDGER_TERM:
    case REJECT_SMALL_LEDGER_END_INDEX:
        biggerLedgerNum.incrementAndGet();
        break;
    case REJECT_TERM_NOT_READY:
        notReadyTermNum.incrementAndGet();
        break;
    default:
        break;

}

if (knownMaxTermInGroup.get() > term) {
    parseResult = VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT;
    nextTimeToRequestVote = getNextTimeToRequestVote();
    changeRoleToCandidate(knownMaxTermInGroup.get());
}

else if (alreadyHasLeader.get()) {
    parseResult = VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT;
    nextTimeToRequestVote = getNextTimeToRequestVote() + heartBeatTimeIntervalMs * maxHeartBeatLeak;
}

else if (!memberState.isQuorum(validNum.get())) {
    parseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
    nextTimeToRequestVote = getNextTimeToRequestVote();
}

else if (memberState.isQuorum(acceptedNum.get())) {
    parseResult = VoteResponse.ParseResult.PASSED;
} 
if (parseResult == VoteResponse.ParseResult.PASSED) {
    logger.info("[{}] [VOTE_RESULT] has been elected to be the leader in term {}", memberState.getSelfId(), term);
    changeRoleToLeader(term);
}

else if (memberState.isQuorum(acceptedNum.get() + notReadyTermNum.get())) {
    parseResult = VoteResponse.ParseResult.REVOTE_IMMEDIATELY;
}

else if (memberState.isQuorum(acceptedNum.get() + biggerLedgerNum.get())) {
    parseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
    nextTimeToRequestVote = getNextTimeToRequestVote();
}

else {
    parseResult = VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT;
    nextTimeToRequestVote = getNextTimeToRequestVote();
}

public synchronized void changeToLeader(long term) {
    PreConditions.check(currTerm == term, DLedgerResponseCode.ILLEGAL_MEMBER_STATE, "%d != %d", currTerm, term);
    this.role = LEADER;
    this.leaderId = selfId;
}

private void maintainAsLeader() throws Exception {
        if (DLedgerUtils.elapsed(lastSendHeartBeatTime) > heartBeatTimeIntervalMs) {
            long term;
            String leaderId;
            synchronized (memberState) {
                if (!memberState.isLeader()) {
                    //stop sending
                    return;
                }
                term = memberState.currTerm();
                leaderId = memberState.getLeaderId();
                lastSendHeartBeatTime = System.currentTimeMillis();
            }
            sendHeartbeats(term, leaderId); // 发送心跳
        }
    }

for (String id : memberState.getPeerMap().keySet()) {
    if (memberState.getSelfId().equals(id)) {
        continue;
    }
    /**
    构造请求, 并且使用dLedgerRpcService发送请求.
    */
    HeartBeatRequest heartBeatRequest = new HeartBeatRequest();
    heartBeatRequest.setGroup(memberState.getGroup());
    heartBeatRequest.setLocalId(memberState.getSelfId());
    heartBeatRequest.setRemoteId(id);
    heartBeatRequest.setLeaderId(leaderId);
    heartBeatRequest.setTerm(term);
    CompletableFuture<HeartBeatResponse> future = dLedgerRpcService.heartBeat(heartBeatRequest);
    future.whenComplete((HeartBeatResponse x, Throwable ex) -> {
        try {

            if (ex != null) {
                throw ex;
            }
            switch (DLedgerResponseCode.valueOf(x.getCode())) {
                case SUCCESS:
                    succNum.incrementAndGet(); // 成功, 增加succNum
                    break;
                case EXPIRED_TERM:
                    maxTerm.set(x.getTerm()); // 存在更大的term
                    break;
                case INCONSISTENT_LEADER:
                    inconsistLeader.compareAndSet(false, true); // 不一致的leader
                    break;
                case TERM_NOT_READY:
                    notReadyNum.incrementAndGet(); // term比leader小
                    break;
                default:
                    break;
            }
            if (memberState.isQuorum(succNum.get())
                || memberState.isQuorum(succNum.get() + notReadyNum.get())) {
                beatLatch.countDown();
            }
        } catch (Throwable t) {
            logger.error("Parse heartbeat response failed", t);
        } finally {
            allNum.incrementAndGet();
            if (allNum.get() == memberState.peerSize()) {
                beatLatch.countDown();
            }
        }
    });
}

if (memberState.isQuorum(succNum.get())) {
    lastSuccHeartBeatTime = System.currentTimeMillis();
} else {
    if (memberState.isQuorum(succNum.get() + notReadyNum.get())) {
        lastSendHeartBeatTime = -1;
    } else if (maxTerm.get() > term) {
        changeRoleToCandidate(maxTerm.get());
    } else if (inconsistLeader.get()) {
        changeRoleToCandidate(term);
    } else if (DLedgerUtils.elapsed(lastSuccHeartBeatTime) > maxHeartBeatLeak * heartBeatTimeIntervalMs) {
        changeRoleToCandidate(term);
    }
}

if (request.getTerm() < memberState.currTerm()) {
    return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.EXPIRED_TERM.getCode()));
} else if (request.getTerm() == memberState.currTerm()) {
    if (request.getLeaderId().equals(memberState.getLeaderId())) {
        lastLeaderHeartBeatTime = System.currentTimeMillis();
        return CompletableFuture.completedFuture(new HeartBeatResponse());
    }
}

synchronized (memberState) {
    if (request.getTerm() < memberState.currTerm()) {
        return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.EXPIRED_TERM.getCode()));
    } else if (request.getTerm() == memberState.currTerm()) {
        if (memberState.getLeaderId() == null) {
            changeRoleToFollower(request.getTerm(), request.getLeaderId());
            return CompletableFuture.completedFuture(new HeartBeatResponse());
        } else if (request.getLeaderId().equals(memberState.getLeaderId())) {
            lastLeaderHeartBeatTime = System.currentTimeMillis();
            return CompletableFuture.completedFuture(new HeartBeatResponse());
        } else {
            //this should not happen, but if happened
            logger.error("[{}][BUG] currTerm {} has leader {}, but received leader {}", memberState.getSelfId(), memberState.currTerm(), memberState.getLeaderId(), request.getLeaderId());
            return CompletableFuture.completedFuture(new HeartBeatResponse().code(DLedgerResponseCode.INCONSISTENT_LEADER.getCode()));
        }
    } else {
        //To make it simple, for larger term, do not change to follower immediately
        //first change to candidate, and notify the state-maintainer thread
        changeRoleToCandidate(request.getTerm());
        needIncreaseTermImmediately = true;
        return CompletableFuture.completedFuture(new HeartBeatResponse().code(DLedgerResponseCode.TERM_NOT_READY.getCode()));
    }
}

private void maintainAsFollower() {
    if (DLedgerUtils.elapsed(lastLeaderHeartBeatTime) > 2 * heartBeatTimeIntervalMs) {
        synchronized (memberState) {
            if (memberState.isFollower() && (DLedgerUtils.elapsed(lastLeaderHeartBeatTime) > maxHeartBeatLeak * heartBeatTimeIntervalMs)) {
                logger.info("[{}][HeartBeatTimeOut] lastLeaderHeartBeatTime: {} heartBeatTimeIntervalMs: {} lastLeader={}", memberState.getSelfId(), new Timestamp(lastLeaderHeartBeatTime), heartBeatTimeIntervalMs, memberState.getLeaderId());
                changeRoleToCandidate(memberState.currTerm());
            }
        }
    }
}

long currTerm = memberState.currTerm();
if (dLedgerEntryPusher.isPendingFull(currTerm)) { //        1
    AppendEntryResponse appendEntryResponse = new AppendEntryResponse();
    appendEntryResponse.setGroup(memberState.getGroup());
    appendEntryResponse.setCode(DLedgerResponseCode.LEADER_PENDING_FULL.getCode());
    appendEntryResponse.setTerm(currTerm);
    appendEntryResponse.setLeaderId(memberState.getSelfId());
    return AppendFuture.newCompletedFuture(-1, appendEntryResponse);
} else {                                           //       2
    DLedgerEntry dLedgerEntry = new DLedgerEntry();
    dLedgerEntry.setBody(request.getBody());
    DLedgerEntry resEntry = dLedgerStore.appendAsLeader(dLedgerEntry);
    return dLedgerEntryPusher.waitAck(resEntry);
}

public DLedgerEntry appendAsLeader(DLedgerEntry entry) {
    ByteBuffer dataBuffer = localEntryBuffer.get();
    ByteBuffer indexBuffer = localIndexBuffer.get();
    DLedgerEntryCoder.encode(entry, dataBuffer);
    int entrySize = dataBuffer.remaining();
    synchronized (memberState) {
        long nextIndex = ledgerEndIndex + 1;
        entry.setIndex(nextIndex);
        entry.setTerm(memberState.currTerm());
        entry.setMagic(CURRENT_MAGIC);
        DLedgerEntryCoder.setIndexTerm(dataBuffer, nextIndex, memberState.currTerm(), CURRENT_MAGIC);
        long prePos = dataFileList.preAppend(dataBuffer.remaining()); // @1
        entry.setPos(prePos);
        DLedgerEntryCoder.setPos(dataBuffer, prePos);
        long dataPos = dataFileList.append(dataBuffer.array(), 0, dataBuffer.remaining()); // @2
        DLedgerEntryCoder.encodeIndex(dataPos, entrySize, CURRENT_MAGIC, nextIndex, memberState.currTerm(), indexBuffer);
        long indexPos = indexFileList.append(indexBuffer.array(), 0, indexBuffer.remaining(), false);
        ledgerEndIndex++;
        ledgerEndTerm = memberState.currTerm();
        if (ledgerBeginIndex == -1) {
            ledgerBeginIndex = ledgerEndIndex;
        }
        updateLedgerEndIndexAndTerm();
        return entry;
    }
}

long prePos = dataFileList.preAppend(dataBuffer.remaining());

public long preAppend(int len, boolean useBlank) {
    MmapFile mappedFile = getLastMappedFile(); // 1
    if (null == mappedFile || mappedFile.isFull()) {
        mappedFile = getLastMappedFile(0);
    }
    int blank = useBlank ? MIN_BLANK_LEN : 0;
    if (len + blank > mappedFile.getFileSize() - mappedFile.getWrotePosition()) { // 2
        ByteBuffer byteBuffer = ByteBuffer.allocate(mappedFile.getFileSize() - mappedFile.getWrotePosition());
        byteBuffer.putInt(BLANK_MAGIC_CODE);
        byteBuffer.putInt(mappedFile.getFileSize() - mappedFile.getWrotePosition());
        if (mappedFile.appendMessage(byteBuffer.array())) {
            //need to set the wrote position
            mappedFile.setWrotePosition(mappedFile.getFileSize());
        } else {
            logger.error("Append blank error for {}", storePath);
            return -1;
        }
        mappedFile = getLastMappedFile(0);
        if (null == mappedFile) {
            logger.error("Create mapped file for {}", storePath);
            return -1;
        }
    }
    return mappedFile.getFileFromOffset() + mappedFile.getWrotePosition(); // 3

}

public CompletableFuture<AppendEntryResponse> waitAck(DLedgerEntry entry) {
    updatePeerWaterMark(entry.getTerm(), memberState.getSelfId(), entry.getIndex()); // 1. 更新水位线
    if (memberState.getPeerMap().size() == 1) { // 2. 只有一个节点
        AppendEntryResponse response = new AppendEntryResponse();
        response.setGroup(memberState.getGroup());
        response.setLeaderId(memberState.getSelfId());
        response.setIndex(entry.getIndex());
        response.setTerm(entry.getTerm());
        response.setPos(entry.getPos());
        return AppendFuture.newCompletedFuture(entry.getPos(), response);
    } else {
        checkTermForPendingMap(entry.getTerm(), "waitAck");
        AppendFuture<AppendEntryResponse> future = new AppendFuture<>(dLedgerConfig.getMaxWaitAckTimeMs());
        future.setPos(entry.getPos());
        wakeUpDispatchers(); // 唤醒dispatcher, 方法日志
        return future;
    }
}

private Map<Long, ConcurrentMap<String, Long>> peerWaterMarksByTerm = new ConcurrentHashMap<>();
private Map<Long, ConcurrentMap<Long, TimeoutFuture<AppendEntryResponse>>> pendingAppendResponsesByTerm = new ConcurrentHashMap<>();

private EntryHandler entryHandler = new EntryHandler(logger);

private QuorumAckChecker quorumAckChecker = new QuorumAckChecker(logger);

private Map<String, EntryDispatcher> dispatcherMap = new HashMap<>();

public void doWork() {
    try {
        if (!checkAndFreshState()) { // @1
            waitForRunning(1);
            return;
        }

        if (type.get() == PushEntryRequest.Type.APPEND) {
            doAppend(); // @2
        } else {
            doCompare(); // @3
        }
        waitForRunning(1);
    } catch (Throwable t) {
    }
}

private boolean checkAndFreshState() {
    if (!memberState.isLeader()) { // @1
        return false;
    }
    if (term != memberState.currTerm() || leaderId == null || !leaderId.equals(memberState.getLeaderId())) { // @2
        synchronized (memberState) {
            if (!memberState.isLeader()) {
                return false;
            }
            PreConditions.check(memberState.getSelfId().equals(memberState.getLeaderId()), DLedgerResponseCode.UNKNOWN);
            term = memberState.currTerm();
            leaderId = memberState.getSelfId();
            changeState(-1, PushEntryRequest.Type.COMPARE);
        }
    }
    return true;
}

private synchronized void changeState(long index, PushEntryRequest.Type target) {
    switch (target) {
        case APPEND:
            compareIndex = -1;
            updatePeerWaterMark(term, peerId, index);
            quorumAckChecker.wakeup();
            writeIndex = index + 1;
            break;
        case COMPARE:
            if (this.type.compareAndSet(PushEntryRequest.Type.APPEND, PushEntryRequest.Type.COMPARE)) {
                compareIndex = -1;
                pendingMap.clear();
            }
            break;
        case TRUNCATE:
            compareIndex = -1;
            break;
        default:
            break;
    }
    type.set(target);
}

private void doAppend() throws Exception {
    while (true) {
        if (!checkAndFreshState()) { // @1
            break;
        }
        if (type.get() != PushEntryRequest.Type.APPEND) { // @2
            break;
        }
        if (writeIndex > dLedgerStore.getLedgerEndIndex()) { // @3
            doCommit();
            doCheckAppendResponse();
            break;
        }
        if (pendingMap.size() >= maxPendingSize || (DLedgerUtils.elapsed(lastCheckLeakTimeMs) > 1000)) { // @4
            long peerWaterMark = getPeerWaterMark(term, peerId);
            for (Long index : pendingMap.keySet()) {
                if (index < peerWaterMark) {
                    pendingMap.remove(index);
                }
            }
            lastCheckLeakTimeMs = System.currentTimeMillis();
        }
        if (pendingMap.size() >= maxPendingSize) { // @5
            doCheckAppendResponse();
            break;
        }
        doAppendInner(writeIndex); // @6
        writeIndex++;
    }
}

private void doAppendInner(long index) throws Exception {
    DLedgerEntry entry = dLedgerStore.get(index);
    PreConditions.check(entry != null, DLedgerResponseCode.UNKNOWN, "writeIndex=%d", index);
    checkQuotaAndWait(entry); // @1
    PushEntryRequest request = buildPushRequest(entry, PushEntryRequest.Type.APPEND); // @2
    CompletableFuture<PushEntryResponse> responseFuture = dLedgerRpcService.push(request); // @3
    pendingMap.put(index, System.currentTimeMillis()); // @4
    responseFuture.whenComplete((x, ex) -> {
        try {
            PreConditions.check(ex == null, DLedgerResponseCode.UNKNOWN);
            DLedgerResponseCode responseCode = DLedgerResponseCode.valueOf(x.getCode());
            switch (responseCode) {
                case SUCCESS: // @5
                    pendingMap.remove(x.getIndex());
                    updatePeerWaterMark(x.getTerm(), peerId, x.getIndex());
                    quorumAckChecker.wakeup();
                    break;
                case INCONSISTENT_STATE: // @6
                    logger.info("[Push-{}]Get INCONSISTENT_STATE when push index={} term={}", peerId, x.getIndex(), x.getTerm());
                    changeState(-1, PushEntryRequest.Type.COMPARE);
                    break;
                default:
                    logger.warn("[Push-{}]Get error response code {} {}", peerId, responseCode, x.baseInfo());
                    break;
            }
        } catch (Throwable t) {
            logger.error("", t);
        }
    });
    lastPushCommitTimeMs = System.currentTimeMillis();
}

private PushEntryRequest buildPushRequest(DLedgerEntry entry, PushEntryRequest.Type target) {
    PushEntryRequest request = new PushEntryRequest();
    request.setGroup(memberState.getGroup());
    request.setRemoteId(peerId);
    request.setLeaderId(leaderId);
    request.setTerm(term);
    request.setEntry(entry);
    request.setType(target);
    request.setCommitIndex(dLedgerStore.getCommittedIndex());
    return request;
}

while (true) {
    if (!checkAndFreshState()) {
        break;
    }
    if (type.get() != PushEntryRequest.Type.COMPARE
        && type.get() != PushEntryRequest.Type.TRUNCATE) {
        break;
    }
    if (compareIndex == -1 && dLedgerStore.getLedgerEndIndex() == -1) {
        break;
    }
    // 对compareIndex进行校验修改
    if (compareIndex == -1) {
        compareIndex = dLedgerStore.getLedgerEndIndex();
    } else if (compareIndex > dLedgerStore.getLedgerEndIndex() || compareIndex < dLedgerStore.getLedgerBeginIndex()) {
        compareIndex = dLedgerStore.getLedgerEndIndex();
    }

    DLedgerEntry entry = dLedgerStore.get(compareIndex); // @1
    PushEntryRequest request = buildPushRequest(entry, PushEntryRequest.Type.COMPARE);
    CompletableFuture<PushEntryResponse> responseFuture = dLedgerRpcService.push(request); // @2
    PushEntryResponse response = responseFuture.get(3, TimeUnit.SECONDS); // @3
    long truncateIndex = -1;

    if (response.getCode() == DLedgerResponseCode.SUCCESS.getCode()) { 
        if (compareIndex == response.getEndIndex()) { // @4
            changeState(compareIndex, PushEntryRequest.Type.APPEND);
            break;
        } else { // @5
            truncateIndex = compareIndex;
        }
    } else if (response.getEndIndex() < dLedgerStore.getLedgerBeginIndex()
        || response.getBeginIndex() > dLedgerStore.getLedgerEndIndex()) { // @6
        truncateIndex = dLedgerStore.getLedgerBeginIndex();
    } else if (compareIndex < response.getBeginIndex()) { // @7
        truncateIndex = dLedgerStore.getLedgerBeginIndex();
    } else if (compareIndex > response.getEndIndex()) { // @8
        compareIndex = response.getEndIndex();
    } else { // @9
        compareIndex--;
    }
    if (compareIndex < dLedgerStore.getLedgerBeginIndex()) { // @10
        truncateIndex = dLedgerStore.getLedgerBeginIndex();
    }
    if (truncateIndex != -1) {
        changeState(truncateIndex, PushEntryRequest.Type.TRUNCATE);
        doTruncate(truncateIndex);
        break;
    }
}

private void doTruncate(long truncateIndex) throws Exception {
    DLedgerEntry truncateEntry = dLedgerStore.get(truncateIndex);
    PushEntryRequest truncateRequest = buildPushRequest(truncateEntry, PushEntryRequest.Type.TRUNCATE);
    PushEntryResponse truncateResponse = dLedgerRpcService.push(truncateRequest).get(3, TimeUnit.SECONDS);
    lastPushCommitTimeMs = System.currentTimeMillis();
    changeState(truncateIndex, PushEntryRequest.Type.APPEND);
}

public void doWork() {
    try {
        if (!memberState.isFollower()) {
            waitForRunning(1);
            return;
        }
        if (compareOrTruncateRequests.peek() != null) { // @1
            Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>> pair = compareOrTruncateRequests.poll();
            PreConditions.check(pair != null, DLedgerResponseCode.UNKNOWN);
            switch (pair.getKey().getType()) { 
                case TRUNCATE:// @2
                    handleDoTruncate(pair.getKey().getEntry().getIndex(), pair.getKey(), pair.getValue());
                    break;
                case COMPARE: // @3
                    handleDoCompare(pair.getKey().getEntry().getIndex(), pair.getKey(), pair.getValue());
                    break;
                case COMMIT: // @4
                    handleDoCommit(pair.getKey().getCommitIndex(), pair.getKey(), pair.getValue());
                    break;
                default:
                    break;
            }
        } else {
            long nextIndex = dLedgerStore.getLedgerEndIndex() + 1; // @5
            Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>> pair = writeRequestMap.remove(nextIndex);
            if (pair == null) {
                checkAbnormalFuture(dLedgerStore.getLedgerEndIndex()); // @6
                waitForRunning(1);
                return;
            }
            PushEntryRequest request = pair.getKey();
            handleDoAppend(nextIndex, request, pair.getValue()); // @7
        }
    } catch (Throwable t) {
        DLedgerEntryPusher.logger.error("Error in {}", getName(), t);
        DLedgerUtils.sleep(100);
    }
}

private CompletableFuture<PushEntryResponse> handleDoTruncate(long truncateIndex, PushEntryRequest request,
    CompletableFuture<PushEntryResponse> future) {
    try {
        long index = dLedgerStore.truncate(request.getEntry(), request.getTerm(), request.getLeaderId());
        PreConditions.check(index == truncateIndex, DLedgerResponseCode.INCONSISTENT_STATE);
        future.complete(buildResponse(request, DLedgerResponseCode.SUCCESS.getCode()));
        dLedgerStore.updateCommittedIndex(request.getTerm(), request.getCommitIndex());
    } catch (Throwable t) {
        logger.error("[HandleDoTruncate] truncateIndex={}", truncateIndex, t);
        future.complete(buildResponse(request, DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
    }
    return future;
}

private CompletableFuture<PushEntryResponse> handleDoCompare(long compareIndex, PushEntryRequest request,
    CompletableFuture<PushEntryResponse> future) {
    try {
        DLedgerEntry local = dLedgerStore.get(compareIndex);
        PreConditions.check(request.getEntry().equals(local), DLedgerResponseCode.INCONSISTENT_STATE);
        future.complete(buildResponse(request, DLedgerResponseCode.SUCCESS.getCode()));
    } catch (Throwable t) {
        logger.error("[HandleDoCompare] compareIndex={}", compareIndex, t);
        future.complete(buildResponse(request, DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
    }
    return future;
}

private CompletableFuture<PushEntryResponse> handleDoCommit(long committedIndex, PushEntryRequest request,
    CompletableFuture<PushEntryResponse> future) {
    try {
        dLedgerStore.updateCommittedIndex(request.getTerm(), committedIndex);
        future.complete(buildResponse(request, DLedgerResponseCode.SUCCESS.getCode()));
    } catch (Throwable t) {
        logger.error("[HandleDoCommit] committedIndex={}", request.getCommitIndex(), t);
        future.complete(buildResponse(request, DLedgerResponseCode.UNKNOWN.getCode()));
    }
    return future;
}

private void handleDoAppend(long writeIndex, PushEntryRequest request,
    CompletableFuture<PushEntryResponse> future) {
    try {
        DLedgerEntry entry = dLedgerStore.appendAsFollower(request.getEntry(), request.getTerm(), request.getLeaderId());
        future.complete(buildResponse(request, DLedgerResponseCode.SUCCESS.getCode()));
        dLedgerStore.updateCommittedIndex(request.getTerm(), request.getCommitIndex());
    } catch (Throwable t) {
        logger.error("[HandleDoWrite] writeIndex={}", writeIndex, t);
        future.complete(buildResponse(request, DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
    }
}

private void checkAbnormalFuture(long endIndex) {
    if (DLedgerUtils.elapsed(lastCheckFastForwardTimeMs) < 1000) {
        return;
    }
    lastCheckFastForwardTimeMs  = System.currentTimeMillis();
    if (writeRequestMap.isEmpty()) {
        return;
    }
    long minFastForwardIndex = Long.MAX_VALUE;
    for (Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>> pair : writeRequestMap.values()) {
        long index = pair.getKey().getEntry().getIndex();
        //Fall behind
        if (index <= endIndex) {
            try {
                DLedgerEntry local = dLedgerStore.get(index);
                PreConditions.check(pair.getKey().getEntry().equals(local), DLedgerResponseCode.INCONSISTENT_STATE);
                pair.getValue().complete(buildResponse(pair.getKey(), DLedgerResponseCode.SUCCESS.getCode()));
                logger.warn("[PushFallBehind]The leader pushed an entry index={} smaller than current ledgerEndIndex={}, maybe the last ack is missed", index, endIndex);
            } catch (Throwable t) {
                logger.error("[PushFallBehind]The leader pushed an entry index={} smaller than current ledgerEndIndex={}, maybe the last ack is missed", index, endIndex, t);
                pair.getValue().complete(buildResponse(pair.getKey(), DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
            }
            writeRequestMap.remove(index);
            continue;
        }
        //Just OK
        if (index ==  endIndex + 1) {
            //The next entry is coming, just return
            return;
        }
        //Fast forward
        TimeoutFuture<PushEntryResponse> future  = (TimeoutFuture<PushEntryResponse>) pair.getValue();
        if (!future.isTimeOut()) {
            continue;
        }
        if (index < minFastForwardIndex) {
            minFastForwardIndex = index;
        }
    }
    if (minFastForwardIndex == Long.MAX_VALUE) {
        return;
    }
    Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>> pair = writeRequestMap.get(minFastForwardIndex);
    if (pair == null) {
        return;
    }
    logger.warn("[PushFastForward] ledgerEndIndex={} entryIndex={}", endIndex, minFastForwardIndex);
    pair.getValue().complete(buildResponse(pair.getKey(), DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
}

if (pendingAppendResponsesByTerm.size() > 1) {
    for (Long term : pendingAppendResponsesByTerm.keySet()) {
        if (term == currTerm) {
            continue;
        }
        for (Map.Entry<Long, TimeoutFuture<AppendEntryResponse>> futureEntry : pendingAppendResponsesByTerm.get(term).entrySet()) {
            AppendEntryResponse response = new AppendEntryResponse();
            response.setGroup(memberState.getGroup());
            response.setIndex(futureEntry.getKey());
            response.setCode(DLedgerResponseCode.TERM_CHANGED.getCode());
            response.setLeaderId(memberState.getLeaderId());
            logger.info("[TermChange] Will clear the pending response index={} for term changed from {} to {}", futureEntry.getKey(), term, currTerm);
            futureEntry.getValue().complete(response);
        }
        pendingAppendResponsesByTerm.remove(term);
    }
}
if (peerWaterMarksByTerm.size() > 1) {
    for (Long term : peerWaterMarksByTerm.keySet()) {
        if (term == currTerm) {
            continue;
        }
        logger.info("[TermChange] Will clear the watermarks for term changed from {} to {}", term, currTerm);
        peerWaterMarksByTerm.remove(term);
    }
}

Map<String, Long> peerWaterMarks = peerWaterMarksByTerm.get(currTerm);

long quorumIndex = -1;
for (Long index : peerWaterMarks.values()) {
    int num = 0;
    for (Long another : peerWaterMarks.values()) {
        if (another >= index) {
            num++;
        }
    }
    if (memberState.isQuorum(num) && index > quorumIndex) {
        quorumIndex = index;
    }
}

if (quorumIndex >= 0) {
    for (Long i = quorumIndex; i >= 0; i--) {
        try {
            CompletableFuture<AppendEntryResponse> future = responses.remove(i);
            if (future == null) {
                needCheck = lastQuorumIndex != -1 && lastQuorumIndex != quorumIndex && i != lastQuorumIndex;
                break;
            } else if (!future.isDone()) {
                AppendEntryResponse response = new AppendEntryResponse();
                response.setGroup(memberState.getGroup());
                response.setTerm(currTerm);
                response.setIndex(i);
                response.setLeaderId(memberState.getSelfId());
                response.setPos(((AppendFuture) future).getPos());
                future.complete(response);
            }
            ackNum++; // 本次完成确认提交的日志数
        } catch (Throwable t) {
            logger.error("Error in ack to index={} term={}", i, currTerm, t);
        }
    }
}

if (ackNum == 0) {
    for (long i = quorumIndex + 1; i < Integer.MAX_VALUE; i++) {
        TimeoutFuture<AppendEntryResponse> future = responses.get(i);
        if (future == null) {
            break;
        } else if (future.isTimeOut()) {
            AppendEntryResponse response = new AppendEntryResponse();
            response.setGroup(memberState.getGroup());
            response.setCode(DLedgerResponseCode.WAIT_QUORUM_ACK_TIMEOUT.getCode());
            response.setTerm(currTerm);
            response.setIndex(i);
            response.setLeaderId(memberState.getSelfId());
            future.complete(response);
        } else {
            break;
        }
    }
    waitForRunning(1);
}

if (DLedgerUtils.elapsed(lastCheckLeakTimeMs) > 1000 || needCheck) {
    updatePeerWaterMark(currTerm, memberState.getSelfId(), dLedgerStore.getLedgerEndIndex());
    for (Map.Entry<Long, TimeoutFuture<AppendEntryResponse>> futureEntry : responses.entrySet()) {
        if (futureEntry.getKey() < quorumIndex) {
            AppendEntryResponse response = new AppendEntryResponse();
            response.setGroup(memberState.getGroup());
            response.setTerm(currTerm);
            response.setIndex(futureEntry.getKey());
            response.setLeaderId(memberState.getSelfId());
            response.setPos(((AppendFuture) futureEntry.getValue()).getPos());
            futureEntry.getValue().complete(response);
            responses.remove(futureEntry.getKey());
        }
    }
    lastCheckLeakTimeMs = System.currentTimeMillis();
}
lastQuorumIndex = quorumIndex;

while (running) {
                switch (getPeerState()) {
                case LOOKING:
                    LOG.info("LOOKING");

                    if (Boolean.getBoolean("readonlymode.enabled")) {...} else {
                        try {
                           reconfigFlagClear();
                           ...
                            setCurrentVote(makeLEStrategy().lookForLeader());
                        } catch (Exception e) {
                            LOG.warn("Unexpected exception", e);
                            setPeerState(ServerState.LOOKING);
                        }                        
                    }
                    break;
                case FLOWING: ...
                case ...

HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();
HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();
int notTimeout = finalizeWait;
synchronized(this){
    logicalclock.incrementAndGet();
    updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
}

sendNotifications();

for (long sid : self.getCurrentAndNextConfigVoters()) {
    QuorumVerifier qv = self.getQuorumVerifier();
    ToSend notmsg = new ToSend(ToSend.mType.notification, // type
            proposedLeader, // 投给谁? 这里是myid
            proposedZxid, // zxid
            logicalclock.get(), // 本地投票的epoch
            QuorumPeer.ServerState.LOOKING,
            sid, // 对面的id
            proposedEpoch, // 当前投的leader的epoch
            qv.toString().getBytes());
    sendqueue.offer(notmsg); // 放入发送队列中, 另外一个WorkSender线程会去发送
}

while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){
    Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);
    if (validVoter(n.sid) && validVoter(n.leader)) {
        switch (n.state) {
            case LOOKING:
            case FOLLOWING AND LEADING:
        }
    }
}

if (n.electionEpoch > logicalclock.get()) {
    logicalclock.set(n.electionEpoch); // 设置新的投票轮次
    recvset.clear(); // 当前进行的投票过时了, 将recvset清除
    if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
            getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
        updateProposal(n.leader, n.zxid, n.peerEpoch);
    } else {
        updateProposal(getInitId(),
                getInitLastLoggedZxid(),
                getPeerEpoch());
    }
    sendNotifications();
} else if (n.electionEpoch < logicalclock.get()) {
    if(LOG.isDebugEnabled()){
        LOG.debug("Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x"
                + Long.toHexString(n.electionEpoch)
                + ", logicalclock=0x" + Long.toHexString(logicalclock.get()));
    }
    break; // break switch
} else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
        proposedLeader, proposedZxid, proposedEpoch)) {
    updateProposal(n.leader, n.zxid, n.peerEpoch);
    sendNotifications();
}
recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));

protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
      return ((newEpoch > curEpoch) ||
                ((newEpoch == curEpoch) &&
                ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))));
    }

if (termPredicate(recvset,
        new Vote(proposedLeader, proposedZxid,
                logicalclock.get(), proposedEpoch))) {
    // 此处在确定在finalizeWait时间内没有其他获胜者存在,
    // 因为zk在退出follower之后会提交所有没提交的txn, 所以不会
    // 像raft那样确保大多数节点具有最新提交的事务.
    while((n = recvqueue.poll(finalizeWait,
            TimeUnit.MILLISECONDS)) != null){
        if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                proposedLeader, proposedZxid, proposedEpoch)){
            recvqueue.put(n);
            break;
        }
    }
    if (n == null) {
        self.setPeerState((proposedLeader == self.getId()) ?
                ServerState.LEADING: learningState());
        Vote endVote = new Vote(proposedLeader,
                proposedZxid, logicalclock.get(), 
                proposedEpoch);
        leaveInstance(endVote);
        return endVote;
    }
}

if(n.electionEpoch == logicalclock.get()){
    recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
    if(termPredicate(recvset, new Vote(n.version, n.leader,
                    n.zxid, n.electionEpoch, n.peerEpoch, n.state))
                    && checkLeader(outofelection, n.leader, n.electionEpoch)) {
        self.setPeerState((n.leader == self.getId()) ?
                ServerState.LEADING: learningState());
        Vote endVote = new Vote(n.leader, 
                n.zxid, n.electionEpoch, n.peerEpoch);
        leaveInstance(endVote);
        return endVote;
    }
}

outofelection.put(n.sid, new Vote(n.version, n.leader, 
        n.zxid, n.electionEpoch, n.peerEpoch, n.state));
if (termPredicate(outofelection, new Vote(n.version, n.leader,
        n.zxid, n.electionEpoch, n.peerEpoch, n.state))
        && checkLeader(outofelection, n.leader, n.electionEpoch)) {
    synchronized(this){
        logicalclock.set(n.electionEpoch);
        self.setPeerState((n.leader == self.getId()) ?
                ServerState.LEADING: learningState());
    }
    Vote endVote = new Vote(n.leader, n.zxid, 
            n.electionEpoch, n.peerEpoch);
    leaveInstance(endVote);
    return endVote;
}

protected boolean checkLeader(
            Map<Long, Vote> votes,
            long leader,
            long electionEpoch){

        boolean predicate = true;

        if(leader != self.getId()){
            if(votes.get(leader) == null) predicate = false;
            else if(votes.get(leader).getState() != ServerState.LEADING) predicate = false;
        } else if(logicalclock.get() != electionEpoch) {
            predicate = false;
        }

        return predicate;
    }

cnxAcceptor = new LearnerCnxAcceptor(); 
cnxAcceptor.start();

long epoch = getEpochToPropose(self.getId(), self.getAcceptedEpoch());

public long getEpochToPropose(long sid, long lastAcceptedEpoch) throws InterruptedException, IOException {
    synchronized(connectingFollowers) {
        if (!waitingForNewEpoch) {
            return epoch;
        }
        if (lastAcceptedEpoch >= epoch) {
            epoch = lastAcceptedEpoch+1;
        }
        if (isParticipant(sid)) {
            connectingFollowers.add(sid);
        }
        QuorumVerifier verifier = self.getQuorumVerifier();
        if (connectingFollowers.contains(self.getId()) &&
                                        verifier.containsQuorum(connectingFollowers)) {
            waitingForNewEpoch = false;
            self.setAcceptedEpoch(epoch);
            connectingFollowers.notifyAll();
        } else {
            long start = Time.currentElapsedTime();
            long cur = start;
            long end = start + self.getInitLimit()*self.getTickTime();
            while(waitingForNewEpoch && cur < end) {
                connectingFollowers.wait(end - cur);
                cur = Time.currentElapsedTime();
            }
            if (waitingForNewEpoch) {
                throw new InterruptedException("Timeout while waiting for epoch from quorum");
            }
        }
        return epoch;
    }
}

connectToLeader(leaderServer.addr, leaderServer.hostname); // 与LEADER建立连接
long newEpochZxid = registerWithLeader(Leader.FOLLOWERINFO); // 发送握手消息以及当前的acceptedEpoch,并且返回与LEADER达成一致的新的EPOCH

QuorumPacket qp = new QuorumPacket();                
qp.setType(pktType);
qp.setZxid(ZxidUtils.makeZxid(self.getAcceptedEpoch(), 0));
LearnerInfo li = new LearnerInfo(self.getId(), 0x10000, self.getQuorumVerifier().getVersion());
writePacket(qp, true);
readPacket(qp);

private int type; // 指明消息类别
  private long zxid; // zxid
  private byte[] data; // 数据
  private java.util.List<org.apache.zookeeper.data.Id> authinfo; // 认证信息

// .. 省略IO序列化相关代码
QuorumPacket qp = new QuorumPacket(); // @1
readRecord(qp, "packet");
byte learnerInfoData[] = qp.getData();
long lastAcceptedEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());
long zxid = qp.getZxid();
long newEpoch = leader.getEpochToPropose(this.getSid(), lastAcceptedEpoch);
long newLeaderZxid = ZxidUtils.makeZxid(newEpoch, 0);
QuorumPacket newEpochPacket = new QuorumPacket(Leader.LEADERINFO, newLeaderZxid, ver, null); // @2
oa.writeRecord(newEpochPacket, "packet");

ia.readRecord(ackEpochPacket, "packet");
ByteBuffer bbepoch = ByteBuffer.wrap(ackEpochPacket.getData());
ss = new StateSummary(bbepoch.getInt(), ackEpochPacket.getZxid());
leader.waitForEpochAck(this.getSid(), ss); // @3

readPacket(qp);
final long newEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());
if (newEpoch > self.getAcceptedEpoch()) {
    wrappedEpochBytes.putInt((int)self.getCurrentEpoch());
    self.setAcceptedEpoch(newEpoch);
} else {
    throw new IOException("Leaders epoch, " + newEpoch + " is less than accepted epoch, " + self.getAcceptedEpoch());
}
QuorumPacket ackNewEpoch = new QuorumPacket(Leader.ACKEPOCH, lastLoggedZxid, epochBytes, null);
writePacket(ackNewEpoch, true);
return ZxidUtils.makeZxid(newEpoch, 0);

waitForEpochAck(self.getId(), leaderStateSummary);
self.setCurrentEpoch(epoch);

waitForNewLeaderAck(self.getId(), zk.getZxid());

maxCommittedLog: 提交的最大zxid
minCommittedLog: 内存中提交的最小zxid
lastProcessedZxid: dataTree上提交的最大zxid
peerLastZxid: follower的最大zxid,这个zxid是日志中的,可能是没有提交的.

if (lastProcessedZxid == peerLastZxid) {
    queueOpPacket(Leader.DIFF, peerLastZxid);
} else if (peerLastZxid > maxCommittedLog && !isPeerNewEpochZxid) {
    // 对应单TRUNC,不会发送额外的packet
    queueOpPacket(Leader.TRUNC, maxCommittedLog);
    currentZxid = maxCommittedLog;
} else if ((maxCommittedLog >= peerLastZxid) && (minCommittedLog <= peerLastZxid)) {
    // 对应DIFF或者TRUNC+packets
    Iterator<Proposal> itr = db.getCommittedLog().iterator();
    currentZxid = queueCommittedProposals(itr, peerLastZxid,null, maxCommittedLog);
} else if (peerLastZxid < minCommittedLog && txnLogSyncEnabled) {
    // 对应DIFF/SNAP
    long sizeLimit = db.calculateTxnLogSizeLimit();
    Iterator<Proposal> txnLogItr = db.getProposalsFromTxnLog(
                        peerLastZxid, sizeLimit);
    if (txnLogItr.hasNext()) {
        currentZxid = queueCommittedProposals(txnLogItr, peerLastZxid,
                                                minCommittedLog, maxCommittedLog);

        Iterator<Proposal> committedLogItr = db.getCommittedLog().iterator();
        currentZxid = queueCommittedProposals(committedLogItr, currentZxid,
                                                null, maxCommittedLog);
    } else {
        snap = true;
    }
}
leaderLastZxid = leader.startForwarding(this, currentZxid);

long prevProposalZxid = -1;
long queuedZxid = peerLastZxid;
while (itr.hasNext()) {
    Proposal propose = itr.next();
    long packetZxid = propose.packet.getZxid();
    // abort if we hit the limit
    if ((maxZxid != null) && (packetZxid > maxZxid)) {
        break;
    }
        // 这个proposal follower已经有了,继续下一条
    if (packetZxid < peerLastZxid) {
        prevProposalZxid = packetZxid;
        continue;
    }

    if (needOpPacket) {
        // 说明之后掉事务都需要同步
        if (packetZxid == peerLastZxid) {
            queueOpPacket(Leader.DIFF, lastCommittedZxid);
            needOpPacket = false;
            continue;
        }

        if (isPeerNewEpochZxid) {
            queueOpPacket(Leader.DIFF, lastCommittedZxid);
            needOpPacket = false;
        } else if (packetZxid > peerLastZxid  ) {
            if (ZxidUtils.getEpochFromZxid(packetZxid) !=
                    ZxidUtils.getEpochFromZxid(peerLastZxid)) {
                // epoch不同,不能trunc https://issues.apache.org/jira/projects/ZOOKEEPER/issues/ZOOKEEPER-3607?filter=allopenissues
                LOG.warn("Cannot send TRUNC to peer sid: " + getSid() +
                            " peer zxid is from different epoch" );
                return queuedZxid;
            }
            // 代表这个提交follower 有,但是leader没有,从上一个大家都有的开始truncate
            queueOpPacket(Leader.TRUNC, prevProposalZxid);
            needOpPacket = false;
        }
    }
    // 发送proposal
    queuePacket(propose.packet);
    // 发送commit
    queueOpPacket(Leader.COMMIT, packetZxid);
    queuedZxid = packetZxid;
}

writePacket(new QuorumPacket(Leader.ACK, newLeaderZxid, null, null), true);

QuorumPacket newLeaderQP = new QuorumPacket(Leader.NEWLEADER,
        newLeaderZxid, leader.self.getLastSeenQuorumVerifier()
                .toString().getBytes(), null);
queuedPackets.add(newLeaderQP);

leader.waitForNewLeaderAck(getSid(), qp.getZxid());

case Leader.ACK:
    leader.processAck(this.sid, qp.getZxid(), sock.getLocalSocketAddress());
case Leader.PING:
    leader.zk.touch(sess, to);
case Leader.REQUEST:
    leader.zk.submitLearnerRequest(si);

QuorumPacket qp = new QuorumPacket();
while (this.isRunning()) {
    readPacket(qp);
    processPacket(qp);
}

startZkServer();
// 判断心跳超时时间内是否与绝大多数follower保持连接

kafka.network.SocketServer#startup
createDataPlaneAcceptorsAndProcessors // 1
	for each endpoint:
		dataPlaneAcceptor = createAcceptor()
			new Acceoptor
		addDataPlaneProcessors(acceptor, processorNum) // 2
			new Processor // loop processorNum
			dataPlaneRequestChannel.addProcessor(processor) 
			acceptor.addProcessors // 3
startProcessingRequests // 4
	startDataPlaneProcessorsAndAcceptors
		acceptor.startProcessors
			for each processor.start()

serverChannel = openServerSocket(endPoint.host, endPoint.port) // 1
	serverChannel = ServerSocketChannel.open()
	serverChannel.socket.bind(socketAddress)
serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT) // 2
while(isRunning)
	ready = nioSelector.select(500)
	if (ready > 0) {
		keys = nioSelector.selectedKeys() //3
		for key in keys is acceptable
		socketChannel = accept(key)
			socketChannel = serverSocketChannel.accept()
		assignNewConnection(socketChannel, processor) // 4
			processor.accept(socketChannel)
				newConnections.offer(socketChannel)
				~ async
				processor.configureNewConnections // 5
					channel = newConnections.poll()
					selector.register(connectionId(channel.socket), channel)
						registerChannel(id, socketChannel, SelectionKey.OP_READ);
					...
	}

private val newConnections = new ArrayBlockingQueue[SocketChannel](connectionQueueSize)
private val inflightResponses = mutable.Map[String, RequestChannel.Response]()
private val responseQueue = new LinkedBlockingDeque[RequestChannel.Response]()
private val selector = createSelector(...)

while (isRunning) {
  configureNewConnections() // 1
  processNewResponses() // 2
  poll() // 3
  processCompletedReceives() // 4
  processCompletedSends() // 5
  processDisconnected() // 6
  closeExcessConnections() // 7
}

startRWLoop:
  connection.startReadLoop
    connection.doRead
	  c.readBuffer.ReadOnce(c.rawConnection)
	  c.onRead
	    c.filterManager.OnRead()

FilterManager.onContinueReading
  activeReadFilter.filter.OnData(fm.conn.GetReadBuffer())

proxy.serverStreamConn.Dispatch(buf)
  streamConn.Dispatch(buf)
  frame = streamConn.protocol.Decode(buf)
  // 如果读到了足够多的数据
  streamConn.handleFrame(streamCtx, frame)
    switch frame.GetStreamType() {
	case xprotocol.Request:
		sc.handleRequest(ctx, frame, false)
	case xprotocol.RequestOneWay:
		sc.handleRequest(ctx, frame, true)
	case xprotocol.Response:
		sc.handleResponse(ctx, frame)
	}

if frame.IsHeartbeatFrame() {
  // 处理心跳逻辑
}

serverStream := sc.newServerStream(ctx, frame)
proxy.NewStreamDetect
  downstream = newActiveStream // downstream是一个重要的东西
  downstream.OnReceive
    ~ downStream.receive

downstream.matchRoute()
  routersImpl.CallMakeHandlerChain
    DefaultMakeHandlerChain
	  routersImpl.MatchRoute(headers)
	    routersImpl.findVirtualHost(headers) // 根据virtualHost的domain配置可能返回默认的
  s.snapshot, s.route = handlerChain.DoNextHandler()

downStream.chooseHost()
  s.cluster = s.snapshot.ClusterInfo()
  s.initializeUpstreamConnectionPool
    connPool = s.proxy.clusterManager.ConnPoolForCluster(s.snapshot, currentProtocol) // 这里会进行连接池的创建和初始化
	  clusterManager.getActiveConnectionPool // 这里会进行负载均衡
	    clusterSnapshot.LoadBalancer().ChooseHost(balancerContext)
  s.requestInfo.OnUpstreamHostSelected(pool.Host()) // 设置host

downstream.receiveHeaders()
  downstream.upstreamRequest.appendHeaders()
  upstreamRequest.connPool.NewStream(downStream.context, r, r)
    streamEncoder = activeClient.client.NewStream(ctx, responseDecoder) // responseDecoder就是upstreamRequest
      streamSender := client.ClientStreamConnection.NewStream(context, receiver)
	    clientStream:=streamConn.newClientStream()
	    clientStream.receiver = receiver
		streamConn.clientStreams[clientStream.id] = clientStream // 方便响应到来时进行对应
	  receiver.stream = streamSender.GetStream()
    upstreamRequest.OnReady(streamEncoder, host)

downstream.receiveData()
  s.onUpstreamRequestSent() // 设置超时
  s.upstreamRequest.appendData(endStream)
    r.requestSender.AppendData(r.downStream.context, r.convertData(data), endStream)
      if endStream {
		  s.endStream() // 发送请求
		    buf, err := s.sc.protocol.Encode(s.ctx, s.frame)
			s.sc.netConn.Write(buf)
	  }

streamConn.handleResponse
  clientStream, ok := sc.clientStreams[requestId]
  clientStream.receiver.OnReceive(clientStream.ctx, frame.GetHeader(), frame.GetData(), nil) // clientStream是xStream,receiver是StreamReceiveListener即upstreamRequest
    r.downStream.downstreamRespHeaders = headers
	r.downStream.downstreamRespDataBuf = data
	r.downStream.downstreamRespTrailers = trailers
	r.downStream.sendNotify() // 发送响应通知

upstreamRequest.receiveData()
  downStream.onUpstreamData()
    downStream.onUpstreamResponseRecvFinished()
	  downStream.cleanUp()
	downStream.appendData(endStream)
	  data := s.convertData(s.downstreamRespDataBuf) // 当下游数据和上游数据当协议不一致时进行转换
	  xStream.AppendData(data types.IoBuffer)
	    xStream.endStream()
		  buf, err := s.sc.protocol.Encode(s.ctx, s.frame) // 给响应编码
		  err = s.sc.netConn.Write(buf) // 写出响应
		downStream.endStream()
		  downStream.cleanStream()

new Zookeeper() // @1
  ClientCnxnSocket getClientCnxnSocket()
    new ClientCnxnSocketNetty().connect()
	  bootstrap.connect() // @2
	    incomingBuffer = lenBuffer; // @3
		ClientCncx.primeConnection() // @4
		ZKClientPipelineFactory.initChannel() 
		  pipeline.addLast("handler", new ZKClientHandler()) // @5

requestHeader // xid type
request  // Record
replyHeader // xid zxid err
response // Record
callback
context

if (!clientCnxnSocket.isConnected())
  clientCnxnSocket.connect(addr) // @1
clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn)
  Packet head = outgoingQueue.poll(waitTimeOut, TimeUnit.MILLISECONDS) // @2
  doWrite(pendingQueue, head, cnxn);
    while (true) {
	  p.requestHeader.setXid(cnxn.getXid()); // @3
	  synchronized (pendingQueue) { // @4
	    pendingQueue.add(p);
	  }
	  sendPktOnly(p); // 不flush @5
	  anyPacketsSent = true;
	  if (outgoingQueue.isEmpty()) // @6
	    break;
	}
	if (anyPacketsSent)
	  channel.flush(); // @7

// 处理pings、AuthPacket、notification的响应... 略过,不是我们分析的重点
SendThread.readResponse() // @1
  packet = pendingQueue.remove(); // @2
  lastZxid = replyHdr.getZxid(); // @3
  SendThread.finishPacket() 
    if (p.cb == null) { // @4
	  p.notifyAll();
	} else { 
	  eventThread.queuePacket(p); // @5
	}

Zookeeper.create()
  ClientCnxn.queuePacket()
    packet = new Packet()
	outgoingQueue.add(packet);
	sendThread.getClientCnxnSocket().packetAdded();

while (true) {
  Object event = waitingEvents.take(); // @1
  processEvent(event); // @2
  ...
	  if (p.response instanceof Create2Response) {
		cb.processResult()
	  }
}

Zookeeper.create()
	ClientCnxn.submitRequest() // @1
	  Packet packet = queuePacket()  // @2
	  while (!packet.finished) {
		packet.wait(); // @3
	  }

protected void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) throws Exception {
while (buf.isReadable()) {
	if (incomingBuffer.remaining() > buf.readableBytes()) {
		int newLimit = incomingBuffer.position()
				+ buf.readableBytes();
		incomingBuffer.limit(newLimit);
	}
	buf.readBytes(incomingBuffer); // @1
	incomingBuffer.limit(incomingBuffer.capacity());

	if (!incomingBuffer.hasRemaining()) {
		incomingBuffer.flip();
		if (incomingBuffer == lenBuffer) { // @2
			recvCount.getAndIncrement();
			readLength();
			  -> incomingBuffer = ByteBuffer.allocate(len);
		} else if (!initialized) { // @3
			readConnectResult();
			lenBuffer.clear();
			incomingBuffer = lenBuffer;
			initialized = true;
			updateLastHeard();
		} else {
			sendThread.readResponse(incomingBuffer); // @4
			lenBuffer.clear();
			incomingBuffer = lenBuffer;
			updateLastHeard();
		}
	}
}
wakeupCnxn();
}

void channelActive()
  NettyServerCnxn cnxn = new NettyServerCnxn()
  ctx.channel().attr(CONNECTION_ATTRIBUTE).set(cnxn); // @1

void channelRead()
  NettyServerCnxn cnxn = ctx.channel().attr(CONNECTION_ATTRIBUTE).get(); // @1
  cnxn.processMessage((ByteBuf) msg); // @2
    NettyServerCnxn.receiveMessage(buf);
	  ZooKeeperServer.processPacket(buf);
	    ZooKeeperServer.submitRequest(Request);
		  firstProcessor.processRequest(Request); // @3

void processRequest(Request request);
void shutdown();

sessionId
cxid
type
TxnHeader hdr
zxid

loop:
  Request request = queuedRequests.take(); // @1
  nextProcessor.processRequest(request); // @2
  switch (request.type) {
    case OpCode.create 
	...
	  zks.getFollower().request(request); // @3
  }

processRequest:
  zks.processTxn(request); // @1
    dataTree.processTxn(hdr, txn);
  if (request.isQuorum()) // @2
    zks.getZKDatabase().addCommittedProposal(request);
    zks.getZKDatabase().addCommittedProposal(request);
  rsp = new CreateResponse(); // @3
  cnxn.sendResponse(rsp)

Follower#processPacket
  case Leader.PROPOSAL:
    FollowerZooKeeperServer.logRequest(hdr, txn);
	  syncProcessor.processRequest(request);

SyncRequestProcessor用于以下三个场景:
1. Leader: 将request写到磁盘,并且转发给AckRequestProcessor,AckRequestProcessor的作用是回复一个ack给自己.
2. Follower: 将request写到磁盘,并且转发给SendAckRequestProcessor,它将给Leader发送响应
3. Observer: 将已提交的请求刷至磁盘,它的下一个processor是null,也就说它不会给Leader发送响应.

public void processRequest(Request request) {
	// request.addRQRec(">sync");
	queuedRequests.add(request);
}

loop:
  si = queuedRequests.poll(); // @1
  if si == null // @2
    flush(toFlush)
  zks.getZKDatabase().append(si) // @3
  toFlush.add(si) // @4
  if (toFlush.size() > 1000)  // @5
    flush(toFlush);

QuorumPacket qp = new QuorumPacket
  learner.writePacket

prepRequestProcessor.processRequest(request);

public void processRequest(Request request) {
	submittedRequests.add(request);
}

while (true) {
	Request request = submittedRequests.take();
	if (Request.requestOfDeath == request) { // "毒丸"设计模式
		break;
	}
	pRequest(request);
}
}

pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request, true)
  request.setHdr(new TxnHeader(request.sessionId, request.cxid, zxid, Time.currentWallTime(), type)); // @1
  pRequest2TxnCreate(type, request, record, deserialize);
    int newCversion = parentRecord.stat.getCversion()+1; // @2
	request.setTxn(new CreateTxn(path, data, listACL, createMode.isEphemeral(),
                    newCversion)); // @3 ?
	parentRecord.childCount++; // @4
	parentRecord.stat.setCversion(newCversion); 
	addChangeRecord(parentRecord); // @5
	  zks.outstandingChanges.add(c);
	  zks.outstandingChangesForPath.put(c.path, c);
	addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, s, 0, listACL)); // @6

nextProcessor.processRequest(request);
if (request.getHdr() != null) {
	// We need to sync and get consensus on any transactions
	try {
		zks.getLeader().propose(request);
		  QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, data);
		  outstandingProposals.put(lastProposed, p);
		  sendPacket(pp);
	} catch (XidRolloverException e) {
		throw new RequestProcessorException(e.getMessage(), e);
	}
	syncProcessor.processRequest(request);
}

QuorumPeer self = leader.self;
if(self != null)
	leader.processAck(self.getId(), request.zxid, null);

Follower.followLeader
  while (this.isRunning()) {
	readPacket(qp);
	processPacket(qp);
	  case Leader.PROPOSAL:
	    // 反序列化...
		syncProcessor.processRequest(request);
  }

LearnerHandler.run
  ...
  case Leader.ACK: // @1
    leader.processAck
	  Proposal p = outstandingProposals.get(zxid); // @2
	  tryToCommit(p, zxid, followerAddr) 
	    if (p.hasAllQuorums()) // @3
		{
		  outstandingProposals.remove(zxid); // @3.1
		  toBeApplied.add(p); // @3.2
		  zk.commitProcessor.commit(p.request); // @3.3
		}

template <typename T>
class Grid {
    ...
    Grid(const Grid& src) = default;
    Grid<T>& operator=(const Grid& rhs) = default;
    
    const std::optional<T>& at(size_t x, size_t y) const;
}

template <typename T>
const std::optional<T>& Grid<T>::at(size_t x, size_t y) const {
    ...
}

template <typename T>
class Grid
{
    public:
        // Omitted for brevity

        template <typename E>
        Grid(const Grid<E>& src);

        template <typename E>
        Grid<T>& operator=(const Grid<E>& rhs);

        void swap(Grid& other) noexcept;

        // Omitted for brevity
};

template <typename T>
template <typename E>
Grid<T>::Grid(const Grid<E>& src)
    : Grid(src.getWidth(), src.getHeight())
{
    // The ctor-initializer of this constructor delegates first to the
    // non-copy constructor to allocate the proper amount of memory.

    // The next step is to copy the data.
    for (size_t i = 0; i < mWidth; i++) {
        for (size_t j = 0; j < mHeight; j++) {
            mCells[i][j] = src.at(i, j);
        }
    }
}

template <typename T>
template <typename E>
Grid<T>& Grid<T>::operator=(const Grid<E>& rhs)
{
    // no need to check for self-assignment because this version of
    // assignment is never called when T and E are the same

    // Copy-and-swap idiom
    Grid<T> temp(rhs); // Do all the work in a temporary instance
    swap(temp); // Commit the work with only non-throwing operations
    return *this;
}

template <typename T>
class Grid
{
    // Class definition omitted for brevity
}
#include "GridDefinitions.h"

template <typename T, size_t MAX_WIDTH = 10>
class Grid {
    ...
}

std::pair<int, double> pair1(1, 2.3);

std::pair pair3(1, 2.3);

// 必须包含原template class声明
#include "Grid.h"

// 必须要有!
template <>
class Grid<const char*>
{
    public:
        explicit Grid(size_t width = kDefaultWidth,
            size_t height = kDefaultHeight);
        virtual ~Grid() = default;

        // Explicitly default a copy constructor and assignment operator.
        Grid(const Grid& src) = default;
        Grid<const char*>& operator=(const Grid& rhs) = default;

        // Explicitly default a move constructor and assignment operator.
        Grid(Grid&& src) = default;
        Grid<const char*>& operator=(Grid&& rhs) = default;

        std::optional<std::string>& at(size_t x, size_t y);
        const std::optional<std::string>& at(size_t x, size_t y) const;

        size_t getHeight() const { return mHeight; }
        size_t getWidth() const { return mWidth; }

        static const size_t kDefaultWidth = 10;
        static const size_t kDefaultHeight = 10;

    private:
        void verifyCoordinate(size_t x, size_t y) const;

        std::vector<std::vector<std::optional<std::string>>> mCells;
        size_t mWidth, mHeight;
};

#include "Grid.h"

template <typename T>
class GameBoard : public Grid<T>
{
    public:
        explicit GameBoard(size_t width = Grid<T>::kDefaultWidth,
            size_t height = Grid<T>::kDefaultHeight);
        void move(size_t xSrc, size_t ySrc, size_t xDest, size_t yDest);
};

template <typename T>
GameBoard<T>::GameBoard(size_t width, size_t height)
    : Grid<T>(width, height)
{
}

template <typename T>
void GameBoard<T>::move(size_t xSrc, size_t ySrc, size_t xDest, size_t yDest)
{
    // 注意这里, 调用at方法时显式指名了父类
    Grid<T>::at(xDest, yDest) = std::move(Grid<T>::at(xSrc, ySrc));
    Grid<T>::at(xSrc, ySrc).reset();  // Reset source cell
    // Or:
    // this->at(xDest, yDest) = std::move(this->at(xSrc, ySrc));
    // this->at(xSrc, ySrc).reset();
}

extern void ThirdPartyLibraryMethod(char* str);

void f(const char* str)
{
    ThirdPartyLibraryMethod(const_cast<char*>(str));
}

std::string str = "C++";
const std::string& constStr = std::as_const(str);

class Base {
public:
    virtual ~Base() = default;    
}

class Derived : public Base {
public:
    virtual ~Derived() = default;
}

int main() {
    Base* b;
    Derived* d = new Derived;
    b = d;
    d = static_cast<Derived*>(b);
    
    Derived derived;
    Base& br = derived;
    Derived& dr = static_cast<Derived&>(br);
}

Base* b = new Base{};
Derived* d = static_cast<Derived*>(b);

class A {};
class B {};

int main() {
    A a;
    B b;
    A* ap = &a;
    B* bp = &b;
    
    // 隐式转换
    void* p = ap;
    // 此时需要使用reinterpret_cast显式转换
    ap = reinterpret_cast<A*>(p);
    
    // 对于毫不相关类型的引用转换也可以使用reinterpret_cast
    A& ar = a;
    B& br = reinterpret_cast<B&>(ar);
    return 0;
}

class Base {
public:
    virtual ~Base() = default;    
}

class Derived : public Base {
public:
    virtual ~Derived() = default;
}

int main() {
    // 正常的情况
    Derived* d = new Derived();
    Base* b = d
    d = dynamic_cast<Derived*>(b);
    
    // 错误的情况
    Base b;
    Base& br = b;
    try {
        Derived& dr = dynamic_cast<Derived&>(br);
    } catch (std::bad_cast&) {
        cout << "bad cast" << endl;
    }
    
    return 0;
}

// 开启channel，并且设置为非阻塞模式
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
// 绑定端口
serverSocketChannel.bind(new InetSocketAddress(8080));
// 创建selector
Selector selector = Selector.open();
// 将selector和channel关联，这里我们关心从客户端到来的新连接，所以这里设置为ACCEPT
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
while (selector.select() > 0) {
    Set<SelectionKey> selectionKeys = selector.selectedKeys();
    Iterator<SelectionKey> iterator = selectionKeys.iterator();
    while (iterator.hasNext()) {
        SelectionKey key = iterator.next();
        if (key.isAcceptable()) {
            // ...
        } else if (key.isReadable()) {
            // ...
        } // 省略...
    }
}

Selector selector = initServer();
while (selector.select() > 0) {
    Set<SelectionKey> selectionKeys = selector.selectedKeys();
    Iterator<SelectionKey> iterator = selectionKeys.iterator();
    while (iterator.hasNext()) {
        SelectionKey key = iterator.next();
        if (key.isAcceptable()) {
            ServerSocketChannel channel = (ServerSocketChannel) key.channel();
            SocketChannel clientChannel = channel.accept();
            System.out.println("建立连接：" + clientChannel.getRemoteAddress());
            clientChannel.configureBlocking(false);
            clientChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
        } else if (key.isReadable()) {
            SocketChannel socketChannel = (SocketChannel) key.channel();
            ByteBuffer buffer = (ByteBuffer) key.attachment();
            buffer.clear();
            int read = 0;
            StringBuilder sb = new StringBuilder();
            try {
                while ((read = socketChannel.read(buffer)) > 0) {
                    buffer.flip();
                    sb.append(StandardCharsets.UTF_8.decode(buffer));
                    buffer.clear();
                }
                System.out.printf("收到客户端-(%s) 发来的消息：%s\n", socketChannel.getRemoteAddress(), sb.toString());
                // 返回相同的消息
                buffer.put(("收到，你发来的是：" + sb + "\r\n").getBytes());
                buffer.flip();

                socketChannel.write(buffer);
                socketChannel.register(selector, SelectionKey.OP_READ, buffer.clear());
            } catch (IOException e) {
                e.printStackTrace();
                key.cancel();
                key.channel().close();
            }
        }
        iterator.remove();
    }
}

SocketChannel channel = SocketChannel.open();
channel.connect(new InetSocketAddress("127.0.0.1", 8080));
channel.configureBlocking(false);
Selector selector = Selector.open();
ByteBuffer buffer = ByteBuffer.allocate(1024);
buffer.put(("你好，我是client：" + channel.getLocalAddress() + "\r\n").getBytes());
buffer.flip();
channel.write(buffer);
buffer.clear();
channel.register(selector, OP_READ, buffer);
while (selector.select() > 0) {
    Set<SelectionKey> selectionKeys = selector.selectedKeys();
    Iterator<SelectionKey> iterator = selectionKeys.iterator();
    while (iterator.hasNext()) {
        SelectionKey key = iterator.next();
        if (key.isReadable()) {
            SocketChannel socketChannel = (SocketChannel) key.channel();
            ByteBuffer byteBuffer = (ByteBuffer) key.attachment();
            byteBuffer.clear();
            StringBuilder sb = new StringBuilder();
            while (socketChannel.read(byteBuffer) > 0) {
                byteBuffer.flip();
                sb.append(StandardCharsets.UTF_8.decode(byteBuffer));
                byteBuffer.clear();
            }
            System.out.println("接收到server端的消息：" + sb);

            buffer.put(("你好，我是client：" + channel.getLocalAddress() + "\r\n").getBytes());
            buffer.flip();
            channel.write(buffer);
            buffer.clear();
            channel.register(selector, OP_READ, buffer);
        }
        iterator.remove();
        TimeUnit.MILLISECONDS.sleep(100);
    }
}

收到客户端-(/127.0.0.1:61139) 发来的消息：你好，我是client：/127.0.0.1:61139

收到客户端-(/127.0.0.1:61139) 发来的消息：你好，我是client：/127.0.0.1:61139

收到客户端-(/127.0.0.1:61139) 发来的消息：你好，我是client：/127.0.0.1:61139

接收到server端的消息：收到，你发来的是：你好，我是client：/127.0.0.1:61139


接收到server端的消息：收到，你发来的是：你好，我是client：/127.0.0.1:61139


接收到server端的消息：收到，你发来的是：你好，我是client：/127.0.0.1:61139

20083 socket(AF_INET, SOCK_STREAM, IPPROTO_IP) = 4
20084 setsockopt(4, SOL_SOCKET, SO_REUSEADDR, [1], 4) = 0
20085 clock_gettime(CLOCK_MONOTONIC, {tv_sec=242065, tv_nsec=887240727}) = 0
20086 fcntl(4, F_GETFL)                       = 0x2 (flags O_RDWR)
20087 fcntl(4, F_SETFL, O_RDWR|O_NONBLOCK)    = 0
20088 bind(4, {sa_family=AF_INET, sin_port=htons(9999), sin_addr=inet_addr("0.0.0.0")}, 16) = 0
20089 listen(4, 50)                           = 0
20090 getsockname(4, {sa_family=AF_INET, sin_port=htons(9999), sin_addr=inet_addr("0.0.0.0")}, [16]) = 0
20091 getsockname(4, {sa_family=AF_INET, sin_port=htons(9999), sin_addr=inet_addr("0.0.0.0")}, [16]) = 0
20092 epoll_create(256)                       = 7
21100 epoll_ctl(7, EPOLL_CTL_ADD, 4, {EPOLLIN, {u32=4, u64=158913789956}}) = 0
21101 epoll_wait(7, [{EPOLLIN, {u32=4, u64=158913789956}}], 8192, -1) = 1

if (t.getAndAdd(1) % 20 == 0) {
    TimeUnit.MILLISECONDS.sleep(1000);
}

int nThread = 100;
for (int i = 0; i < nThread; i++) {
    new Thread(() -> {
        try {
            client(); // 这里做连接server以及发消息
        } catch (Exception e) {
            e.printStackTrace();
        }
    }).start();
}

接收到server端的消息：收到，你发来的是：你好，我是client：/127.0.0.1:62221,距离上次：5024
接收到server端的消息：收到，你发来的是：你好，我是client：/127.0.0.1:62256,距离上次：5025
接收到server端的消息：收到，你发来的是：你好，我是client：/127.0.0.1:62199,距离上次：5025

executorService.execute(() -> {
    SocketChannel socketChannel = (SocketChannel) key.channel();
    if (!key.channel().isOpen()) return;
    try {
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        buffer.clear();
        int read = 0;
        StringBuilder sb = new StringBuilder();
        while ((read = socketChannel.read(buffer)) > 0) {
            buffer.flip();
            sb.append(StandardCharsets.UTF_8.decode(buffer));
            buffer.clear();
        }
        System.out.printf("收到客户端-(%s) 发来的消息：%s\n", socketChannel.getRemoteAddress(), sb.toString());
        // 返回相同的消息
        buffer = ByteBuffer.allocate(1024);
        buffer.put(("收到，你发来的是：" + sb + "\r\n").getBytes());
        buffer.flip();
        TimeUnit.MILLISECONDS.sleep(1000);
        socketChannel.write(buffer);
        socketChannel.register(selector, SelectionKey.OP_READ, buffer.clear());
    } catch (Exception e) {
        e.printStackTrace();
        try {
            key.cancel();
            key.channel().close();
        } catch (Exception ioException) {
            ioException.printStackTrace();
        }
    }
});

static class Reactor {
    private Selector selector;

    private void initServer() {
            try {
                ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
                serverSocketChannel.configureBlocking(false);
                serverSocketChannel.bind(new InetSocketAddress(8080));
                selector = Selector.open();
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, new Acceptor());
            } catch (IOException e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
    }

    public void start() throws IOException {
        initServer();
        while (selector.select() > 0) {
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                Handler handler = (Handler) key.attachment();
                if (key.isValid() && handler != null) {
                    handler.process(key);
                }
                iterator.remove();
            }
        }
    }

    interface Handler {
        void process(SelectionKey key);
    }

    class Acceptor implements Handler {

        @Override
        public void process(SelectionKey key) {
            try {
                ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
                SocketChannel clientChannel = serverChannel.accept();
                clientChannel.configureBlocking(false);
                key.attach(new ProcessHandler());
                clientChannel.register(selector, SelectionKey.OP_READ);
                System.out.println("新连接：" + clientChannel.getRemoteAddress());
            } catch (IOException e) {
                key.cancel();
                try {
                    key.channel().close();
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
            }
        }
    }

    class ProcessHandler implements Handler {

        @Override
        public void process(SelectionKey key) {
            // 处理逻辑
        }
    }
}

synchronized(obj) {
    synchronized (obj) {
        synchronized (obj) {
            // ...
        }
    }
}

public void acquire() throws InterruptedException {
	sync.acquireSharedInterruptibly(1);
}

public final void acquireSharedInterruptibly(int arg)
		throws InterruptedException {
	if (Thread.interrupted())
		throw new InterruptedException();
	if (tryAcquireShared(arg) < 0)
		doAcquireSharedInterruptibly(arg);
}

protected int tryAcquireShared(int acquires) {
	return nonfairTryAcquireShared(acquires);
}

final int nonfairTryAcquireShared(int acquires) {
	for (;;) {
		int available = getState();
		int remaining = available - acquires;
		if (remaining < 0 ||
			compareAndSetState(available, remaining))
			return remaining;
	}
}

if (tryAcquireShared(arg) < 0)
	doAcquireSharedInterruptibly(arg);

private void doAcquireSharedInterruptibly(int arg)
	throws InterruptedException {
	final Node node = addWaiter(Node.SHARED); // @1
	boolean failed = true;
	try {
		for (;;) {
			final Node p = node.predecessor(); // @2
			if (p == head) {
				int r = tryAcquireShared(arg);
				if (r >= 0) {
					setHeadAndPropagate(node, r);
					p.next = null; // help GC
					failed = false;
					return;
				}
			}
			if (shouldParkAfterFailedAcquire(p, node) &&
				parkAndCheckInterrupt())
				throw new InterruptedException();
		}
	} finally {
		if (failed)
			cancelAcquire(node);
	}
}

private void setHeadAndPropagate(Node node, int propagate) {
	Node h = head; 
	// 更新head，相当于出队
	setHead(node);
	// 当1.剩余信号量大于0
	// 2.当前head是PROPAGATE或SIGNAL
	// 3.当前节点是SIGNAL
	// 唤醒后继节点（如果有的话）
	if (propagate > 0 || h == null || h.waitStatus < 0 ||
		(h = head) == null || h.waitStatus < 0) {
		Node s = node.next;
		if (s == null || s.isShared())
			doReleaseShared();
	}
}

private void doReleaseShared() {
	for (;;) {
		// 获取当前head
		Node h = head;
		// 队列中存在等待节点的话进入下面的逻辑
		if (h != null && h != tail) {
			int ws = h.waitStatus;
			// 如果是SIGNAL代表有节点等待被唤醒
			if (ws == Node.SIGNAL) {
				// 尝试CAS，如果失败，那么可能其他线程也执行到了这里，重试
				if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
					continue;
				// 唤醒后继节点	
				unparkSuccessor(h);
			}
			// 如果ws==0，意味着后面可能存在节点在尝试获取但是可能还没有
			// 更新前驱为SIGNAL，此时为了快速唤醒线程，更新为PROPAGATE
			else if (ws == 0 &&
					 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
				continue;
		}
		if (h == head) 
			break;
	}
}