我是初学者 gopher，我为我正在从事的项目编写了一个事件监听器工作队列。

我已将其部署在临时服务器上。在触发大约 100 个事件后，监听器将在事件发布时停止调用。服务器也没有崩溃。

这是我的实现:

// Event struct 
type Event struct {
  Name string
  Data interface{}
}

// Stream to publish events to
var stream = make(chan *Event, 100)

// Publish sends new event data to the stream by the event name
func Publish(name string, data interface{}) {
  ev := &Event{name, data}
  stream <- ev
}

// Handler provides the interface for all event handlers.
// The Work will be called with the event that it should process
type Handler interface {
  Work(*Event)
}

type worker struct {
  Handler
  Listen chan *Event
  Quit   chan bool
}

// Stop shuts down the worker
func (w *worker) Stop() {
  go func() {
    w.Quit <- true
  }()
}

// Queue of worker Listen channels
type workerQueue chan chan *Event

// registry of workers
var registry = make(map[string][]workerQueue)

// Register creates 20 workers, assigns them to a queue, and 
// appends the resulting worker queue to an event on the handler registry
func Register(name string, handlers ...Handler) {
  if _, ok := registry[name]; !ok {
    registry[name] = make([]workerQueue, 0)
  }

  // Create workerQueues for each handler
  for _, h := range handlers {
    queue := make(workerQueue, numListeners)

    // Create 20 workers
    for i := 0; i < 20; i++ {
      newWorker := worker{
        Handler: h,
        Listen:  make(chan *Event),
        Quit:    make(chan bool),
      }

      go func() {
        for {
          select {
          case ev := <-newWorker.Listen:
            nl.Work(ev)
          case <-newWorker.Quit:
            return
          }
        }
      }()

      queue <- newWorker.Listen
    }

    registry[name] = append(registry[name], queue)
  }
}

// Start begins listening for events on stream
func Start() {
  go func() {
    for {
      select {
      // listen for events
      case ev := <-stream:
        go func() {
          // get registered queues for the event
          queues, ok := registry[ev.Name]

          if !ok {
            return
          }

          // Get worker channel from queue and send the event
          for _, queue := range queues {
            worker := <-queue
            worker <- ev
          }
        }()
      }
    }
  }()
}

这是一个示例用法。

// Usage

Start()

type demoHandler struct {
  db *sql.DB
}

type eventData struct {}

func (h *demoHandler) Work(ev *Event) {
  // Do something
  return
}

// Register handler
Register('some-event', &demoHandler{r})

Publish('some-event', &eventData{})

我将一个指向 demoHandler 的指针作为事件处理程序传递，因为它们需要访问底层的 sql 实例。每个worker queue使用同一个demoHandler是不是有问题？

我这辈子都想不通哪里出了问题!除了处理程序代码中的错误之外，我的代码中是否存在错误导致我所有的工作人员都宕机？

最佳答案

“在 go worker/event 系统中，worker 是否应该访问相同的结构(通过指针)来完成工作？” 不，这不是问题。如果您的处理程序中的代码访问 critical section，那将是一个问题，但我认为这不会导致您的程序阻塞。

您的服务器不会崩溃或阻塞，因为没有触发 panic ，并且您的程序正在单独的 goroutine 上监听和执行，这些 goroutine 是轻量级执行线程。

它可能必须与您用来发送和接收事件的 channel 有关。

发送和接收到 channel默认情况下是阻塞的。这意味着当您从 channel 发送或接收时，它将阻塞其 goroutine，直到另一端准备就绪。

在buffered channels的情况下，当缓冲区已满时发送 block ，当缓冲区为空时接收 block ，就像在您的流 channel 中一样:

var stream = make(chan *Event, 100)

您说:“在触发大约 100 个事件后，当事件发布时，监听器将停止被调用”。

因此，如果您调用 Publish 函数并执行 stream <- ev当“流” channel 缓冲区已满时，它将阻塞，直到 channel 有地方接收另一个元素。

我建议阅读一些关于 non-blocking channel operations 的内容.

也许阻塞发生在您实际使用代码的某些部分。

关于go - 在 go worker/event 系统中，worker 是否应该访问相同的结构(通过指针)来工作？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/54517080/