func (c *controller) processLoop() {
	for {
    // Pop出Object元素
		obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
		if err != nil {
			if err == ErrFIFOClosed {
				return
			}
			if c.config.RetryOnError {
				// 重新進隊列
				c.config.Queue.AddIfNotPresent(obj)
			}
		}
	}
}
// 去查看Pop的具體實現(xiàn) 點進Pop 找到fifo.go
func (f *FIFO) Pop(process PopProcessFunc) (interface{}, error) {
	f.lock.Lock()
	defer f.lock.Unlock()
	for {
		// 調(diào)用process去處理item，然后返回
		item, ok := f.items[id]
		delete(f.items, id)
		err := process(item)
		return item, err
	}
}
// 然后去查一下 PopProcessFunc 的定義，在創(chuàng)建controller前 share_informer.go的Run()里面
cfg := &Config{
		Process:           s.HandleDeltas,
	}
func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
	s.blockDeltas.Lock()
	defer s.blockDeltas.Unlock()
	for _, d := range obj.(Deltas) {
		switch d.Type {
    // 增、改、替換、同步
		case Sync, Replaced, Added, Updated:
			s.cacheMutationDetector.AddObject(d.Object)
      // 先去indexer查詢
			if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {
        // 如果數(shù)據(jù)已經(jīng)存在，就執(zhí)行Update邏輯
				if err := s.indexer.Update(d.Object); err != nil {
					return err
				}
				isSync := false
				switch {
				case d.Type == Sync:
					isSync = true
				case d.Type == Replaced:
					if accessor, err := meta.Accessor(d.Object); err == nil {
							isSync = accessor.GetResourceVersion() == oldAccessor.GetResourceVersion()
						}
					}
				}
      	// 分發(fā)Update事件
				s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)
			} else {
      	// 沒查到數(shù)據(jù)，就執(zhí)行Add操作
				if err := s.indexer.Add(d.Object); err != nil {
					return err
				}
      	// 分發(fā) Add 事件
				s.processor.distribute(addNotification{newObj: d.Object}, false)
			}
   	// 刪除
		case Deleted:
    	// 去indexer刪除
			if err := s.indexer.Delete(d.Object); err != nil {
				return err
			}
    	// 分發(fā) delete 事件
			s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
		}
	}
	return nil
}

// 我們?nèi)タ纯碔ndex是怎么創(chuàng)建的
func NewSharedIndexInformer(lw ListerWatcher, exampleObject runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer {
	realClock := &clock.RealClock{}
	sharedIndexInformer := &sharedIndexInformer{
		processor:                       &sharedProcessor{clock: realClock},
    // indexer 的初始化
		indexer:                         NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers),
		listerWatcher:                   lw,
		objectType:                      exampleObject,
		resyncCheckPeriod:               defaultEventHandlerResyncPeriod,
		defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod,
		cacheMutationDetector:           NewCacheMutationDetector(fmt.Sprintf("%T", exampleObject)),
		clock:                           realClock,
	}
	return sharedIndexInformer
}
// 生成一個map和func組合而成的Indexer
func NewIndexer(keyFunc KeyFunc, indexers Indexers) Indexer {
	return &cache{
		cacheStorage: NewThreadSafeStore(indexers, Indices{}),
		keyFunc:      keyFunc,
}
// ThreadSafeStore的底層是一個并發(fā)安全的map，具體實現(xiàn)我們暫不考慮
func NewThreadSafeStore(indexers Indexers, indices Indices) ThreadSafeStore {
	return &threadSafeMap{
		items:    map[string]interface{}{},
		indexers: indexers,
		indices:  indices,
	}
}

// 在上面的Process代碼中，我們看到了將數(shù)據(jù)存儲到Indexer后，調(diào)用了一個分發(fā)的函數(shù)
s.processor.distribute()
// 分發(fā)process的創(chuàng)建
func NewSharedIndexInformer() SharedIndexInformer {
	sharedIndexInformer := &sharedIndexInformer{
		processor:                       &sharedProcessor{clock: realClock},
	}
	return sharedIndexInformer
}
// sharedProcessor的結(jié)構(gòu)
type sharedProcessor struct {
	listenersStarted bool
 	// 讀寫鎖
	listenersLock    sync.RWMutex
  // 普通監(jiān)聽列表
	listeners        []*processorListener
  // 同步監(jiān)聽列表
	syncingListeners []*processorListener
	clock            clock.Clock
	wg               wait.Group
}
// 查看distribute函數(shù)
func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
	p.listenersLock.RLock()
	defer p.listenersLock.RUnlock()
	// 將object分發(fā)到 同步監(jiān)聽 或者 普通監(jiān)聽 的列表
	if sync {
		for _, listener := range p.syncingListeners {
			listener.add(obj)
		}
	} else {
		for _, listener := range p.listeners {
			listener.add(obj)
		}
	}
}
// 這個add的操作是利用了channel
func (p *processorListener) add(notification interface{}) {
	p.addCh <- notification
}

