【源码阅读】事件订阅包v2

代码位置

1、Feed

Feed 实现一对多订阅，其中事件的载体是通道。发送到 Feed 的值会同时传送到所有订阅的通道。

与Typemux的对比
链接: link
TypeMux是一个同步的事件框架，当有一个被订阅的事件发生的时候，会遍历该事件对应的订阅者通道，通知其中的订阅者，但是当订阅者1没有接受该消息的时候，发送进程会被阻塞，会影响对订阅者2的发送。
所以Feed作为流式事件框架，是否是异步的取决于是否有缓存通道，当设计有缓存通道的时候是异步的，否则就是同步的。

type Feed struct {
	once      sync.Once        // ensures that init only runs once
	sendLock  chan struct{}    // sendLock has a one-element buffer and is empty when held.It protects sendCases.
	removeSub chan interface{} // interrupts Send
	sendCases caseList         // the active set of select cases used by Send

	// The inbox holds newly subscribed channels until they are added to sendCases.
	mu    sync.Mutex
	inbox caseList
	etype reflect.Type
}

once字段是用来确保init只会执行一次
sendLock是一个缓存通道，
这是一个名为Feed的结构体，它包含以下字段：
once：sync.Once类型，确保init只运行一次。
sendLock：chan struct{}类型，sendLock具有一个元素缓冲区，当持有时为空。可以保护sendCases。
removeSub：chan interface{}类型，用于中断Send。
sendCases：caseList类型，Send使用的活动选择集。
mu：sync.Mutex类型，用于保护inbox。
inbox：caseList类型，用于存储新订阅的通道，直到它们被添加到sendCases中。
etype：reflect.Type类型，表示结构体的反射类型。

1.1 init

func (f *Feed) init(etype reflect.Type) {
	f.etype = etype
	f.removeSub = make(chan interface{})
	f.sendLock = make(chan struct{}, 1)
	f.sendLock <- struct{}{}
	f.sendCases = caseList{{Chan: reflect.ValueOf(f.removeSub), Dir: reflect.SelectRecv}}
}

初始化一个名为Feed的结构体。该方法接受一个参数etype，表示结构体的反射类型。

将传入的etype赋值给f.etype，表示结构体的类型。
创建一个名为f.removeSub的通道，用于接收移除订阅的通知。
创建一个名为f.sendLock的通道，并设置其缓冲区大小为1，以确保只有一个发送操作可以同时进行。
向f.sendLock通道发送一个空的结构体，以解锁该通道。
创建一个名为f.sendCases的caseList类型的变量，并将其初始化为包含一个元素的列表。这个元素是一个case结构体，其中Chan字段表示要监听的通道（即f.removeSub），Dir字段表示监听的方向（即接收方向）。
初始化一个Feed结构体，并设置其相关属性和状态。

1.2 Subscribe

func (f *Feed) Subscribe(channel interface{}) Subscription

订阅将channel添加到feed中，添加一个订阅。未来的发送将在通道上传送，直到subscription为止。 添加的所有通道必须具有相同的元素类型。
通道应该有足够的缓冲空间以避免阻塞其他订阅者。
慢速订阅者不会被丢弃。
这里大量用到了reflect类型，针对不同类型数据的相同处理，就可以用到反射。反射中的send,recv,default分别对应case中的不同情况。而reflect中还有其他类型，Bool、String 和所有数字类型的基础类型；Array 和 Struct 对应的聚合类型；Chan、Func、Ptr、Slice 和 Map 对应的引用类型。

将channel的值映射给chanval
将类型赋值给chantyp
进行判断是否是通道类型，并且发送方向是否是想要的。
因为对应的case中其实有三种情况：send,recv,default。其中的send和recv中的chan都是channel，如果是default就是0。
chantyp.ChanDir()&reflect.SendDir == 0用来判断是不是发送方向，有一个有错就会报错

if chantyp.Kind() != reflect.Chan || chantyp.ChanDir()&reflect.SendDir == 0 {
		panic(errBadChannel)
	}

创建一个feedSub类型的对象sub，并将其与当前的Feed对象和channel关联起来。同时，初始化一个错误通道err，用于接收可能的错误信息。

sub := &feedSub{feed: f, channel: chanval, err: make(chan error, 1)}

