Go观察者模式:事件驱动架构

张

张建站

2026/8/19 6:11:37

10分钟阅读

Go观察者模式:事件驱动架构摘要: 本篇讲解Go语言观察者模式定义Subject和Observer接口用channel实现事件总线sync.Map管理订阅者支持事件过滤与优先级调度分享观察者goroutine泄漏的踩坑经验。开篇故事去年做订单系统下单成功后要触发库存扣减、发短信、发优惠券、写积分记录四件事。第一版在下单函数里直接调用这四个函数强耦合改一个发券逻辑要重新部署整个订单服务。某次发券服务超时3秒整个下单流程跟着卡了3秒。后来重构成事件驱动: 下单成功后发一个OrderCreated事件库存服务、短信服务、优惠券服务各自订阅这个事件。下单函数只管创建订单和发事件不关心谁消费。发券服务超时不再影响下单各消费方异步处理。但重构成事件总线后遇到一个新问题某个消费者goroutine阻塞了channel事件堆积内存涨了2个G。这篇把观察者模式和事件总线的实现讲清楚再说说goroutine泄漏那个坑。一、Subject和Observer接口观察者模式的核心是: 主题状态变化时通知所有观察者。Go里用接口定义Subject和Observer的角色。packageobserver// Event 事件结构typeEventstruct{Topicstring// 事件主题如order.createdDatainterface{}// 事件数据}// Observer 观察者接口// 订阅者实现这个接口处理事件typeObserverinterface{// OnEvent 处理事件OnEvent(event Event)// ID 观察者唯一标识ID()string}// Subject 主题接口// 事件发布者实现这个接口typeSubjectinterface{// Subscribe 订阅事件Subscribe(topicstring,o Observer)// Unsubscribe 取消订阅Unsubscribe(topicstring,idstring)// Publish 发布事件Publish(topicstring,datainterface{})}// --- 邮件通知观察者 ---typeemailObserverstruct{idstring}func(e*emailObserver)OnEvent(event Event){// 处理事件发送邮件orderID:event.Data.(string)// sendEmail(orderID) ..._orderID}func(e*emailObserver)ID()string{returne.id}// NewEmailObserver 创建邮件观察者funcNewEmailObserver(idstring)Observer{returnemailObserver{id:id}}这是最基础的同步实现。Subject的Publish方法遍历所有观察者逐个调用OnEvent。问题是同步调用一个观察者慢所有观察者跟着等。二、channel实现事件总线实际项目里用channel做事件总线发布者和订阅者解耦事件通过channel异步传递。packageeventbusimport(sync)// Event 事件结构typeEventstruct{TopicstringDatainterface{}}// Handler 事件处理函数类型typeHandlerfunc(Event)// Subscription 订阅信息typeSubscriptionstruct{idstring// 订阅ID用于取消订阅topicstring// 订阅的主题handler Handler// 处理函数chchanEvent// 事件channelquitchanstruct{}// 退出信号}// EventBus 事件总线// 用channel异步分发事件typeEventBusstruct{mu sync.RWMutex subsmap[string][]*Subscription// topic到订阅列表}// NewEventBus 创建事件总线funcNewEventBus()*EventBus{returnEventBus{subs:make(map[string][]*Subscription),}}// Subscribe 订阅事件// topic: 事件主题// handler: 处理函数// bufferSize: channel缓冲区大小// 返回订阅ID用于取消订阅func(bus*EventBus)Subscribe(topicstring,handler Handler,bufferSizeint)string{sub:Subscription{id:topic_nextID(),topic:topic,handler:handler,ch:make(chanEvent,bufferSize),quit:make(chanstruct{}),}// 启动goroutine消费事件gofunc(){for{select{caseevent:-sub.ch:// 调用处理函数sub.handler(event)case-sub.quit:// 收到退出信号停止消费return}}}()// 注册到订阅列表bus.mu.Lock()bus.subs[topic]append(bus.subs[topic],sub)bus.mu.Unlock()returnsub.id}// Publish 发布事件// 非阻塞: 如果订阅者channel满了丢弃事件func(bus*EventBus)Publish(topicstring,datainterface{}){event:Event{Topic:topic,Data:data}bus.mu.RLock()subs:bus.subs[topic]bus.mu.RUnlock()// 向所有订阅者的channel发送事件for_,sub:rangesubs{select{casesub.ch-event:// 发送成功default:// channel满了丢弃事件// 也可以用log记录}}}// Unsubscribe 取消订阅func(bus*EventBus)Unsubscribe(idstring){bus.mu.Lock()deferbus.mu.Unlock()fortopic,subs:rangebus.subs{fori,sub:rangesubs{ifsub.idid{// 通知goroutine退出close(sub.quit)// 从列表中删除bus.subs[topic]append(subs[:i],subs[i1:]...,)return}}}}// id生成器(简化版)varidCounterintfuncnextID()string{idCounterreturnstring(rune(0idCounter))}事件总线用channel解耦发布者和订阅者。Publish把事件塞进订阅者的channel订阅者各自的goroutine从channel取事件处理。一个订阅者慢不影响其他订阅者也不阻塞发布者。三、事件过滤与优先级实际项目里订阅者可能只关心部分事件。比如订单服务订阅order.created事件但只处理金额大于1000的订单。把过滤逻辑放在订阅端不放在事件总线总线只管路由。packageeventbusimport(sortsync)// FilterFunc 事件过滤函数// 返回true表示订阅者关心这个事件typeFilterFuncfunc(Event)bool// PrioritySubscription 带优先级的订阅typePrioritySubscriptionstruct{*Subscription priorityint// 优先级数字越大越先执行filter FilterFunc// 过滤函数}// PriorityEventBus 支持过滤和优先级的事件总线typePriorityEventBusstruct{mu sync.RWMutex subsmap[string][]*PrioritySubscription}funcNewPriorityEventBus()*PriorityEventBus{returnPriorityEventBus{subs:make(map[string][]*PrioritySubscription),}}// SubscribeWithFilter 带过滤的订阅func(bus*PriorityEventBus)SubscribeWithFilter(topicstring,handler Handler,priorityint,filter FilterFunc,bufferSizeint,)string{sub:PrioritySubscription{Subscription:Subscription{id:topic_nextID(),topic:topic,handler:handler,ch:make(chanEvent,bufferSize),quit:make(chanstruct{}),},priority:priority,filter:filter,}// 启动消费goroutine处理前先过滤gofunc(){for{select{caseevent:-sub.ch:// 过滤函数返回false则跳过ifsub.filter!nil!sub.filter(event){continue}sub.handler(event)case-sub.quit:return}}}()bus.mu.Lock()bus.subs[topic]append(bus.subs[topic],sub)// 按优先级降序排列sort.Slice(bus.subs[topic],func(i,jint)bool{returnbus.subs[topic][i].prioritybus.subs[topic][j].priority})bus.mu.Unlock()returnsub.id}// PublishSync 同步发布按优先级顺序执行// 高优先级的订阅者先处理事件func(bus*PriorityEventBus)PublishSync(topicstring,datainterface{}){event:Event{Topic:topic,Data:data}bus.mu.RLock()subs:make([]*PrioritySubscription,len(bus.subs[topic]))copy(subs,bus.subs[topic])bus.mu.RUnlock()// 按优先级顺序同步调用for_,sub:rangesubs{// 过滤ifsub.filter!nil!sub.filter(event){continue}// 同步调用处理函数sub.handler(event)}}优先级的作用是控制事件的处理顺序。比如鉴权日志的优先级最高审计日志优先级中等普通业务日志最低。同步发布时按优先级顺序执行保证关键日志先写。异步发布时优先级影响goroutine的调度顺序。四、踩坑经验:观察者goroutine泄漏这个坑我踩过。事件总线运行一周后内存从200MB涨到2GB。pprof一看8万多个goroutine全是事件总线的消费goroutine。原因是某个服务动态订阅事件处理完后忘了调用Unsubscribe每次请求都创建一个新订阅消费goroutine永远不会退出。更隐蔽的问题是: 即使调用了Unsubscribe如果channel里还有积压事件没消费goroutine退出时会丢事件。还有种情况是消费handler卡死channel满了新事件全被丢弃。packageeventbusimport(contextlogsynctime)// SafeEventBus 带超时和泄漏防护的事件总线typeSafeEventBusstruct{mu sync.RWMutex subsmap[string][]*safeSubscription}// safeSubscription 带超时控制的订阅typesafeSubscriptionstruct{idstringtopicstringhandler Handler chchanEvent quitchanstruct{}ctx context.Context// 用于超时控制cancel context.CancelFunc}funcNewSafeEventBus()*SafeEventBus{returnSafeEventBus{subs:make(map[string][]*safeSubscription),}}// Subscribe 带超时控制的订阅// handlerTimeout: 单个事件处理超时时间func(bus*SafeEventBus)Subscribe(topicstring,handler Handler,bufferSizeint,handlerTimeout time.Duration,)string{ctx,cancel:context.WithCancel(context.Background())sub:safeSubscription{id:topic_nextID(),topic:topic,handler:handler,ch:make(chanEvent,bufferSize),quit:make(chanstruct{}),ctx:ctx,cancel:cancel,}// 消费goroutine带超时控制gofunc(){for{select{caseevent:-sub.ch:// 用context控制单次处理超时bus.handleWithTimeout(sub,event,handlerTimeout)case-sub.quit:// 正常退出把剩余事件排空bus.drain(sub)returncase-sub.ctx.Done():// 被cancel强制退出return}}}()bus.mu.Lock()bus.subs[topic]append(bus.subs[topic],sub)bus.mu.Unlock()returnsub.id}// handleWithTimeout 带超时的事件处理func(bus*SafeEventBus)handleWithTimeout(sub*safeSubscription,event Event,timeout time.Duration,){done:make(chanstruct{})gofunc(){deferclose(done)deferfunc(){ifr:recover();r!nil{// handler panic不影响其他事件log.Printf(handler panic: %v,r)}}()sub.handler(event)}()select{case-done:// 正常完成case-time.After(timeout):// 处理超时放弃这个事件log.Printf(事件处理超时: topic%s,event.Topic)}}// Unsubscribe 取消订阅// 先发退出信号让消费goroutine退出func(bus*SafeEventBus)Unsubscribe(idstring){bus.mu.Lock()deferbus.mu.Unlock()fortopic,subs:rangebus.subs{fori,sub:rangesubs{ifsub.idid{sub.cancel()// 取消contextclose(sub.quit)// 发退出信号// 从列表删除bus.subs[topic]append(subs[:i],subs[i1:]...,)return}}}}三个关键改进防止泄漏。一是每个handler处理加超时单个事件卡住不会阻塞整个消费goroutine。二是Unsubscribe时cancel context并发quit信号消费goroutine能及时退出。三是handler加recoverpanic不会杀死消费goroutine。定期监控订阅数量发现异常增长及时排查。五、对比分析方案异步性解耦程度吞吐量适用场景同步观察者同步中低少量订阅者处理快channel事件总线异步高中大量订阅者解耦要求高sync.Map事件总线异步高中高订阅频繁增删kafka等消息队列异步最高高跨服务高可靠同步观察者最简单Subject遍历调用Observer一个慢全跟着慢。channel事件总线用goroutine和channel解耦发布者和订阅者独立运行。sync.Map管理的总线读写不互斥订阅频繁增删时性能好。跨服务场景用kafka这类消息队列持久化保证不丢消息但引入了外部依赖。总结观察者模式用Subject和Observer接口解耦事件发布者和订阅者。channel实现的异步事件总线发布者不阻塞各订阅者独立消费。订阅者处理加超时和recover防止单个事件卡死整个消费流程。取消订阅时cancel context并发退出信号让消费goroutine及时退出防止泄漏。下一篇我们聊Go里的依赖注入看看wire怎么自动生成组装代码。