在实际项目中，我们经常需要异步处理事件与数据。比如MVC模型中处理请求的Filter链，又如在nginx中或是linux的iptables中，都会有一个处理链条，来一步步的顺序处理一个请求。此外基于集中存储与分发的模式，实现事件与数据的异步处理，对于提升系统响应程度，实现业务处理的解耦至关重要。本文以eosc(一个高性能中间件开发框架)中的代码为例子，看看如何在我们的实际项目中，实现这样的功能。
eosc提供了关于dispatcher的关键实现的两个文件，分别是dispatch.go和data-dispatch.go,具体的代码地址是：
https://github.com/eolinker/eosc/tree/main/common/dispatcher。
这两个文件中实现的结构体与接口的关系如图所示：
dispatcher关键接口与结构体的关系
在dispatch.go文件中，esco提供了IEvent、CallBackHandler、IListener三个重要的接口。
同时通过CallBackFunc来实现接口CallBackHandler, tListener来实现IListener。
//2个接口
type CallBackHandler interface {
DataEvent(e IEvent) error
}
type IListener interface {
Leave()
Event() <-chan IEvent
}/*CallBackFunc实现了CallBackHandler,同时CallBackFunc又是一个接受IEvent为参数，
返回error的函数*/
type CallBackFunc func(e IEvent) error
func (f CallBackFunc) DataEvent(e IEvent) error {
return f(e)
}//实现了IListener接口
func (t *tListener) Leave() {
t.Once.Do(func() {
atomic.StoreUint32(&t.closed, 1)
close(t.c)
})
}
func (t *tListener) Event() <-chan IEvent {
return t.c
}注意：tListener还提供了一个Handler方法，这个方法的参数与返回结果与CallBackFunc一样，也就是说它实现的Handler方法是一种CallBackFunc，这个在后面的分发处理逻辑的注册中会用到。
func (t *tListener) Handler(e IEvent) error {
if atomic.LoadUint32(&t.closed) == 0 {
t.c <- e
return nil
}
return ErrorIsClosed
}该文件提供了两种dispatcher创建方法，分别是NewDataDispatchCenter、NewEventDispatchCenter。这两个方法都是创建了DataDispatchCenter结构体（这个结构体后面会讲到），但是启动的处理协程不同，NewDataDispatchCenter启动的是doDataLoop,NewEventDispatchCenter启动的是doEventLoop。
//两种DispatchCenter创建方法
func NewDataDispatchCenter() IDispatchCenter {
ctx, cancelFunc := context.WithCancel(context.Background())
center := &DataDispatchCenter{
ctx: ctx,
cancelFunc: cancelFunc,
addChannel: make(chan *_CallbackBox, 10),
eventChannel: make(chan IEvent),
}
go center.doDataLoop()
return center
}
func NewEventDispatchCenter() IDispatchCenter {
ctx, cancelFunc := context.WithCancel(context.Background())
center := &DataDispatchCenter{
ctx: ctx,
cancelFunc: cancelFunc,
addChannel: make(chan *_CallbackBox, 10),
eventChannel: make(chan IEvent),
}
go center.doEventLoop()
return center
}
//DataDispatchCenter 数据广播中心
type DataDispatchCenter struct {
addChannel chan *_CallbackBox
eventChannel chan IEvent
ctx context.Context
cancelFunc context.CancelFunc
}DataDispatchCenter这个结构体中有两个chan,一个是addChannel,一个是eventChannel。
addChannel | 接受_CallbackBox,这个_CallbackBox提供了逻辑处理Handler |
eventChannel | 接受IEvent，触发 |
NewEventDispatchCenter方法中启动的doEventLoop，逻辑相对简单，创建的channels用于存储addChannel发送过来的_CallbackBox，即事件处理Handler.当eventChannel收到事件后，遍历channels中的每一个_CallbackBox，并调用相应的Handler处理。
doEventLoop状态图
具体代码可以查看：
https://github.com/eolinker/eosc/blob/main/common/dispatcher/data-dispatch.go#L48。
NewDataDispatchCenter方法中启动的doDataLoop,这个逻辑稍微复杂点。其实它的大致流程和doEventLoop,不同的是每个新增加的_CallbackBox，需要对当前接收并缓存的所有Event键值对进行处理。而doEventLoop是不会的，新增加的_CallbackBox，只会对在它之后接收的Event生效。下面的代码InitEvent(data.GET())很有意思。
func (d *DataDispatchCenter) doDataLoop() {
data := NewMyData(nil)
channels := make([]*_CallbackBox, 0, 10)
isInit := false
for {
select {
case event, ok := <-d.eventChannel:
if ok {
isInit = true
data.DoEvent(event)
next := channels[:0]
for _, c := range channels {
if err := c.handler(event); err != nil {
close(c.closeChan)
continue
}
next = append(next, c)
}
channels = next
}
case hbox, ok := <-d.addChannel:
{
if ok {
if !isInit {
channels = append(channels, hbox)
} else {
if err := hbox.handler(InitEvent(data.GET())); err == nil {
channels = append(channels, hbox)
}
}
}
}
}
}
}创建EventServer。
type EventServer struct {
IDispatchCenter
}
func NewEventServer() *EventServer {
es := &EventServer{
IDispatchCenter: NewDataDispatchCenter(),
}
return es
}定义事件。
type MyEvent struct {
namespace string
key string
event string
data []byte
}
func (m *MyEvent) Namespace() string {
return m.namespace
}
func (m *MyEvent) Event() string {
return m.event
}
func (m *MyEvent) Key() string {
return m.key
}
func (m *MyEvent) Data() []byte {
return m.data
}定义Handler并注册。
func Handler(e IEvent) error {
//根据自己的业务要求
}
es.Register(Handler)发送事件。
es.Send(&MyEvent{
namespace: "a",
key: "b",
event: "set",
data: []byte(fmt.Sprint(index)),
})转载本文可以通过以下二维码关注联系。
我有一个模型:classItem项目有一个属性“商店”基于存储的值，我希望Item对象对特定方法具有不同的行为。Rails中是否有针对此的通用设计模式？如果方法中没有大的if-else语句，这是如何干净利落地完成的？ 最佳答案 通常通过Single-TableInheritance. 关于ruby-on-rails-Rails-子类化模型的设计模式是什么？，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.co
我主要使用Ruby来执行此操作，但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式，例如使用这个yaml文件，它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i
Rackup通过Rack的默认处理程序成功运行任何Rack应用程序。例如:classRackAppdefcall(environment)['200',{'Content-Type'=>'text/html'},["Helloworld"]]endendrunRackApp.new但是当最后一行更改为使用Rack的内置CGI处理程序时，rackup给出“NoMethodErrorat/undefinedmethod`call'fornil:NilClass”:Rack::Handler::CGI.runRackApp.newRack的其他内置处理程序也提出了同样的反对意见。例如Rack
鉴于我有以下迁移:Sequel.migrationdoupdoalter_table:usersdoadd_column:is_admin,:default=>falseend#SequelrunsaDESCRIBEtablestatement,whenthemodelisloaded.#Atthispoint,itdoesnotknowthatusershaveais_adminflag.#Soitfails.@user=User.find(:email=>"admin@fancy-startup.example")@user.is_admin=true@user.save!ende
有时我需要处理键/值数据。我不喜欢使用数组，因为它们在大小上没有限制(很容易不小心添加超过2个项目，而且您最终需要稍后验证大小)。此外，0和1的索引变成了魔数(MagicNumber)，并且在传达含义方面做得很差(“当我说0时，我的意思是head...”)。散列也不合适，因为可能会不小心添加额外的条目。我写了下面的类来解决这个问题:classPairattr_accessor:head,:taildefinitialize(h,t)@head,@tail=h,tendend它工作得很好并且解决了问题，但我很想知道:Ruby标准库是否已经带有这样一个类？ 最佳
给定一个复杂的对象层次结构，幸运的是它不包含循环引用，我如何实现支持各种格式的序列化？我不是来讨论实际实现的。相反，我正在寻找可能会派上用场的设计模式提示。更准确地说:我正在使用Ruby，我想解析XML和JSON数据以构建复杂的对象层次结构。此外，应该可以将该层次结构序列化为JSON、XML和可能的HTML。我可以为此使用Builder模式吗？在任何提到的情况下，我都有某种结构化数据-无论是在内存中还是文本中-我想用它来构建其他东西。我认为将序列化逻辑与实际业务逻辑分开会很好，这样我以后就可以轻松支持多种XML格式。 最佳答案 我最
是否有简单的方法来更改默认ISO格式(yyyy-mm-dd)的ActiveAdmin日期过滤器显示格式？ 最佳答案 您可以像这样为日期选择器提供额外的选项，而不是覆盖js:=f.input:my_date,as::datepicker,datepicker_options:{dateFormat:"mm/dd/yy"} 关于ruby-on-rails-事件管理员日期过滤器日期格式自定义，我们在StackOverflow上找到一个类似的问题： https://s
我正在尝试使用Curbgem执行以下POST以解析云curl-XPOST\-H"X-Parse-Application-Id:PARSE_APP_ID"\-H"X-Parse-REST-API-Key:PARSE_API_KEY"\-H"Content-Type:image/jpeg"\--data-binary'@myPicture.jpg'\https://api.parse.com/1/files/pic.jpg用这个:curl=Curl::Easy.new("https://api.parse.com/1/files/lion.jpg")curl.multipart_form_
无论您是想搭建桌面端、WEB端或者移动端APP应用，HOOPSPlatform组件都可以为您提供弹性的3D集成架构，同时，由工业领域3D技术专家组成的HOOPS技术团队也能为您提供技术支持服务。如果您的客户期望有一种在多个平台（桌面/WEB/APP，而且某些客户端是“瘦”客户端）快速、方便地将数据接入到3D应用系统的解决方案，并且当访问数据时，在各个平台上的性能和用户体验保持一致，HOOPSPlatform将帮助您完成。利用HOOPSPlatform，您可以开发在任何环境下的3D基础应用架构。HOOPSPlatform可以帮您打造3D创新型产品，HOOPSSDK包含的技术有：快速且准确的CAD
导读：随着叮咚买菜业务的发展，不同的业务场景对数据分析提出了不同的需求，他们希望引入一款实时OLAP数据库，构建一个灵活的多维实时查询和分析的平台，统一数据的接入和查询方案，解决各业务线对数据高效实时查询和精细化运营的需求。经过调研选型，最终引入ApacheDoris作为最终的OLAP分析引擎，Doris作为核心的OLAP引擎支持复杂地分析操作、提供多维的数据视图，在叮咚买菜数十个业务场景中广泛应用。作者｜叮咚买菜资深数据工程师韩青叮咚买菜创立于2017年5月，是一家专注美好食物的创业公司。叮咚买菜专注吃的事业，为满足更多人“想吃什么”而努力，通过美好食材的供应、美好滋味的开发以及美食品牌的孵