调用Feed对象的init方法，对channel的元素类型进行初始化。如果初始化后的元素类型与预期不符，也会抛出一个错误。
使用互斥锁保护对Feed对象的操作，确保线程安全
将一个新的选择情况添加到inbox中，表示当有数据发送到这个频道时，需要进行相应的处理。
返回创建的sub对象。

1.3 remove

func (f *Feed) remove(sub *feedSub)

首先会从inbox中删除，其中涵盖尚未添加到 f.sendCases 的频道。
该方法的作用是从Feed对象的内部数据结构中移除一个订阅对象（feedSub）。这个方法使用了Go语言的反射机制和并发控制机制来实现多通道的同步发送操作。

ch := sub.channel.Interface()是将订阅对象的通道中的值给变量ch，可以方便后续处理
需要添加锁
从index中查找要移除的订阅对象index := f.inbox.find(ch)（参考3.1），如果可以找到，就将其从index中删除f.inbox = f.inbox.delete(index)（参考3.2），释放锁，并返回

index := f.inbox.find(ch)
if index != -1 {
	f.inbox = f.inbox.delete(index)
	f.mu.Unlock()
	return
}

如果没有找到，就会进入select语句
从ch通道发送给f.removeSub，表示将要执行移除订阅对象
等待sendCases通道的信号，如果有正在发送的操作，这个sendLock就不会为空，就无法获得这个锁，只有没有正在发送的操作的时候才会得到该锁进入执行：
从sendCases中查找该订阅f.sendCases.find(ch)（参考3.1），删除f.sendCases = f.sendCases.delete(f.sendCases.find(ch))（参考3.2）
释放sendLock锁

1.4 Send

func (f *Feed) Send(value interface{}) (nsent int)

Send 同时发送到所有订阅的频道，返回发送到的订阅者数量。

将传入的值转换为反射类型的值rvaluervalue := reflect.ValueOf(value)
用传入的值的类型进行唯一一次的注册（通过f.once来限制）
检查是否符合定义的类型
获取发送锁sendLock，以确保在发送过程中不会发生竞争条件<-f.sendLock
将inbox中的值添加到sendCases列表中，并清空inbox

f.mu.Lock()
f.sendCases = append(f.sendCases, f.inbox...)
f.inbox = nil
f.mu.Unlock()

遍历sendCases列表，将每个通道的发送值设置为rvalue，将一个值发送给所有订阅者
尝试发送，有可能会一次成功trysend，遍历一轮，成功的就从case中删除cases = cases.deactivate(i)（参考3.3），且计数nsent加一
使用reflect.Select(cases)语句，类似case语句，获得返回值chosen, recv, _ := reflect.Select(cases)
如果chosen的值为0，表示通道关闭，就需要找到该对象的位置（参考3.1），并将该对象从sendCases中删除（参考3.2），并且缩小大小，重新赋值给cases

if chosen == 0 /* <-f.removeSub */ {
	index := f.sendCases.find(recv.Interface())
	f.sendCases = f.sendCases.delete(index)
	if index >= 0 && index < len(cases) {
		// Shrink 'cases' too because the removed case was still active.
		cases = f.sendCases[:len(cases)-1]
	}
}

否则就表示重新发送发送成功了，就使用cases.deactivate(i)从cases中删除（参考3.3），并且nsent计数加一
使用f.sendCases[i].Send = reflect.Value{}来清空要发送的数据
向f.sendLock通道发送一个空的结构体，以解锁该通道
返回成功发送的值的数量nsent。

2、feedSub

是Subscription的实现，Subscription中有两个函数：Unsubscribe（取消订阅）和Err（错误）。

type feedSub struct {
	feed    *Feed
	channel reflect.Value
	errOnce sync.Once
	err     chan error
}

feed：一个指向Feed类型的指针。
channel：一个反射值，表示通道类型。
errOnce：一个同步原语，用于确保错误处理只执行一次。
err：一个错误通道，用于传递错误信息。

2.1 Unsubscribe

func (sub *feedSub) Unsubscribe() {
	sub.errOnce.Do(func() {
		sub.feed.remove(sub)
		close(sub.err)
	})
}

确认错误处理只能处理一次
遇到该事件的时候，要使用remove（参考1.3）从订阅对象列表中删除
给出错误信息

2.2 Err