// 在setup()中找到scheduler
// 在運行 kube-scheduler 的初期，我們創(chuàng)建了一個Scheduler的數(shù)據(jù)結(jié)構(gòu)，回頭再看看有什么和pod調(diào)度算法相關(guān)的
type Scheduler struct {
	SchedulerCache internalcache.Cache
	Algorithm core.ScheduleAlgorithm
	// 獲取下一個需要調(diào)度的Pod
	NextPod func() *framework.QueuedPodInfo
	Error func(*framework.QueuedPodInfo, error)
	StopEverything <-chan struct{}
	// 等待調(diào)度的Pod隊列，我們重點看看這個隊列是什么
	SchedulingQueue internalqueue.SchedulingQueue
	Profiles profile.Map
	scheduledPodsHasSynced func() bool
	client clientset.Interface
}
// Scheduler的實例化函數(shù) 在最新的版本中少了create這一層 直接是進行里面的邏輯
func New(){
  var sched *Scheduler
	switch {
  // 從 Provider 創(chuàng)建
	case source.Provider != nil:
		sc, err := configurator.createFromProvider(*source.Provider)
		sched = sc
  // 從文件或者ConfigMap中創(chuàng)建
	case source.Policy != nil:
		sc, err := configurator.createFromConfig(*policy)
		sched = sc
	default:
		return nil, fmt.Errorf("unsupported algorithm source: %v", source)
	}
}
// 兩個創(chuàng)建方式，底層都是調(diào)用的 create 函數(shù)
func (c *Configurator) createFromProvider(providerName string) (*Scheduler, error) {
	return c.create()
}
func (c *Configurator) createFromConfig(policy schedulerapi.Policy) (*Scheduler, error){
	return c.create()
}
func (c *Configurator) create() (*Scheduler, error) {
	// 實例化 podQueue
	podQueue := internalqueue.NewSchedulingQueue(
		lessFn,
		internalqueue.WithPodInitialBackoffDuration(time.Duration(c.podInitialBackoffSeconds)*time.Second),
		internalqueue.WithPodMaxBackoffDuration(time.Duration(c.podMaxBackoffSeconds)*time.Second),
		internalqueue.WithPodNominator(nominator),
	)
	return &Scheduler{
		SchedulerCache:  c.schedulerCache,
		Algorithm:       algo,
		Profiles:        profiles,
    // NextPod 函數(shù)依賴于 podQueue
		NextPod:         internalqueue.MakeNextPodFunc(podQueue),
		Error:           MakeDefaultErrorFunc(c.client, c.informerFactory.Core().V1().Pods().Lister(), podQueue, c.schedulerCache),
		StopEverything:  c.StopEverything,
    // 調(diào)度隊列被賦值為podQueue
		SchedulingQueue: podQueue,
	}, nil
}
// 再看看這個調(diào)度隊列的初始化函數(shù)，點進去podQueue，從命名可以看到是一個優(yōu)先隊列，它的實現(xiàn)細節(jié)暫不細看
// 結(jié)合實際情況思考下，pod會有重要程度的區(qū)分，所以調(diào)度的順序需要考慮優(yōu)先級的
func NewSchedulingQueue(lessFn framework.LessFunc, opts ...Option) SchedulingQueue {
	return NewPriorityQueue(lessFn, opts...)
}

