前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >拒绝Go代码臃肿,其实在这几块可以用下观察者模式

拒绝Go代码臃肿,其实在这几块可以用下观察者模式

作者头像
KevinYan
发布2022-10-27 10:42:29
1870
发布2022-10-27 10:42:29
举报
文章被收录于专栏:网管叨bi叨网管叨bi叨

今天学习一下用 Go 实现观察者模式,观察者模式主要是用来实现事件驱动编程。事件驱动编程的应用还是挺广的,除了我们都知道的能够用来解耦:用户修改密码后,给用户发短信进行风险提示之类的典型场景,在微服务架构实现最终一致性、实现事件源(A + ES)这些都会用到。

我们先来简单学习一下用 Go 实现观察者设计模式,给怎么实现事件驱动编程、事件源这些模式做个铺垫。主要也是我也老没看设计模式了,一起再复习一下。以前看的设计模式教程都是 Java 的,这次用 Go 实现一番。

观察者模式

咱们先来看一下观察者模式的概念,我尽量加一些自己的理解,让它变成咱们都能理解的大俗话:

概念

观察者模式 (Observer Pattern),定义对象间的一种一对多依赖关系,使得每当一个对象状态发生改变时,其相关依赖对象皆得到通知,依赖对象在收到通知后,可自行调用自身的处理程序,实现想要干的事情,比如更新自己的状态。

发布者对观察者唯一了解的是它实现了某个接口(观察者接口)。这种松散耦合的设计最大限度地减少了对象之间的相互依赖,因此使我们能够构建灵活的系统。

我的理解

上面这段话看完,相信几乎对于理解观察者模式能起到的作用微乎其微,类似于现实职场里加班对项目进度起到的作用一样,加班的时候谁还没打过几把王者荣耀,嘿。下面我用自己的理解再给你们唠一下。

观察者模式也经常被叫做发布 - 订阅(Publish/Subscribe)模式、上面说的定义对象间的一种一对多依赖关系,一 - 指的是发布变更的主体对象,多 - 指的是订阅变更通知的订阅者对象。

发布的状态变更信息会被包装到一个对象里,这个对象被称为事件,事件一般用英语过去式的语态来命名,比如用户注册时,用户模块在用户创建好后发布一个事件 UserCreated 或者 UserWasCreated 都行,这样从名字上就能看出,这是一个已经发生过的事件。

事件发布给订阅者的过程,其实就是遍历一下已经注册的事件订阅者,逐个去调用订阅者实现的观察者接口方法,比如叫 handleEvent 之类的方法,这个方法的参数一般就是当前的事件对象。

至于很多人会好奇的,事件的处理是不是异步的?主要看我们的需求是什么,一般情况下是同步的,即发布事件后,触发事件的方法会阻塞等到全部订阅者返回后再继续,当然也可以让订阅者的处理异步执行,完全看我们的需求。

大部分场景下其实是同步执行的,单体架构会在一个数据库事务里持久化因为主体状态变更,而需要更改的所有实体类。

微服务架构下常见的做法是有一个事件存储,订阅者接到事件通知后,会把事件先存到事件存储里,这两步也需要在一个事务里完成才能保证最终一致性,后面会再有其他线程把事件从事件存储里搞到消息设施里,发给其他服务,从而在微服务架构下实现各个位于不同服务的实体间的最终一致性。

所以观察者模式,从程序效率上看,大多数情况下没啥提升,更多的是达到一种程序结构上的解耦,让代码不至于那么难维护。

Go 实现观察者模式

说了这么多,我们再看下用 Go 怎么实现最简单的观察者模式:

代码语言:javascript
复制
package main

import "fmt"

// Subject 接口,它相当于是发布者的定义
type Subject interface {
 Subscribe(observer Observer)
 Notify(msg string)
}

// Observer 观察者接口
type Observer interface {
 Update(msg string)
}

// Subject 实现
type SubjectImpl struct {
 observers []Observer
}

// Subscribe 添加观察者(订阅者)
func (sub *SubjectImpl) Subscribe(observer Observer) {
 sub.observers = append(sub.observers, observer)
}


// Notify 发布通知
func (sub *SubjectImpl) Notify(msg string) {
 for _, o := range sub.observers {
  o.Update(msg)
 }
}

// Observer1 Observer1
type Observer1 struct{}

// Update 实现观察者接口
func (Observer1) Update(msg string) {
 fmt.Printf("Observer1: %s\n", msg)
}

// Observer2 Observer2
type Observer2 struct{}

// Update 实现观察者接口
func (Observer2) Update(msg string) {
 fmt.Printf("Observer2: %s\n", msg)
}

func main(){
 sub := &SubjectImpl{}
 sub.Subscribe(&Observer1{})
 sub.Subscribe(&Observer2{})
 sub.Notify("Hello")
}