func (sub *feedSub) Err() <-chan error {
	return sub.err
}

从通道中获取错误信息并返回

3、caseList

type caseList []reflect.SelectCase

caseList是反射机制中的selectcase，类似case，但是可以动态添加所有管道

3.1 find

func (cs caseList) find(channel interface{}) int {
	for i, cas := range cs {
		if cas.Chan.Interface() == channel {
			return i
		}
	}
	return -1
}

find返回包含给定通道的事例的索引。

遍历所有caselist
如果能找到case中的Chan与所给定的通道一样的，就返回该case对应的索引index
如果都没有，就返回-1

3.2 delete

func (cs caseList) delete(index int) caseList {
	return append(cs[:index], cs[index+1:]...)
}

从cs列表中删除指定位置的case并返回新的caselist，使用append来进行

3.3 deactivate

func (cs caseList) deactivate(index int) caseList {
	last := len(cs) - 1
	cs[index], cs[last] = cs[last], cs[index]
	return cs[:last]
}

deactivate将索引处的case移动到cs切片的不可访问部分。
先将index和最后一部分last的case进行对换，然后将cs的长度减一，这样就相当于将最后一部分删除了，也就是函数传入的index部分。
-----------------------------subscription.go----------------------------------------

4、Subscription

type Subscription interface {
	Err() <-chan error // returns the error channel
	Unsubscribe()      // cancels sending of events, closing the error channel
}

只有两个方法，一个是取消订阅，而是错误函数
代码注释：

subscription表示一系列事件。事件的载体通常是一个通道，但不是接口的一部分。
建立订阅时可能会失败。故障通过错误通道报告。如果订阅存在问题（例如，传递事件的网络连接已关闭）。只发送一个值。当订阅成功结束时（即，当事件源关闭时），错误通道将关闭。当调用“取消订阅”时，它也会关闭。
Unsubscribe方法取消发送事件。在任何情况下，您都必须调用Unsubscribe以确保与订阅相关的资源得到释放。它可以被调用任意次数。

4.1 NewSubscription

func NewSubscription(producer func(<-chan struct{}) error) Subscription

NewSubscription在新的goroutine中作为订阅运行生产者函数。当调用取消订阅时，提供给制作人的频道将关闭。如果fn返回错误，则在订阅的错误通道上发送。
该函数接受一个参数producer，该参数是一个生产者函数，返回值是error。这个函数的作用是创建一个新的订阅对象，并启动一个协程来执行生产者函数。通过通道将数据发送给消费者。

创建一个名为s的指针变量，指向一个名为funcSub的结构体实例（参考5）
使用make(chan struct{})创建一个名为unsub的通道，用于取消订阅
使用make(chan error, 1)创建一个名为err的错误通道，用于传递错误信息
启动一个协程来执行生产者函数
使用defer close(s.err)确保错误通道在协程结束时被关闭
调用生产者函数producer(s.unsub)，使用的通道是unsub通道，并将返回的错误赋值给变量err
使用s.mu.Lock()和s.mu.Unlock()对结构体实例进行加锁和解锁操作
检查是否已经取消订阅，如果没有取消订阅，则将错误信息发送到错误通道，并将s.unsubscribed设置为true表示已取消订阅
返回创建的订阅对象s。

5、funcSub

封装匿名函数

type funcSub struct {
	unsub        chan struct{}
	err          chan error
	mu           sync.Mutex
	unsubscribed bool
}

一个unsub通道，用来取消订阅
一个错误通道err
一个锁用来保证安全
布尔值来标志是否已经完成了取消订阅

5.1 Unsubscribe

func (s *funcSub) Unsubscribe() {
	s.mu.Lock()
	if s.unsubscribed {
		s.mu.Unlock()
		return
	}
	s.unsubscribed = true
	close(s.unsub)
	s.mu.Unlock()
	// Wait for producer shutdown.
	<-s.err
}

上锁和释放锁用来保证安全，
通过bool unsubscribed来确定是否已经完成了取消订阅
如果已经完成了订阅，就释放锁，并return
否则就修改值为true，关闭unsub通道，并通过s.mu.Unlock()释放锁
开启err通道

5.2 Err

func (s *funcSub) Err() <-chan error {
	return s.err
}

用于返回错误信息

6、resubscribeSub

