stomp

package module

v0.0.1 Latest Latest Go to latest Published: Jun 11, 2026 License: MIT Imports: 18 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/tx7do/go-wind-plugins

Links

Open Source Insights

README ¶

STOMP

STOMP即Simple (or Streaming) Text Orientated Messaging Protocol，简单(流) 文本定向消息协议，它提供了一个可互操作的连接格式，允许STOMP客户端与任意STOMP消息代理（Broker）进行交互。STOMP协议由于设计简单，易于开发客户端，因此在多种语言和多种平台上得到广泛地应用。

STOMP协议的前身是TTMP协议（一个简单的基于文本的协议），专为消息中间件设计。

STOMP是一个非常简单和容易实现的协议，其设计灵感源自于HTTP的简单性。尽管STOMP协议在服务器端的实现可能有一定的难度，但客户端的实现却很容易。例如，可以使用Telnet登录到任何的STOMP代理，并与STOMP代理进行交互。

STOMP协议与2012年10月22日发布了最新的STOMP 1.2规范。要查看STOMP 1.2规范，见： https://stomp.github.io/stomp-specification-1.2.html

STOMP协议分析

STOMP协议与HTTP协议很相似，它基于TCP协议，使用了以下命令：
CONNECT
SEND
SUBSCRIBE
UNSUBSCRIBE
BEGIN COMMIT
ABORT
ACK NACK
DISCONNECT

STOMP的客户端和服务器之间的通信是通过“帧”（Frame）实现的，每个帧由多“行”（Line）组成。
第一行包含了命令，然后紧跟键值对形式的Header内容。
第二行必须是空行。
第三行开始就是Body内容，末尾都以空字符结尾。
STOMP的客户端和服务器之间的通信是通过MESSAGE帧、RECEIPT帧或ERROR帧实现的，它们的格式相似。

支持STOMP的服务器

项目名	兼容STOMP的版本	描述
Apache Apollo	1.0 1.1 1.2	ActiveMQ的继承者 http://activemq.apache.org/apollo
Apache ActiveMQ	1.0 1.1	流行的开源消息服务器 http://activemq.apache.org/
HornetQ	1.0	来自JBoss的消息中间件 http://www.jboss.org/hornetq
RabbitMQ	1.0 1.1 1.2	基于Erlang、支持多种协议的消息Broker，通过插件支持STOMP协议http://www.rabbitmq.com/plugins.html#rabbitmq-stomp
Stampy	1.2	STOMP 1.2规范的一个Java实现 http://mrstampy.github.com/Stampy/
StompServer	1.0	一个轻量级的纯Ruby实现的STOMP服务器 http://stompserver.rubyforge.org/

使用方式

基础：发布/订阅

b := stomp.NewBroker(
    broker.WithAddress("127.0.0.1:61613"),
    broker.WithCodec("json"),
    stomp.WithAuth("admin", "admin"),
)
b.Init()
b.Connect()
defer b.Disconnect()

// 发布
b.Publish(ctx, "/queue/test", broker.NewMessage(msg))

// 订阅
sub, _ := b.Subscribe("/queue/test", handler, binder)

高级：心跳 + 超时配置

b := stomp.NewBroker(
    broker.WithAddress("127.0.0.1:61613"),
    stomp.WithHeartBeat(10*time.Second, 10*time.Second),
    stomp.WithConnectTimeout(5*time.Second),
    stomp.WithMsgSendTimeout(3*time.Second),
    stomp.WithRcvReceiptTimeout(3*time.Second),
)

高级：持久化订阅

sub, _ := b.Subscribe("/queue/test", handler, binder,
    stomp.WithDurable(),
    stomp.WithAckOnSuccess(),
)

配置选项

Broker 选项

选项	说明
`stomp.WithAuth(user, pass)`	用户名密码认证
`stomp.WithConnectTimeout(d)`	连接超时
`stomp.WithConnectHeaders(h)`	连接头信息
`stomp.WithVirtualHost(h)`	虚拟主机
`stomp.WithHeartBeat(send, recv)`	心跳间隔
`stomp.WithHeartBeatError(d)`	心跳超时
`stomp.WithMsgSendTimeout(d)`	发送超时
`stomp.WithRcvReceiptTimeout(d)`	接收回执超时

Publish 选项

选项	说明
`stomp.WithHeaders(h)`	自定义消息头
`stomp.WithReceipt(d)`	要求回执确认
`stomp.WithSuppressContentLength(d)`	抑制 Content-Length 头

选项	说明
`stomp.WithDurable()`	持久化订阅
`stomp.WithSubscribeHeaders(h)`	订阅头信息
`stomp.WithAckOnSuccess()`	处理成功自动 ACK
`stomp.WithSubscribeContext(ctx)`	自定义 context

Docker部署开发服务器

ActiveMQ

docker pull rmohr/activemq:latest

docker run -d \
      --name activemq-test \
      -p 61616:61616 \
      -p 8161:8161 \
      -p 5672:5672 \
      -p 61613:61613 \
      -p 1883:1883 \
      -p 61614:61614 \
      rmohr/activemq:latest

端口号	协议
61616	JMS
8161	UI
5672	AMQP
61613	STOMP
1883	MQTT
61614	WS

管理后台：http://localhost:8161/admin/
默认账号名密码：admin/admin

参考资料

Documentation ¶

Index ¶

