README
¶
RabbitMQ 实现
基于 github.com/rabbitmq/amqp091-go 的 RabbitMQ 实现。
特性
- ✅ 支持事件驱动模式（广播）
- ✅ 支持任务队列模式（负载均衡）
- ✅ 自动声明 exchange 和 queue
- ✅ 消息持久化
- ✅ 失败自动重试
- ✅ 优雅退出
快速开始
1. 启动 RabbitMQ
# 使用 Docker
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management
访问管理界面：http://localhost:15672 (用户名/密码：guest/guest)
2. 基本使用
import "github.com/FangcunMount/component-base/pkg/messaging/rabbitmq"
bus, _ := rabbitmq.NewEventBus("amqp://guest:guest@localhost:5672/")
defer bus.Close()
// 发布消息
bus.Publisher().Publish(ctx, "user.created", data)
// 订阅消息
bus.Subscriber().Subscribe("user.created", "email-service", handler)
运行示例
go run pkg/messaging/example/rabbitmq/main.go
Documentation
¶
Index ¶
- func NewEventBus(url string) (messaging.EventBus, error)
- func NewEventBusFromConfig(config *messaging.Config) (messaging.EventBus, error)
- func NewPublisher(url string) (messaging.Publisher, error)
- func NewSubscriber(url string) (messaging.Subscriber, error)
- func NewSubscriberWithOptions(url string, opts messaging.SubscriberOptions) (messaging.Subscriber, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewEventBus ¶
NewEventBus 创建 RabbitMQ 事件总线
url 格式：amqp://username:password@host:port/vhost 例如：amqp://guest:guest@localhost:5672/
使用示例：
bus, err := rabbitmq.NewEventBus("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatal(err)
}
defer bus.Close()
// 使用发布者
publisher := bus.Publisher()
publisher.Publish(ctx, "user.created", data)
// 使用订阅者
subscriber := bus.Subscriber()
subscriber.Subscribe("user.created", "email-service", handler)
func NewEventBusFromConfig ¶
NewEventBusFromConfig 从配置创建事件总线
func NewPublisher ¶
NewPublisher 创建 RabbitMQ 发布者
url 格式：amqp://username:password@host:port/vhost 例如：amqp://guest:guest@localhost:5672/
func NewSubscriber ¶
func NewSubscriber(url string) (messaging.Subscriber, error)
func NewSubscriberWithOptions ¶ added in v0.6.6
func NewSubscriberWithOptions(url string, opts messaging.SubscriberOptions) (messaging.Subscriber, error)
NewSubscriberWithOptions enables bounded retry topology when MaxAttempts is positive. Zero options preserve the historical immediate-requeue behavior.
Types ¶
This section is empty.
Source Files
Click to show internal directories.
Click to hide internal directories.