type resubscribeSub struct {
	fn                   ResubscribeErrFunc
	err                  chan error
	unsub                chan struct{}
	unsubOnce            sync.Once
	lastTry              mclock.AbsTime
	lastSubErr           error
	waitTime, backoffMax time.Duration
}

fn：一个类型为ResubscribeErrFunc的函数
err：一个错误通道，用于接收错误信息
unsub：一个结构体通道，用于取消订阅
unsubOnce：一个sync.Once类型的变量，确保unsub只被执行一次
lastTry：一个mclock.AbsTime类型的变量，表示上次尝试的时间
lastSubErr：一个error类型的变量，表示上次订阅时发生的错误
waitTime：一个time.Duration类型的变量，表示等待时间
backoffMax：一个time.Duration类型的变量，表示最大退避时间

6.1 Resubscribe&ResubscribeErr

Resubscribe会反复调用fn以保持订阅已建立。当订阅建立时，Resubscribe会等待它失败，然后再次调用fn。此过程重复进行，直到调用“取消订阅”或活动订阅成功结束。
重新订阅在对fn的调用之间应用回退。调用之间的时间根据错误率进行调整，但永远不会超过backoffMax。

ResubscribeErr反复调用fn以保持订阅已建立。当订阅建立时，ResubscribeErr等待它失败并再次调用fn。此过程重复进行，直到调用“取消订阅”或活动订阅成功结束。
Resubscribe和ResubscribeErr之间的区别在于，使用ResubscripteErr，回调可以使用失败订阅的错误进行日志记录。
ResubscribeErr在对fn的调用之间应用回退。调用之间的时间根据错误率进行调整，但永远不会超过backoffMax。

type ResubscribeFunc func(context.Context) (Subscription, error)

6.2 ResubscribeErr

func ResubscribeErr(backoffMax time.Duration, fn ResubscribeErrFunc) Subscription {
	s := &resubscribeSub{
		waitTime:   backoffMax / 10,
		backoffMax: backoffMax,
		fn:         fn,
		err:        make(chan error),
		unsub:      make(chan struct{}),
	}
	go s.loop()
	return s
}

创建resubscribeSub对象：

waitTime是等待时间，是backoffMax的十分之一
backoffMax是最大的时间限制
fn是传入的重新订阅函数ResubscribeErrFunc类型

type ResubscribeErrFunc func(context.Context, error) (Subscription, error)

err是错误通道，用于接收错误信息
unsub是取消订阅通道

执行loop函数（参考6.5）
返回创建的s对象

6.3 Unsubscribe

func (s *resubscribeSub) Unsubscribe() {
	s.unsubOnce.Do(func() {
		s.unsub <- struct{}{}
		<-s.err
	})
}

取消订阅

因为unsubOnce sync.Once所以取消订阅只能执行一次
读取信息放入取消订阅通道s.unsub中
开启err通道

6.4 Err

func (s *resubscribeSub) Err() <-chan error {
	return s.err
}

用于返回错误信息

6.5 loop

func (s *resubscribeSub) loop() {
	defer close(s.err)
	var done bool
	for !done {
		sub := s.subscribe()
		if sub == nil {
			break
		}
		done = s.waitForError(sub)
		sub.Unsubscribe()
	}
}

defer close(s.err) 用于延迟关闭一个名为 s.err 的错误通道。当函数执行完成时，defer 关键字会确保 close(s.err) 被调用，从而关闭错误通道并释放相关资源
定义一个bool变量done确定是否完成
只要还没有完成（done为false）就会一直执行for循环
每一次循环中：

使用sub := s.subscribe()返回订阅对象（参考6.6）
如果返回的对象为nil就break跳出循环
否则继续执行done = s.waitForError(sub)（参考6.7）
使用sub.Unsubscribe()取消订阅

6.6 subscribe

func (s *resubscribeSub) subscribe() Subscription

实现一个自动重试订阅的逻辑，当订阅失败时会进行一定的退避等待，并在等待结束后再次尝试订阅。