这就是 Go 实现观察者模式的代码,实际应用的时候,一般会定义个事件总线 EventBus 或者事件分发器 Event Dispatcher,来管理事件和订阅者间的关系,以及分发事件,它们两个就是名不一样,角色定位一样。

下面看一下用 Go 怎么实现事件总线。

Go 实现事件总线

下面我们实现一个支持以下功能的事件总线

  1. 异步不阻塞
  2. 支持任意参数值

这个代码不是我自己写的,出处见代码注释首行。

代码

代码语言:javascript
复制
// 代码来自https://lailin.xyz/post/observer.html
package eventbus

import (
 "fmt"
 "reflect"
 "sync"
)

// Bus Bus
type Bus interface {
 Subscribe(topic string, handler interface{}) error
 Publish(topic string, args ...interface{})
}

// AsyncEventBus 异步事件总线
type AsyncEventBus struct {
 handlers map[string][]reflect.Value
 lock     sync.Mutex
}

// NewAsyncEventBus new
func NewAsyncEventBus() *AsyncEventBus {
 return &AsyncEventBus{
  handlers: map[string][]reflect.Value{},
  lock:     sync.Mutex{},
 }
}

// Subscribe 订阅
func (bus *AsyncEventBus) Subscribe(topic string, f interface{}) error {
 bus.lock.Lock()
 defer bus.lock.Unlock()

 v := reflect.ValueOf(f)
 if v.Type().Kind() != reflect.Func {
  return fmt.Errorf("handler is not a function")
 }

 handler, ok := bus.handlers[topic]
 if !ok {
  handler = []reflect.Value{}
 }
 handler = append(handler, v)
 bus.handlers[topic] = handler

 return nil
}

// Publish 发布
// 这里异步执行,并且不会等待返回结果
func (bus *AsyncEventBus) Publish(topic string, args ...interface{}) {
 handlers, ok := bus.handlers[topic]
 if !ok {
  fmt.Println("not found handlers in topic:", topic)
  return
 }

 params := make([]reflect.Value, len(args))
 for i, arg := range args {
  params[i] = reflect.ValueOf(arg)
 }

 for i := range handlers {
  go handlers[i].Call(params)
 }
}

单测

代码语言:javascript
复制
package eventbus

import (
 "fmt"
 "testing"
 "time"
)

func sub1(msg1, msg2 string) {
 time.Sleep(1 * time.Microsecond)
 fmt.Printf("sub1, %s %s\n", msg1, msg2)
}

func sub2(msg1, msg2 string) {
 fmt.Printf("sub2, %s %s\n", msg1, msg2)
}
func TestAsyncEventBus_Publish(t *testing.T) {
 bus := NewAsyncEventBus()
 bus.Subscribe("topic:1", sub1)
 bus.Subscribe("topic:1", sub2)
 bus.Publish("topic:1", "test1", "test2")
 bus.Publish("topic:1", "testA", "testB")
 time.Sleep(1 * time.Second)
}

毫不意外这个事件总线,只是个例子,咱也不能在项目开发里使用,这篇文章咱们先搞清概念,我其实前两天关注了下,没有发现什么好用的事件分发、事件总线的三方库,好在实现起来也不难,后面我准备自己写一个能用的到时候分享给大家,最起码是在学习、练习项目里能使用的吧。

总结

今天给大家用大白话瞎唠了一下观察者模式的原理和实际怎么应用,感觉文章的精髓主要在前半部分,可能有的不你还不能理解,后面我会再通过后续文章逐一解释,其实这些都是事件驱动和事件源这些模式里的基础内容。

至于这次给出的代码,其实没啥实战意义,就是大家先了解一下。Go 里边关于事件驱动之类的内容,感觉不多,有 Spring 使用经验的可以先看看 Spring 提供的@EventListener 注解,需要订阅者异步执行可以配合 @Async 注解使用,至于我上面说的需要保证事件发布的主体和订阅者的原子化的话,则是通过@Transitional 和 @TransactionalEventListener 结合使用来实现。

今天的内容就这些,下期再见,各位读者大佬如果方便还请转发、点赞、关注,支持一波,评论区走起来,让我们一起变得更强。

本文参与 腾讯云自媒体分享计划,分享自微信公众号。
原始发表:2022-07-18,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 网管叨bi叨 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 观察者模式
    • 概念
      • 我的理解
        • Go 实现观察者模式
        • Go 实现事件总线
          • 代码
            • 单测
            • 总结
            相关产品与服务
            事件总线
            腾讯云事件总线(EventBridge)是一款安全,稳定,高效的云上事件连接器,作为流数据和事件的自动收集、处理、分发管道,通过可视化的配置,实现事件源(例如:Kafka,审计,数据库等)和目标对象(例如:CLS,SCF等)的快速连接,当前 EventBridge 已接入 100+ 云上服务,助力分布式事件驱动架构的快速构建。
            领券
            问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档