// 在上面實例化Scheduler后，有個注冊事件 Handler 的函數(shù)：addAllEventHandlers(sched, informerFactory, podInformer)  informer接到消息之后觸發(fā)對應(yīng)的Handler
func addAllEventHandlers(
	sched *Scheduler,
	informerFactory informers.SharedInformerFactory,
	podInformer coreinformers.PodInformer,
) {
	/*
	函數(shù)前后有很多注冊的Handler，但是和未調(diào)度pod添加到隊列相關(guān)的，只有這個
	*/
	podInformer.Informer().AddEventHandler(
		cache.FilteringResourceEventHandler{
      // 定義過濾函數(shù)：必須為未調(diào)度的pod
			FilterFunc: func(obj interface{}) bool {
				switch t := obj.(type) {
				case *v1.Pod:
					return !assignedPod(t) && responsibleForPod(t, sched.Profiles)
				case cache.DeletedFinalStateUnknown:
					if pod, ok := t.Obj.(*v1.Pod); ok {
						return !assignedPod(pod) && responsibleForPod(pod, sched.Profiles)
					}
					utilruntime.HandleError(fmt.Errorf("unable to convert object %T to *v1.Pod in %T", obj, sched))
					return false
				default:
					utilruntime.HandleError(fmt.Errorf("unable to handle object in %T: %T", sched, obj))
					return false
				}
			},
     	// 增改刪三個操作對應(yīng)的Handler，操作到對應(yīng)的Queue
			Handler: cache.ResourceEventHandlerFuncs{
				AddFunc:    sched.addPodToSchedulingQueue,
				UpdateFunc: sched.updatePodInSchedulingQueue,
				DeleteFunc: sched.deletePodFromSchedulingQueue,
			},
		},
	)
}
// 牢記我們第一階段要分析的對象：create nginx pod，所以進入這個add的操作，對應(yīng)加入到隊列
func (sched *Scheduler) addPodToSchedulingQueue(obj interface{}) {
	pod := obj.(*v1.Pod)
	klog.V(3).Infof("add event for unscheduled pod %s/%s", pod.Namespace, pod.Name)
  // 加入到隊列
	if err := sched.SchedulingQueue.Add(pod); err != nil {
		utilruntime.HandleError(fmt.Errorf("unable to queue %T: %v", obj, err))
	}
}
// 在實例化Scheduler的地方 
// 入隊操作我們清楚了，那出隊呢？我們回過頭去看看上面定義的NextPod的方法實現(xiàn)
func MakeNextPodFunc(queue SchedulingQueue) func() *framework.QueuedPodInfo {
	return func() *framework.QueuedPodInfo {
    // 從隊列中彈出
		podInfo, err := queue.Pop()
		if err == nil {
			klog.V(4).Infof("About to try and schedule pod %v/%v", podInfo.Pod.Namespace, podInfo.Pod.Name)
			return podInfo
		}
		klog.Errorf("Error while retrieving next pod from scheduling queue: %v", err)
		return nil
	}
}

// 了解入隊和出隊操作后，我們看一下Scheduler運行的過程
func (sched *Scheduler) Run(ctx context.Context) {
	if !cache.WaitForCacheSync(ctx.Done(), sched.scheduledPodsHasSynced) {
		return
	}
	sched.SchedulingQueue.Run()
  // 調(diào)度一個pod對象
	wait.UntilWithContext(ctx, sched.scheduleOne, 0)
	sched.SchedulingQueue.Close()
}
// 接下來scheduleOne方法代碼很長，我們一步一步來看
func (sched *Scheduler) scheduleOne(ctx context.Context) {
  // podInfo 就是從隊列中獲取到的pod對象
	podInfo := sched.NextPod()
	// 檢查pod的有效性
	if podInfo == nil || podInfo.Pod == nil {
		return
	}
	pod := podInfo.Pod
  // 根據(jù)定義的 pod.Spec.SchedulerName 查到對應(yīng)的profile
	prof, err := sched.profileForPod(pod)
	if err != nil {
		klog.Error(err)
		return
	}
  // 可以跳過調(diào)度的情況，一般pod進不來
	if sched.skipPodSchedule(prof, pod) {
		return
	}
  // 調(diào)用調(diào)度算法，獲取結(jié)果
	scheduleResult, err := sched.Algorithm.Schedule(schedulingCycleCtx, prof, state, pod)
	if err != nil {
		/*
		出現(xiàn)調(diào)度失敗的情況：
		這個時候可能會觸發(fā)搶占preempt，搶占是一套復(fù)雜的邏輯，后面我們專門會講
		目前假設(shè)各類資源充足，能正常調(diào)度
		*/
	}
	metrics.SchedulingAlgorithmLatency.Observe(metrics.SinceInSeconds(start))
  // assumePod 是假設(shè)這個Pod按照前面的調(diào)度算法分配后，進行驗證
	assumedPodInfo := podInfo.DeepCopy()
	assumedPod := assumedPodInfo.Pod
	// SuggestedHost 為建議的分配的Host
	err = sched.assume(assumedPod, scheduleResult.SuggestedHost)
	if err != nil {
		// 失敗就重新分配，不考慮這種情況 
	}
	// 運行相關(guān)插件的代碼先跳過 比如一些搶占插件
	// 異步綁定pod
	go func() {
		// 有一系列的檢查工作
                // 真正做綁定的動作
		err := sched.bind(bindingCycleCtx, prof, assumedPod, scheduleResult.SuggestedHost, state)
		if err != nil {
			// 錯誤處理，清除狀態(tài)并重試
		} else {
			// 打印結(jié)果，調(diào)試時將log level調(diào)整到2以上
			if klog.V(2).Enabled() {
				klog.InfoS("Successfully bound pod to node", "pod", klog.KObj(pod), "node", scheduleResult.SuggestedHost, "evaluatedNodes", scheduleResult.EvaluatedNodes, "feasibleNodes", scheduleResult.FeasibleNodes)
			}
      // metrics中記錄相關(guān)的監(jiān)控指標
			metrics.PodScheduled(prof.Name, metrics.SinceInSeconds(start))
			metrics.PodSchedulingAttempts.Observe(float64(podInfo.Attempts))
      metrics.PodSchedulingDuration.WithLabelValues(getAttemptsLabel(podInfo)).Observe(metrics.SinceInSeconds(podInfo.InitialAttemptTimestamp))
			// 運行綁定后的插件
			prof.RunPostBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
		}
	}()
}

// 調(diào)用算法下的Schedule
func New(){
  scheduleResult, err := sched.Algorithm.Schedule(schedulingCycleCtx, prof, state, pod)
}
func (c *Configurator) create() (*Scheduler, error) {
  algo := core.NewGenericScheduler(
		c.schedulerCache,
		c.nodeInfoSnapshot,
		extenders,
		c.informerFactory.Core().V1().PersistentVolumeClaims().Lister(),
		c.disablePreemption,
		c.percentageOfNodesToScore,
	)
  return &Scheduler{
		Algorithm:       algo,
	}, nil
}
// genericScheduler 的 Schedule 的實現(xiàn)
func (g *genericScheduler) Schedule(ctx context.Context, prof *profile.Profile, state *framework.CycleState, pod *v1.Pod) (result ScheduleResult, err error) {
	// 對 pod 進行 pvc 的信息檢查
	if err := podPassesBasicChecks(pod, g.pvcLister); err != nil {
		return result, err
	}
	// 對當前的信息做一個快照
	if err := g.snapshot(); err != nil {
		return result, err
	}
	// Node 節(jié)點數(shù)量為0，表示無可用節(jié)點
	if g.nodeInfoSnapshot.NumNodes() == 0 {
		return result, ErrNoNodesAvailable
	}
  // Predict階段：找到所有滿足調(diào)度條件的節(jié)點feasibleNodes，不滿足的就直接過濾
	feasibleNodes, filteredNodesStatuses, err := g.findNodesThatFitPod(ctx, prof, state, pod)
	// 沒有可用節(jié)點直接報錯
	if len(feasibleNodes) == 0 {
		return result, &FitError{
			Pod:                   pod,
			NumAllNodes:           g.nodeInfoSnapshot.NumNodes(),
			FilteredNodesStatuses: filteredNodesStatuses,
		}
	}
	// 只有一個節(jié)點就直接選用
	if len(feasibleNodes) == 1 {
		return ScheduleResult{
			SuggestedHost:  feasibleNodes[0].Name,
			EvaluatedNodes: 1 + len(filteredNodesStatuses),
			FeasibleNodes:  1,
		}, nil
	}
	// Priority階段：通過打分，找到一個分數(shù)最高、也就是最優(yōu)的節(jié)點
	priorityList, err := g.prioritizeNodes(ctx, prof, state, pod, feasibleNodes)
	host, err := g.selectHost(priorityList)
	return ScheduleResult{
		SuggestedHost:  host,
		EvaluatedNodes: len(feasibleNodes) + len(filteredNodesStatuses),
		FeasibleNodes:  len(feasibleNodes),
	}, err
}
/*
Predict 和 Priority 是選擇調(diào)度節(jié)點的兩個關(guān)鍵性步驟， 它的底層調(diào)用了各種algorithm算法。我們暫時不細看。
以我們前面講到過的 NodeName 算法為例，節(jié)點必須與 NodeName 匹配，它是屬于Predict階段的。
在新版本中 這部分算法的實現(xiàn)放到了extenders，邏輯是一樣的
*/