创建一个名为subscribed的错误通道，用于接收订阅结果
初始化一个名为sub的Subscription变量（参考4）（取消订阅+错误）
进入一个无限循环，每次循环执行以下操作：
- 更新上一次尝试时间s.lastTry为当前时间
- 创建一个上下文ctx和一个取消函数cancel
- 启动一个新的goroutine，调用s.fn(ctx, s.lastSubErr)来获取订阅结果（参考6.2），并将结果赋值给rsub和err。
- 将rsub赋值给sub，并将err信息发送到subscribed通道。
- 使用select语句等待以下情况之一发生：
  - 从subscribed通道接收到信号，取消上下文并检查错误是否为nil。如果是，则返回sub作为订阅结果，没有错误；否则，根据s.backoffWait()的结果决定是否继续尝试订阅（参考6.8）
  - 从s.unsub通道接收到取消信号，取消上下文并，使用<-subscribed接受消息但不使用。然后返回nil表示已取消订阅。
如果循环结束仍未成功订阅，则抛出异常或返回错误信息。

6.7 waitForError

func (s *resubscribeSub) waitForError(sub Subscription) bool

使用defer sub.Unsubscribe()确保函数执行完成的时候会执行取消订阅函数
select语句

当从sub.Err()通道中接收到错误时，将错误赋值给s.lastSubErr，并返回是否有错误
当从s.unsub通道中接收到取消信号时，直接返回true表示已取消

6.8 backoffWait

func (s *resubscribeSub) backoffWait() bool

用于在订阅失败时进行重试等待，这个方法通常用于网络通信中，当订阅失败时，通过退避算法逐渐增加等待时间，直到达到最大等待时间或收到取消信号为止。

判断当前时间与上次尝试的时间差s.lastTry是否大于退避的最大时间s.backoffMax。如果是，则将等待时间设置为退避最大时间的1/10；否则，将等待时间翻倍，但不超过退避最大时间

s.waitTime *= 2
if s.waitTime > s.backoffMax {
	s.waitTime = s.backoffMax
}

创建一个定时器t，并设置等待时间为更新后的等待时间，它会在 s.waitTime 时间后触发。
使用select语句等待定时器超时或收到取消信号
如果定时器超时，返回false表示继续重试；如果收到取消信号，返回true表示已取消

7、SubscriptionScope&scopeSub

type SubscriptionScope struct {
	mu     sync.Mutex
	subs   map[*scopeSub]struct{}
	closed bool
}

type scopeSub struct {
	sc *SubscriptionScope
	s  Subscription
}

SubscriptionScope提供了一种同时取消订阅多个订阅的功能。
对于处理多个订阅的代码，可以使用作用域通过一个调用方便地取消订阅所有订阅。该示例演示了在大型程序中的典型使用。

名为SubscriptionScope的结构体，它包含以下字段：

mu：一个sync.Mutex类型的变量，用于实现互斥锁，确保在多线程环境下对subs和closed字段的操作是线程安全的。
subs：一个map[*scopeSub]struct{}类型的变量，用于存储订阅者信息。键为scopeSub类型的指针，值为空结构体。
closed：一个布尔类型的变量，表示订阅范围是否已关闭。如果为true，则表示已关闭；如果为false，则表示未关闭。

名为scopeSub的结构体，它包含以下字段：

sc：一个指向SubscriptionScope类型的指针，表示订阅范围。
s：一个Subscription类型的变量，表示订阅信息。

7.1 sc.Track

func (sc *SubscriptionScope) Track(s Subscription) Subscription {
	sc.mu.Lock()
	defer sc.mu.Unlock()
	if sc.closed {
		return nil
	}
	if sc.subs == nil {
		sc.subs = make(map[*scopeSub]struct{})
	}
	ss := &scopeSub{sc, s}
	sc.subs[ss] = struct{}{}
	return ss
}

Track开始跟踪订阅。如果作用域已关闭，Track将返回nil。返回的订阅是一个包装。取消订阅包装会将其从作用域中删除。

上锁和延迟当函数完成的时候释放锁
如果订阅范围已经关闭，就返回nil
如果sc.subs为nil，就要重新创建一个
以sc和s为参数创建一个scopeSub的对象
将其加入到sc中的subs存储中
这个方法的主要作用是将订阅信息与订阅范围关联起来，并将它们存储在一个映射中。

7.2 sc.Close

func (sc *SubscriptionScope) Close() {
	sc.mu.Lock()
	defer sc.mu.Unlock()
	if sc.closed {
		return
	}
	sc.closed = true
	for s := range sc.subs {
		s.s.Unsubscribe()
	}
	sc.subs = nil
}

