Golang轻量级NSQ主题与通道封装插件库nsq-event-bus的使用

Event Bus NSQ

• 一个围绕go-nsq主题和通道的轻量级封装
• 使用gobreaker保护NSQ调用

安装

go get -u github.com/rafaeljesus/nsq-event-bus

使用

nsq-event-bus包提供了发送和监听事件的接口。

发送器(Emitter)

import "github.com/rafaeljesus/nsq-event-bus"

topic := "events"
emitter, err := bus.NewEmitter(bus.EmitterConfig{
  Address: "localhost:4150",
  MaxInFlight: 25,
})

e := event{}
if err = emitter.Emit(topic, &e); err != nil {
  // 处理发送消息失败的情况
}

// 异步发送消息
if err = emitter.EmitAsync(topic, &e); err != nil {
  // 处理发送消息失败的情况
}

监听器(Listener)

import "github.com/rafaeljesus/nsq-event-bus"

if err = bus.On(bus.ListenerConfig{
  Topic:              "topic",
  Channel:            "test_on",
  HandlerFunc:        handler,
  HandlerConcurrency: 4,
}); err != nil {
  // 处理监听消息失败的情况
}

func handler(message *Message) (reply interface{}, err error) {
  e := event{}
  if err = message.DecodePayload(&e); err != nil {
    message.Finish()
    return
  }

  if message.Attempts > MAX_DELIVERY_ATTEMPTS {
    message.Finish()
    return
  }

  err, _ = doWork(&e)
  if err != nil {
    message.Requeue(BACKOFF_TIME)
    return
  }

  message.Finish()
  return
}

请求(Request/Reply)

import "github.com/rafaeljesus/nsq-event-bus"

topic := "user_signup"
emitter, err = bus.NewEmitter(bus.EmitterConfig{})

e := event{Login: "rafa", Password: "ilhabela_is_the_place"}
if err = bus.Request(topic, &e, handler); err != nil {
  // 处理监听消息失败的情况
}

func handler(message *Message) (reply interface{}, err error) {
  e := event{}
  if err = message.DecodePayload(&e); err != nil {
    message.Finish()
    return
  }

  reply = &Reply{}
  message.Finish()
  return
}

完整示例Demo

生产者示例

package main

import (
	"github.com/rafaeljesus/nsq-event-bus"
	"log"
)

type UserEvent struct {
	ID    int    `json:"id"`
	Name  string `json:"name"`
	Email string `json:"email"`
}

func main() {
	// 创建发送器
	emitter, err := bus.NewEmitter(bus.EmitterConfig{
		Address:    "localhost:4150",
		MaxInFlight: 10,
	})
	if err != nil {
		log.Fatal(err)
	}

	// 创建事件
	user := UserEvent{
		ID:    1,
		Name:  "John Doe",
		Email: "john@example.com",
	}

	// 发送事件
	topic := "user_created"
	if err := emitter.Emit(topic, &user); err != nil {
		log.Printf("Failed to emit message: %v", err)
	}

	log.Printf("Event emitted to topic %s", topic)
}

消费者示例

package main

import (
	"github.com/rafaeljesus/nsq-event-bus"
	"log"
)

type UserEvent struct {
	ID    int    `json:"id"`
	Name  string `json:"name"`
	Email string `json:"email"`
}

func main() {
	// 监听事件
	err := bus.On(bus.ListenerConfig{
		Topic:              "user_created",
		Channel:            "user_service",
		HandlerFunc:        handleUserCreated,
		HandlerConcurrency: 5,
	})
	if err != nil {
		log.Fatal(err)
	}

	// 保持程序运行
	select {}
}

func handleUserCreated(message *bus.Message) (reply interface{}, err error) {
	var user UserEvent
	if err := message.DecodePayload(&user); err != nil {
		log.Printf("Failed to decode message: %v", err)
		message.Finish()
		return nil, err
	}

	log.Printf("Received new user: ID=%d, Name=%s, Email=%s", 
		user.ID, user.Name, user.Email)

	// 处理业务逻辑...

	message.Finish()
	return nil, nil
}

贡献

• Fork仓库
• 创建你的特性分支 (git checkout -b my-new-feature)
• 提交你的更改 (git commit -am 'Add some feature')
• 推送到分支 (git push origin my-new-feature)
• 创建新的Pull Request