Constants
func LogDebug(args ...any)
func LogDebugf(format string, args ...any)
func LogError(args ...any)
func LogErrorf(format string, args ...any)
func LogFatal(args ...any)
func LogFatalf(format string, args ...any)
func LogInfo(args ...any)
func LogInfof(format string, args ...any)
func LogWarn(args ...any)
func LogWarnf(format string, args ...any)
func NewBroker(opts ...broker.Option) broker.Broker
func WithAckOnSuccess() broker.SubscribeOption
func WithAuth(username string, password string) broker.Option
func WithConnectHeaders(h map[string]string) broker.Option
func WithConnectTimeout(ct time.Duration) broker.Option
func WithDurable() broker.SubscribeOption
func WithHeaders(h map[string]string) broker.PublishOption
func WithHeartBeat(sendTimeout, recvTimeout time.Duration) broker.Option
func WithHeartBeatError(errorTimeout time.Duration) broker.Option
func WithMsgSendTimeout(tm time.Duration) broker.Option
func WithRcvReceiptTimeout(tm time.Duration) broker.Option
func WithReceipt(_ time.Duration) broker.PublishOption
func WithSubscribeContext(ctx context.Context) broker.SubscribeOption
func WithSubscribeHeaders(headers map[string]string) broker.SubscribeOption
func WithSuppressContentLength(_ time.Duration) broker.PublishOption
func WithVirtualHost(h string) broker.Option
type ConsumerMessageCarrier
- func NewConsumerMessageCarrier(msg *stompV3.Message) ConsumerMessageCarrier
- func (c ConsumerMessageCarrier) Get(key string) string
- func (c ConsumerMessageCarrier) Keys() []string
- func (c ConsumerMessageCarrier) Set(key, val string)
type ProducerMessageCarrier
- func NewProducerMessageCarrier(msg *[]func(*frameV3.Frame) error) ProducerMessageCarrier
- func (c ProducerMessageCarrier) Get(key string) string
- func (c ProducerMessageCarrier) Keys() []string
- func (c ProducerMessageCarrier) Set(key, val string)

Constants ¶

View Source

const (
	TracerMessageSystemKey = "stomp"
	SpanNameProducer       = "stomp-producer"
	SpanNameConsumer       = "stomp-consumer"
)

Variables ¶

This section is empty.

Functions ¶

func LogDebug ¶

func LogDebug(args ...any)

func LogDebugf ¶

func LogDebugf(format string, args ...any)

func LogError ¶

func LogError(args ...any)

func LogErrorf ¶

func LogErrorf(format string, args ...any)

func LogFatal ¶

func LogFatal(args ...any)

func LogFatalf ¶

func LogFatalf(format string, args ...any)

func LogInfo ¶

func LogInfo(args ...any)

func LogInfof ¶

func LogInfof(format string, args ...any)

func LogWarn ¶

func LogWarn(args ...any)

func LogWarnf ¶

func LogWarnf(format string, args ...any)

func NewBroker ¶

func NewBroker(opts ...broker.Option) broker.Broker

func WithAckOnSuccess ¶

func WithAckOnSuccess() broker.SubscribeOption

func WithAuth ¶

func WithAuth(username string, password string) broker.Option

func WithConnectHeaders ¶

func WithConnectHeaders(h map[string]string) broker.Option

func WithConnectTimeout ¶

func WithConnectTimeout(ct time.Duration) broker.Option

func WithDurable ¶

func WithDurable() broker.SubscribeOption

func WithHeaders ¶

func WithHeaders(h map[string]string) broker.PublishOption

func WithHeartBeat ¶

func WithHeartBeat(sendTimeout, recvTimeout time.Duration) broker.Option

func WithHeartBeatError ¶

func WithHeartBeatError(errorTimeout time.Duration) broker.Option

func WithMsgSendTimeout ¶

func WithMsgSendTimeout(tm time.Duration) broker.Option

func WithRcvReceiptTimeout ¶

func WithRcvReceiptTimeout(tm time.Duration) broker.Option

func WithReceipt ¶

func WithReceipt(_ time.Duration) broker.PublishOption

func WithSubscribeContext ¶

func WithSubscribeContext(ctx context.Context) broker.SubscribeOption

func WithSubscribeHeaders ¶

func WithSubscribeHeaders(headers map[string]string) broker.SubscribeOption

func WithSuppressContentLength ¶

func WithSuppressContentLength(_ time.Duration) broker.PublishOption

func WithVirtualHost ¶

func WithVirtualHost(h string) broker.Option

Types ¶

type ConsumerMessageCarrier ¶

type ConsumerMessageCarrier struct {
	// contains filtered or unexported fields
}

func NewConsumerMessageCarrier ¶

func NewConsumerMessageCarrier(msg *stompV3.Message) ConsumerMessageCarrier

func (ConsumerMessageCarrier) Get ¶

func (c ConsumerMessageCarrier) Get(key string) string

func (ConsumerMessageCarrier) Keys ¶

func (c ConsumerMessageCarrier) Keys() []string

func (ConsumerMessageCarrier) Set ¶

func (c ConsumerMessageCarrier) Set(key, val string)

type ProducerMessageCarrier ¶

type ProducerMessageCarrier struct {
	// contains filtered or unexported fields
}

func NewProducerMessageCarrier ¶

func NewProducerMessageCarrier(msg *[]func(*frameV3.Frame) error) ProducerMessageCarrier

func (ProducerMessageCarrier) Get ¶

func (c ProducerMessageCarrier) Get(key string) string

func (ProducerMessageCarrier) Keys ¶

func (c ProducerMessageCarrier) Keys() []string

func (ProducerMessageCarrier) Set ¶

func (c ProducerMessageCarrier) Set(key, val string)

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL