Documentation
¶
Overview ¶
Package rocketmq provides a RocketMQ backend for the ling-base/mq broker abstraction. It wraps the official apache/rocketmq-client-go/v2 client, mapping the broker-agnostic mq.Message / mq.Delivery / mq.Producer / mq.Consumer / mq.Broker interfaces onto RocketMQ topics, producer groups and push consumers.
RocketMQ has no concept of exchanges, queues or bindings in the AMQP sense: it routes messages purely by topic (and optionally tag/sharding key). The topology-management methods on mq.Broker (DeclareExchange, DeclareQueue, Bind, Unbind, DeleteQueue, DeleteExchange) are therefore implemented as no-ops that return nil, except that DeclareExchange records the topic name so that producers can publish to it.
Index ¶
- type Broker
- func (b *Broker) Bind(queue, exchange, routingKey string) error
- func (b *Broker) Close() error
- func (b *Broker) Connect() error
- func (b *Broker) Consumer(topic string, opts mq.ConsumeOptions) (mq.Consumer, error)
- func (b *Broker) DeclareExchange(name string, opts mq.ExchangeOptions) error
- func (b *Broker) DeclareQueue(name string, opts mq.QueueOptions) error
- func (b *Broker) DeleteExchange(name string) error
- func (b *Broker) DeleteQueue(name string) error
- func (b *Broker) IsConnected() bool
- func (b *Broker) Metrics() mq.Metrics
- func (b *Broker) Producer(topic string, opts mq.PublishOptions) (mq.Producer, error)
- func (b *Broker) Unbind(queue, exchange, routingKey string) error
- type Config
- type Consumer
- type Credentials
- type Delivery
- func (d *Delivery) Ack() error
- func (d *Delivery) Body() []byte
- func (d *Delivery) DeliveryTag() uint64
- func (d *Delivery) Exchange() string
- func (d *Delivery) Headers() map[string]any
- func (d *Delivery) Message() *mq.Message
- func (d *Delivery) Nack(requeue bool) error
- func (d *Delivery) Redelivered() bool
- func (d *Delivery) Reject(requeue bool) error
- func (d *Delivery) RoutingKey() string
- type Producer
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker implements mq.Broker for RocketMQ.
func New ¶
New creates a new RocketMQ broker. Call Connect() to initialize the underlying RocketMQ client.
func (*Broker) Connect ¶
Connect initializes the RocketMQ client. RocketMQ clients are lazy: producers and consumers are created on demand, so Connect simply marks the broker as ready. A lightweight name-server resolver is validated here so that an invalid configuration is reported early.
func (*Broker) Consumer ¶
Consumer creates a consumer for the given topic with the given options. The consumer is not started; call Start to begin consuming.
func (*Broker) DeclareExchange ¶
func (b *Broker) DeclareExchange(name string, opts mq.ExchangeOptions) error
DeclareExchange records a topic name. RocketMQ auto-creates topics on first publish (when autoCreateTopicEnable is on at the broker), so this is effectively a no-op that simply remembers the name.
func (*Broker) DeclareQueue ¶
func (b *Broker) DeclareQueue(name string, opts mq.QueueOptions) error
DeclareQueue is a no-op: RocketMQ has no AMQP-style queues.
func (*Broker) DeleteExchange ¶
DeleteExchange removes a recorded topic from the in-memory registry. It does not delete the topic from the broker.
func (*Broker) DeleteQueue ¶
DeleteQueue is a no-op: RocketMQ has no AMQP-style queues.
func (*Broker) IsConnected ¶
IsConnected reports whether the broker is currently connected.
type Config ¶
type Config struct {
// NameServer is the list of name-server addresses
// (e.g. []string{"127.0.0.1:9876"}). Required.
NameServer []string
// GroupName is the default producer/consumer group name.
// Default: "DEFAULT_PRODUCER" / "DEFAULT_CONSUMER".
GroupName string
// InstanceName is the client instance name. If empty the
// client uses the process host name.
InstanceName string
// Credentials are the optional ACL credentials.
Credentials Credentials
// RetryCount is the number of retries on send failure.
// Default: 2.
RetryCount int
}
Config configures a RocketMQ broker.
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns a Config with sensible defaults for a local name server. The NameServer field is left empty and must be set before calling New.
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
Consumer implements mq.Consumer for RocketMQ using a push consumer.
type Credentials ¶
type Credentials struct {
// AccessKey is the ACL access key.
AccessKey string
// SecretKey is the ACL secret key.
SecretKey string
// SecurityToken is an optional temporary security token.
SecurityToken string
}
Credentials holds the ACL credentials for a RocketMQ cluster.
type Delivery ¶
type Delivery struct {
// contains filtered or unexported fields
}
Delivery adapts a RocketMQ primitive.MessageExt to the mq.Delivery interface.
RocketMQ push consumers acknowledge messages by returning a ConsumeResult from the subscribe callback rather than via an explicit ACK RPC. Therefore Ack/Nack/Reject here simply record the user's intent in the ackedManually/nacked flags; the consumer loop inspects these flags (together with the handler error) to decide the ConsumeResult. When AutoAck is enabled the framework ignores these flags entirely.
func (*Delivery) Ack ¶
Ack acknowledges successful processing. For RocketMQ this records the intent; the consumer loop returns ConsumeSuccess.
func (*Delivery) DeliveryTag ¶
DeliveryTag returns the queue offset as a stable delivery tag.
func (*Delivery) Nack ¶
Nack negatively acknowledges the message. If requeue is true the consumer loop returns ConsumeRetryLater so the broker redelivers; otherwise the message is dropped (routed to the retry/dead-letter topic after exceeding the max reconsume count).
func (*Delivery) Redelivered ¶
Redelivered reports whether this message has been delivered before.
func (*Delivery) RoutingKey ¶
RoutingKey returns the message tag (RocketMQ's closest analogue to a routing key).
Source Files