Close调用所有跟踪订阅的Unsubscribe，并阻止进一步添加到跟踪集。关闭后对Track的调用返回nil。

上锁和延迟当函数完成的时候释放锁
如果sc已经关闭，就直接return
否则将closed字段设置为true
同时遍历sc.subs中的所有订阅信息，进行取消订阅
最后将sc.subs的订阅map设置为nil

7.3 sc.Count

func (sc *SubscriptionScope) Count() int {
	sc.mu.Lock()
	defer sc.mu.Unlock()
	return len(sc.subs)
}

Count返回跟踪的订阅数

上锁和延迟当函数完成的时候释放锁
返回其中的sc.subs的长度

7.4 s.Unsubscribe

func (s *scopeSub) Unsubscribe() {
	s.s.Unsubscribe()
	s.sc.mu.Lock()
	defer s.sc.mu.Unlock()
	delete(s.sc.subs, s)
}

使用其中的Subscription类型的s的Unsubscribe()函数来取消订阅
上锁和延迟当函数完成的时候释放锁
使用delete从sc.subs中删除s

7.5 s.Err

func (s *scopeSub) Err() <-chan error {
	return s.s.Err()
}

返回错误信息
----------------------------event.go-----------------------------------

8、Event

type Event struct {
	once sync.Once

	feeds      map[string]*Feed
	feedsLock  sync.RWMutex
	feedsScope map[string]*SubscriptionScope
}

once sync.Once：一个只执行一次的同步原语，用于确保某个操作只执行一次。
feeds map[string]*Feed：一个字符串到Feed指针的映射，用于存储订阅信息。
feedsLock sync.RWMutex：一个读写互斥锁，用于保护feeds字段的并发访问。
feedsScope map[string]*SubscriptionScope：一个字符串到SubscriptionScope指针的映射，用于存储订阅范围信息。

8.1 init

func (e *Event) init() {
	e.feeds = make(map[string]*Feed)
	e.feedsScope = make(map[string]*SubscriptionScope)
}

初始化函数，创建feeds和feedsscope

8.2 initKey

func (e *Event) initKey(key string) {
	e.feedsLock.Lock()
	defer e.feedsLock.Unlock()
	if _, ok := e.feeds[key]; !ok {
		e.feeds[key] = new(Feed)
		e.feedsScope[key] = new(SubscriptionScope)
	}
}

初始化一个键值对，其中键为传入的字符串参数key，值为一个新的Feed对象和一个新的SubscriptionScope对象

上锁和延迟释放锁
如果feeds中还没有key对应的值，就创建一个新的Feed对象和一个新的SubscriptionScope对象

8.3 Subscribe

func (e *Event) Subscribe(channel interface{}) Subscription

是订阅一个通道（channel），并返回一个订阅对象（Subscription）。

调用e.once.Do(e.init)来确保在第一次调用Subscribe方法时执行e.init()函数进行初始化操作且只执行一次
通过reflect.TypeOf(channel).Elem().String()获取通道的类型字符串作为键值key
调用e.initKey(key)方法对键值进行初始化，传入key作为参数
使用e.feedsLock.RLock()加锁和延迟释放锁
通过e.feedsScope[key].Track(e.feeds[key].Subscribe(channel))创建一个订阅对象sub，并将其与对应的通道关联起来。
返回创建的订阅对象sub。

8.4 Send

func (e *Event) Send(value interface{}) int

向订阅对象发送消息

调用e.once.Do(e.init)来确保在第一次调用Subscribe方法时执行e.init()函数进行初始化操作且只执行一次
通过key := reflect.TypeOf(value).String()获取value的类型字符串作为键值key
调用e.initKey(key)方法对键值进行初始化，传入key作为参数
定义nsent来计数已发送的数量
使用e.feedsLock.RLock()加锁和延迟释放锁
如果订阅范围为0就返回nsent
否则调用send函数进行发送（参考1.4）

8.5 Close

func (e *Event) Close() {
	e.feedsLock.Lock()
	defer e.feedsLock.Unlock()

	for _, scope := range e.feedsScope {
		scope.Close()
	}
}

上锁和延迟释放锁
遍历所有的订阅范围，全部执行close函数，Close调用所有跟踪订阅的Unsubscribe（参考7.2）