rocketmq安装与基本操作

Posted 2020-09-20 太白的技术博客

如果不是因为政治原因，就rocketmq的社区活跃度、版本、特性和文档完善度，我是无论如何也不会使用rocketmq的。

rocketmq严格意义上并不支持高可靠性，因为其持久化只支持异步，有另外一个线程flush，不支持配置同步刷新到磁盘。只能说多个节点宕机的概率很低很低，外加现在的服务器一般都是UPS。

rocketmq官方提供了一份与activemq，kafka的特性对比（但没有包括与rabbitmq的比较，可参考http://www.cnblogs.com/zhjh256/p/6985889.html）。引用如下：

Messaging Product	Client SDK	Protocol and Specification	Order Message	Message Filter	Server Triggered Redelivery	Persistent Message	Retroactive Consumers	Message Priority	High Availability and Failover	Message Track	Configuration	Management and Operation Tools
ActiveMQ	Java, .NET, C++ etc.	Push model, support OpenWire, STOMP, AMQP, MQTT, JMS	Exclusive Consumer or Exclusive Queues can ensure ordering	Supported	Not Supported	Supports very fast persistence using JDBC along with a high performance journal，such as levelDB, kahaDB	Supported	Supported	Supported, depending on storage,if using kahadb it requires a ZooKeeper server	Not Supported	The default configuration is low level, user need to optimize the configuration parameters	Supported
Kafka	Java, Scala etc.	Pull model, support TCP	Ensure ordering of messages within a partition	Supported, you can use Kafka Streams to filter messages	Not Supported	High performance file storage	Supported offset indicate	Not Supported	Supported, requires a ZooKeeper server	Not Supported	Kafka uses key-value pairs format for configuration. These values can be supplied either from a file or programmatically.	Supported, use terminal command to expose core metrics
RocketMQ	Java, .NET, C++	Pull model, support TCP, JMS	Ensure strict ordering of messages, have no hot spot problem,and can scale out gracefully	Supported, you can even upload yourself custom-built filter code snippets	Supported	High performance and low latency file storage	Supported timestamp and offset 2 indicates	Not Supported	Supported, Master-Slave model, without another kit	Supported	Work out of box,user only need to pay attention to a few configurations	Supported, rich web and terminal command to expose core metrics

rocketmq的整体架构如下：

集群本身没有什么特殊之处，唯一的差别就是阿里啥都喜欢搞分布式，加了注册中心的概念（说白了就是抄袭zookeeper理念）。所以，要使用rocketmq，至少需要启动两个进程，nameserver、broker，前者是各种topic注册中心，后者是真正的broker。

从apache rocketmq官方http://rocketmq.apache.org/release_notes/release-notes-4.0.0-incubating/现在最新版本二进制，当前是4.0.0孵化版。

下载解压后得到apache-rocketmq-all。

在启动前，如果测试服务器配置较低，则应该修改启动命令runserver.sh（mqnamesrv里面调用）和runbroker.sh（mqbroker里面调用），将Xmx和Xms、Xmn值注释或者降低，否则可能会在启动时报内存分配失败。

 

首先启动nameserver。

nohup sh bin/mqnamesrv &

The Name Server boot success...
默认情况下，nameserver监听的是9876端口。

其次启动broker。

nohup sh bin/mqbroker -n localhost:9876 &

The broker[%s, 172.30.30.233:10911] boot success...

默认情况下，监听的是10911端口。

安装rocket web console。

 默认情况下，rocketmq没有提供二进制监控控制台，需要自己build，可从github下载maven源码builder。为方便，笔者打包了一份，https://pan.baidu.com/s/1jIp6Age。放到tomcat下启动后，如下：

 

下载后，只要更改WEB-INF\\classes\\config.properties中的rocketmq.namesrv.addr为nameserver地址即可。

停止的时候，顺序相反，要先停止broker，其次停止nameserver。

sh bin/mqshutdown broker
The mqbroker(36695) is running...
Send shutdown request to mqbroker(36695) OK

sh bin/mqshutdown namesrv
The mqnamesrv(36664) is running...
Send shutdown request to mqnamesrv(36664) OK

同其他应用服务器应用一样，rocketmq也提供了一些命令行工具用于不需要通过编程API可以管理mq服务器本身，主命令是mqadmin，支持的子命令包括：

The most commonly used mqadmin commands are:
   updateTopic          Update or create topic
   deleteTopic          Delete topic from broker and NameServer.
   updateSubGroup       Update or create subscription group
   deleteSubGroup       Delete subscription group from broker.
   updateBrokerConfig   Update broker\'s config
   updateTopicPerm      Update topic perm
   topicRoute           Examine topic route info
   topicStatus          Examine topic Status info
   topicClusterList     get cluster info for topic
   brokerStatus         Fetch broker runtime status data
   queryMsgById         Query Message by Id
   queryMsgByKey        Query Message by Key
   queryMsgByUniqueKey  Query Message by Unique key
   queryMsgByOffset     Query Message by offset
   queryMsgByUniqueKey  Query Message by Unique key
   printMsg             Print Message Detail
   sendMsgStatus        send msg to broker.
   brokerConsumeStats   Fetch broker consume stats data
   producerConnection   Query producer\'s socket connection and client version
   consumerConnection   Query consumer\'s socket connection, client version and subscription
   consumerProgress     Query consumers\'s progress, speed
   consumerStatus       Query consumer\'s internal data structure
   cloneGroupOffset     clone offset from other group.
   clusterList          List all of clusters
   topicList            Fetch all topic list from name server
   updateKvConfig       Create or update KV config.
   deleteKvConfig       Delete KV config.
   wipeWritePerm        Wipe write perm of broker in all name server
   resetOffsetByTime    Reset consumer offset by timestamp(without client restart).
   updateOrderConf      Create or update or delete order conf
   cleanExpiredCQ       Clean expired ConsumeQueue on broker.
   cleanUnusedTopic     Clean unused topic on broker.
   startMonitoring      Start Monitoring
   statsAll             Topic and Consumer tps stats
   syncDocs             Synchronize wiki and issue to github.com
   allocateMQ           Allocate MQ
   checkMsgSendRT       check message send response time
   clusterRT            List All clusters Message Send RT

See \'mqadmin help <command>\' for more information on a specific command.

 更多文档可参考：http://rocketmq.apache.org/docs/quick-start/

rocketmq集群安装可参考https://my.oschina.net/tantexian/blog/698109。
生产者、消费者最佳实践参考http://blog.csdn.net/zhu_tianwei/article/details/40948447。
各种简单、有序、批量、广播、过滤、轮训拉取实例可参考http://rocketmq.apache.org/docs/simple-example/。