rocketmqClients

README

RocketMQ 5.x

端口号	所属服务器	用途
9876	NameServer	命名服务对外端口
10911	BrokerServer	Broker 接收生产者的消息，处理消费者拉取消息。（上报给NameServer）
10909	BrokerServer	Broker 接收生产者的消息，不能处理消费者拉取消息。
10912	BrokerServer	Broker用于主从同步
8080	Proxy	Remoting 协议
8081	Proxy	gRPC 协议

架构图

Docker部署开发环境

至少启动一个NameServer，一个Broker。

5.x版本下，官方建议使用Local模式部署，即Broker和Proxy同进程部署。

创建网桥

docker network create rmq-network

NameServer

# NameServer
docker run -d \
    --name rocketmq-namesrv \
    --network rmq-network \
    -e "MAX_HEAP_SIZE=256M" \
    -e "HEAP_NEWSIZE=128M" \
    -e "JAVA_OPT_EXT=-server -Xms512M -Xmx512M -Xmn128m" \
    -p 9876:9876 \
    apache/rocketmq:5.1.4 \
    sh mqnamesrv

Broker

# Broker
docker run -d \
    --name rocketmq-broker \
    --link rocketmq-namesrv \
    --network rmq-network \
    -p 10911:10911 \
    -p 10909:10909 \
    -p 8080:8080 \
    -p 8081:8081 \
    -e "MAX_HEAP_SIZE=256M" \
    -e "HEAP_NEWSIZE=128M" \
    -e "JAVA_OPTS=-server -Xmx256M -Xms256M -Xmn128M" \
    -e "NAMESRV_ADDR=rocketmq-namesrv:9876" \
    apache/rocketmq:5.1.4 \
    sh mqbroker --enable-proxy autoCreateTopicEnable=true autoCreateSubscriptionGroup=true \
    -c /home/rocketmq/rocketmq-5.1.4/conf/broker.conf

Web控制台

docker run -d \
    --restart=always \
    --name rocketmq-dashboard \
    --link rocketmq-namesrv \
    --network rmq-network \
    -e "JAVA_OPTS=-Xmx256M -Xms256M -Xmn128M -Drocketmq.namesrv.addr=rocketmq-namesrv:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" \
    -p 9800:8080 \
    apacherocketmq/rocketmq-dashboard

• 控制台访问地址： http://localhost:9800/#/

参考资料

• 什么是消息队列RocketMQ版？
• RocketMQ 介绍及核心概念
• RocketMQ 简介
• RocketMQ 5.0：无状态代理模式的探索与实践

Documentation

Index

• func NewBroker(opts ...broker.Option) broker.Broker
• type ConsumerMessageCarrier
  • func NewConsumerMessageCarrier(msg *rmqClient.MessageView) 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 *rmqClient.Message) ProducerMessageCarrier
  • func (c ProducerMessageCarrier) Get(key string) string
  • func (c ProducerMessageCarrier) Keys() []string
  • func (c ProducerMessageCarrier) Set(key, val string)

Functions

func NewBroker

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

Types

type ConsumerMessageCarrier

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

func NewConsumerMessageCarrier

func NewConsumerMessageCarrier(msg *rmqClient.MessageView) 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 *rmqClient.Message) 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)