func (sched *Scheduler) assume(assumed *v1.Pod, host string) error {
  // 將 host 填入到 pod spec字段的nodename，假定分配到對應(yīng)的節(jié)點上
	assumed.Spec.NodeName = host
  // 調(diào)用 SchedulerCache 下的 AssumePod
	if err := sched.SchedulerCache.AssumePod(assumed); err != nil {
		klog.Errorf("scheduler cache AssumePod failed: %v", err)
		return err
	}
	if sched.SchedulingQueue != nil {
		sched.SchedulingQueue.DeleteNominatedPodIfExists(assumed)
	}
	return nil
}
// 回頭去找 SchedulerCache 初始化的地方
func (c *Configurator) create() (*Scheduler, error) {
	return &Scheduler{
		SchedulerCache:  c.schedulerCache,
	}, nil
}
func New() (*Scheduler, error) {
  // 這里就是初始化的實例 schedulerCache
	schedulerCache := internalcache.New(30*time.Second, stopEverything)
	configurator := &Configurator{
		schedulerCache:           schedulerCache,
	}
}
// 看看AssumePod做了什么
func (cache *schedulerCache) AssumePod(pod *v1.Pod) error {
  // 獲取 pod 的 uid
	key, err := framework.GetPodKey(pod)
	if err != nil {
		return err
	}
	// 加鎖操作，保證并發(fā)情況下的一致性
	cache.mu.Lock()
	defer cache.mu.Unlock()
      // 根據(jù) uid 找不到 pod 當前的狀態(tài)  看看被調(diào)度了沒有 
	if _, ok := cache.podStates[key]; ok {
		return fmt.Errorf("pod %v is in the cache, so can't be assumed", key)
	}
  // 把 Assume Pod 的信息放到對應(yīng) Node 節(jié)點中
	cache.addPod(pod)
  // 把 pod 狀態(tài)設(shè)置為 Assume 成功
	ps := &podState{
		pod: pod,
	}
	cache.podStates[key] = ps
	cache.assumedPods[key] = true
	return nil
}

func (sched *Scheduler) bind(ctx context.Context, prof *profile.Profile, assumed *v1.Pod, targetNode string, state *framework.CycleState) (err error) {
	start := time.Now()
  // 把 assumed 的 pod 信息保存下來
	defer func() {
		sched.finishBinding(prof, assumed, targetNode, start, err)
	}()
	// 階段1： 運行擴展綁定進行驗證，如果已經(jīng)綁定報錯
	bound, err := sched.extendersBinding(assumed, targetNode)
	if bound {
		return err
	}
  // 階段2：運行綁定插件驗證狀態(tài)
	bindStatus := prof.RunBindPlugins(ctx, state, assumed, targetNode)
	if bindStatus.IsSuccess() {
		return nil
	}
	if bindStatus.Code() == framework.Error {
		return bindStatus.AsError()
	}
	return fmt.Errorf("bind status: %s, %v", bindStatus.Code().String(), bindStatus.Message())
}

// 這塊的代碼我不做細致的逐層分析了，大家根據(jù)興趣自行探索
func (b DefaultBinder) Bind(ctx context.Context, state *framework.CycleState, p *v1.Pod, nodeName string) *framework.Status {
	klog.V(3).Infof("Attempting to bind %v/%v to %v", p.Namespace, p.Name, nodeName)
	binding := &v1.Binding{
		ObjectMeta: metav1.ObjectMeta{Namespace: p.Namespace, Name: p.Name, UID: p.UID},
		Target:     v1.ObjectReference{Kind: "Node", Name: nodeName},
	}
  // ClientSet就是訪問kube-apiserver的客戶端，將數(shù)據(jù)更新上去
	err := b.handle.ClientSet().CoreV1().Pods(binding.Namespace).Bind(ctx, binding, metav1.CreateOptions{})
	if err != nil {
		return framework.NewStatus(framework.Error, err.Error())
	}
	return nil
}