Rocketmq文档

Rocketmq使用记录。

消息顺序写入&&负载均衡

1、在多终端生成者的场景下，通过MessageQueueSelector实现消息的负载均衡和特定场景下的顺序写入。

Message msg = new Message("testTopic", // topic
            "TagA", // tag
            "OrderID001", // key
            ("Hello lehoon " + i).getBytes());// body
    producer.send(msg, new MessageQueueSelector() {
        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
            final String keys = msg.getKeys();
            Integer id = Math.abs(keys.hashCode());
            int index = id % mqs.size();
            return mqs.get(index);
        }
    }, new SendCallback()
    {
        public void onSuccess(SendResult sendResult) {
            System.out.println(sendResult.getMsgId() + "  " + sendResult.getMessageQueue().getQueueId());
        }

        public void onException(Throwable e) {
            System.out.println("send fail....");
        }
    });

一个topic默认有master*8个队列，把指定终端的消息写入指定的queue，实现消息的负载均衡和顺序写入，消费者可以通过绑定指定的队列进行消费。

rocketmq命令

rocketadmin提供了很多帮助指令。

[root@txbdserver02 bin]# ./mqadmin
The most commonly used mqadmin commands are:
   updateTopic          创建、更新主题
   deleteTopic          从broker和命名服务器删除主题
   updateSubGroup       创建、修改组
   deleteSubGroup       从broker上删除组
   updateBrokerConfig   修改broker的配置
   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.

查看集群信息

[root@lehoon bin]# ./mqadmin help clusterList
usage: mqadmin clusterList [-h] [-i <arg>] [-m] [-n <arg>]
 -h,--help                指令使用帮助
 -i,--interval <arg>      指定间隔，单位秒
 -m,--moreStats           但因更多的状态信息
 -n,--namesrvAddr <arg>   命名服务地址列表，例如: 192.168.0.1:9876;192.168.0.2:9876


./mqadmin clusterList -n 192.168.1.182:9876