【提问】DeltaFIFO 消费者方法中 process 回调函数的实现

[linxuyalun]

【提问】DeltaFIFO 消费者方法中 process 回调函数的实现

[luffyao]

我个人理解是，这里的只是做了个多个obj的处理，其实在真正处理的地方，并没有多个obj 传入。具体的实现，可以参考这段代码：(我这个是1.18的，也许会有点小出入，但是应该整体的逻辑是一样的)

func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
	f.lock.Lock()
	defer f.lock.Unlock()
	for {
		for len(f.queue) == 0 {
			// When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
			// When Close() is called, the f.closed is set and the condition is broadcasted.
			// Which causes this loop to continue and return from the Pop().
			if f.IsClosed() {
				return nil, ErrFIFOClosed
			}

			f.cond.Wait()
		}
		id := f.queue[0]
		f.queue = f.queue[1:]
		if f.initialPopulationCount > 0 {
			f.initialPopulationCount--
		}
		item, ok := f.items[id]
		if !ok {
			// Item may have been deleted subsequently.
			continue
		}
		delete(f.items, id)
		err := process(item)
		if e, ok := err.(ErrRequeue); ok {
			f.addIfNotPresent(id, item)
			err = e.Err
		}
		// Don't need to copyDeltas here, because we're transferring
		// ownership to the caller.
		return item, err
	}
}

[lianghao208]

上面的Pop方法的源码有写到，每次传一个 Deltas 进 process 中，而 Deltas 里面有 多个 Delta 对象（每个 Delta 对象对应一个 DeltaType，表示对一个资源对象的多次操作）。所以我认为应该是队列中每个资源对象(Deltas)做一次 process 操作，而每个资源对象(Deltas)会有多个 DeltaType，所以要进行遍历。从最老的 DeltaType 开始遍历到最新的 DeltaType。

type DeltaFIFO struct {
	...
	items map[string]Deltas
	queue []string
	...
}
// Deltas is a list of one or more 'Delta's to an individual object.
// The oldest delta is at index 0, the newest delta is the last one.
type Deltas []Delta

type Delta struct {
	Type   DeltaType
	Object interface{}
}

[luffyao]

嗯 你的理解是对的，我刚看了下代码 确实是个Deltas，是一个 Delta 的 slice。

[linxuyalun]

上面的Pop方法的源码有写到，每次传一个 Deltas 进 process 中，而 Deltas 里面有 多个 Delta 对象（每个 Delta 对象对应一个 DeltaType，表示对一个资源对象的多次操作）。所以我认为应该是队列中每个资源对象(Deltas)做一次 process 操作，而每个资源对象(Deltas)会有多个 DeltaType，所以要进行遍历。从最老的 DeltaType 开始遍历到最新的 DeltaType。

type DeltaFIFO struct {
	...
	items map[string]Deltas
	queue []string
	...
}
// Deltas is a list of one or more 'Delta's to an individual object.
// The oldest delta is at index 0, the newest delta is the last one.
type Deltas []Delta

type Delta struct {
	Type   DeltaType
	Object interface{}
}

对，我的疑问是这里 Deltas 存了一个 Object 的整个操作事件，假设Object1 的 Delta 一开始存储了 {"Added",Obj1}, {"Updated", Obj1}，然后这时候新生产了一个 {"Deleted",Obj1}，存进 Delta，因此这个时候它的内容变为 {"Added",Obj1}, {"Updated", Obj1}，{"Deleted",Obj1}。因此，HandleDeltas 会把这三次事件都 distribute 一遍，但是我理解前两次事件在之前已经执行过了，为什么还要再执行一次呢

[liangyuanpeng]

但是我理解前两次事件在之前已经执行过了，为什么还要再执行一次呢

事件处理了就没有了，不会再次处理,

[lianghao208]

上面的Pop方法的源码有写到，每次传一个 Deltas 进 process 中，而 Deltas 里面有 多个 Delta 对象（每个 Delta 对象对应一个 DeltaType，表示对一个资源对象的多次操作）。所以我认为应该是队列中每个资源对象(Deltas)做一次 process 操作，而每个资源对象(Deltas)会有多个 DeltaType，所以要进行遍历。从最老的 DeltaType 开始遍历到最新的 DeltaType。

type DeltaFIFO struct {
	...
	items map[string]Deltas
	queue []string
	...
}
// Deltas is a list of one or more 'Delta's to an individual object.
// The oldest delta is at index 0, the newest delta is the last one.
type Deltas []Delta

type Delta struct {
	Type   DeltaType
	Object interface{}
}

对，我的疑问是这里 Deltas 存了一个 Object 的整个操作事件，假设Object1 的 Delta 一开始存储了 {"Added",Obj1}, {"Updated", Obj1}，然后这时候新生产了一个 {"Deleted",Obj1}，存进 Delta，因此这个时候它的内容变为 {"Added",Obj1}, {"Updated", Obj1}，{"Deleted",Obj1}。因此，HandleDeltas 会把这三次事件都 distribute 一遍，但是我理解前两次事件在之前已经执行过了，为什么还要再执行一次呢

如果前两次事件已经执行过了，这个 Deltas 就已经从队列Pop出去了（process前面执行了delete(f.items, id)），Delta FIFO里面就不会有{"Added",Obj1}, {"Updated", Obj1}了，时候新生产了一个 {"Deleted",Obj1}，Obj1这个Deltas切片应该只有一个元素。

[linxuyalun]

上面的Pop方法的源码有写到，每次传一个 Deltas 进 process 中，而 Deltas 里面有 多个 Delta 对象（每个 Delta 对象对应一个 DeltaType，表示对一个资源对象的多次操作）。所以我认为应该是队列中每个资源对象(Deltas)做一次 process 操作，而每个资源对象(Deltas)会有多个 DeltaType，所以要进行遍历。从最老的 DeltaType 开始遍历到最新的 DeltaType。

type DeltaFIFO struct {
	...
	items map[string]Deltas
	queue []string
	...
}
// Deltas is a list of one or more 'Delta's to an individual object.
// The oldest delta is at index 0, the newest delta is the last one.
type Deltas []Delta

type Delta struct {
	Type   DeltaType
	Object interface{}
}

对，我的疑问是这里 Deltas 存了一个 Object 的整个操作事件，假设Object1 的 Delta 一开始存储了 {"Added",Obj1}, {"Updated", Obj1}，然后这时候新生产了一个 {"Deleted",Obj1}，存进 Delta，因此这个时候它的内容变为 {"Added",Obj1}, {"Updated", Obj1}，{"Deleted",Obj1}。因此，HandleDeltas 会把这三次事件都 distribute 一遍，但是我理解前两次事件在之前已经执行过了，为什么还要再执行一次呢

如果前两次事件已经执行过了，这个 Deltas 就已经从队列Pop出去了（process前面执行了delete(f.items, id)），Delta FIFO里面就不会有{"Added",Obj1}, {"Updated", Obj1}了，时候新生产了一个 {"Deleted",Obj1}，Obj1这个Deltas切片应该只有一个元素。

啊明白了明白了，谢谢，我理解这个问题了，之前没有注意到这个 delete，陷入僵局……谢谢谢谢

[linxuyalun]

但是我理解前两次事件在之前已经执行过了，为什么还要再执行一次呢

事件处理了就没有了，不会再次处理,

明白了，谢谢～