public TopicList getSystemTopicListFromBroker(final String addr, final long timeoutMillis)
        throws RemotingException, MQClientException, InterruptedException {
    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_BROKER, null);

    RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),
            request, timeoutMillis);
    assert response != null;
    switch (response.getCode()) {
        case ResponseCode.SUCCESS: {
            byte[] body = response.getBody();
            if (body != null) {
                TopicList topicList = TopicList.decode(body, TopicList.class);
                return topicList;
            }
        }
        default:
            break;
    }

    throw new MQClientException(response.getCode(), response.getRemark());
}
 

@Override
public boolean cleanUnusedTopic(String cluster) throws RemotingConnectException, RemotingSendRequestException,
        RemotingTimeoutException, MQClientException, InterruptedException {
    boolean result = false;
    try {
        ClusterInfo clusterInfo = examineBrokerClusterInfo();
        if (null == cluster || "".equals(cluster)) {
            for (String targetCluster : clusterInfo.retrieveAllClusterNames()) {
                result = cleanUnusedTopicByCluster(clusterInfo, targetCluster);
            }
        }
        else {
            result = cleanUnusedTopicByCluster(clusterInfo, cluster);
        }
    }
    catch (MQBrokerException e) {
        log.error("cleanExpiredConsumerQueue error.", e);
    }

    return result;
}
 

public static void main(String[] args) throws MQClientException, InterruptedException {
    DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
    producer.start();

    try {
        for (int i = 0; i < 6000000; i++) {
            Message msg = new Message("TopicFilter7", // topic
                "TagA", // tag
                "OrderID001", // key
                ("Hello MetaQ").getBytes());// body

            msg.putUserProperty("SequenceId", String.valueOf(i));

            SendResult sendResult = producer.send(msg);
            System.out.println(sendResult);
        }
    }
    catch (Exception e) {
        e.printStackTrace();
    }

    producer.shutdown();
}
 

@Override
public GroupList queryTopicConsumeByWho(String topic) throws InterruptedException, MQBrokerException, RemotingException,
        MQClientException {
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);

    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            return this.mqClientInstance.getMQClientAPIImpl().queryTopicConsumeByWho(addr, topic, timeoutMillis);
        }

        break;
    }

    return null;
}
 

public long minOffset(MessageQueue mq) throws MQClientException {
    String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
    if (null == brokerAddr) {
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
        brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
    }

    if (brokerAddr != null) {
        try {
            return this.mQClientFactory.getMQClientAPIImpl().getMinOffset(brokerAddr, mq.getTopic(),
                mq.getQueueId(), 1000 * 3);
        }
        catch (Exception e) {
            throw new MQClientException("Invoke Broker[" + brokerAddr + "] exception", e);
        }
    }

    throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
}
 

/**
 * 克隆某一个组的消费进度到新的组
 * 
 * @param addr
 * @param srcGroup
 * @param destGroup
 * @param topic
 * @param isOffline
 * @param timeoutMillis
 * @throws RemotingException
 * @throws MQClientException
 * @throws InterruptedException
 */
public void cloneGroupOffset(final String addr, final String srcGroup, final String destGroup,
        final String topic, final boolean isOffline, final long timeoutMillis)
                throws RemotingException, MQClientException, InterruptedException {
    CloneGroupOffsetRequestHeader requestHeader = new CloneGroupOffsetRequestHeader();
    requestHeader.setSrcGroup(srcGroup);
    requestHeader.setDestGroup(destGroup);
    requestHeader.setTopic(topic);
    requestHeader.setOffline(isOffline);
    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CLONE_GROUP_OFFSET, null);

    RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
    assert response != null;
    switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        return;
    }
    default:
        break;
    }

    throw new MQClientException(response.getCode(), response.getRemark());
}
 

public static void main(String[] args) throws MQClientException {
    DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
    consumer.start();

    try {
        MessageQueue mq = new MessageQueue();
        mq.setQueueId(0);
        mq.setTopic("TopicTest3");
        mq.setBrokerName("vivedeMacBook-Pro.local");

        long offset = 26;

        long beginTime = System.currentTimeMillis();
        PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, offset, 32);
        System.out.println(System.currentTimeMillis() - beginTime);
        System.out.println(pullResult);
    }
    catch (Exception e) {
        e.printStackTrace();
    }

    consumer.shutdown();
}
 

@Override
public TopicStatsTable examineTopicStats(String topic) throws RemotingException, MQClientException, InterruptedException,
        MQBrokerException {
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
    TopicStatsTable topicStatsTable = new TopicStatsTable();

    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            TopicStatsTable tst = this.mqClientInstance.getMQClientAPIImpl().getTopicStatsInfo(addr, topic, timeoutMillis);
            topicStatsTable.getOffsetTable().putAll(tst.getOffsetTable());
        }
    }

    if (topicStatsTable.getOffsetTable().isEmpty()) {
        throw new MQClientException("Not found the topic stats info", null);
    }

    return topicStatsTable;
}
 

public long earliestMsgStoreTime(MessageQueue mq) throws MQClientException {
    String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
    if (null == brokerAddr) {
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
        brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
    }

    if (brokerAddr != null) {
        try {
            return this.mQClientFactory.getMQClientAPIImpl().getEarliestMsgStoretime(brokerAddr,
                mq.getTopic(), mq.getQueueId(), 1000 * 3);
        }
        catch (Exception e) {
            throw new MQClientException("Invoke Broker[" + brokerAddr + "] exception", e);
        }
    }

    throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
}
 

void queryByKey(final DefaultMQAdminExt admin, final String topic, final String key)
        throws MQClientException, InterruptedException {
    admin.start();

    QueryResult queryResult = admin.queryMessage(topic, key, 64, 0, Long.MAX_VALUE);
    System.out.printf("%-50s %4s %40s\n", //
        "#Message ID", //
        "#QID", //
        "#Offset");
    for (MessageExt msg : queryResult.getMessageList()) {
        System.out.printf("%-50s %4d %40d\n", msg.getMsgId(), msg.getQueueId(), msg.getQueueOffset());
    }
}
 

public static void main(String[] args) throws MQClientException {
    DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");

    consumer.start();

    Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest1");
    for (MessageQueue mq : mqs) {
        System.out.println("Consume from the queue: " + mq);
        SINGLE_MQ: while (true) {
            try {
                PullResult pullResult =
                        consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
                System.out.println(pullResult);
                putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
                switch (pullResult.getPullStatus()) {
                case FOUND:
                    break;
                case NO_MATCHED_MSG:
                    break;
                case NO_NEW_MSG:
                    break SINGLE_MQ;
                case OFFSET_ILLEGAL:
                    break;
                default:
                    break;
                }
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    consumer.shutdown();
}
 

@Override
public void start() {

    ClientLogger.setLog(new RMQClientLogger());

    try {
        producer.start();
    }
    catch (MQClientException e) {
        log.err("com.creditease.uav.mq.rocketmq.RocketMQProducer.start", "MQProducer启动失败", e);
    }
}
 

private String computPullFromWhichFilterServer(final String topic, final String brokerAddr)
        throws MQClientException {
    ConcurrentHashMap<String, TopicRouteData> topicRouteTable = this.mQClientFactory.getTopicRouteTable();
    if (topicRouteTable != null) {
        TopicRouteData topicRouteData = topicRouteTable.get(topic);
        List<String> list = topicRouteData.getFilterServerTable().get(brokerAddr);

        if (list != null && !list.isEmpty()) {
            return list.get(randomNum() % list.size());
        }
    }

    throw new MQClientException(
        "Find Filter Server Failed, Broker Addr: " + brokerAddr + " topic: " + topic, null);
}
 

public static void main(String[] args) throws MQClientException, InterruptedException {
    CommandLine commandLine = buildCommandline(args);
    if (commandLine != null) {
        String group = commandLine.getOptionValue('g');
        String topic = commandLine.getOptionValue('t');
        String tags = commandLine.getOptionValue('a');
        String keys = commandLine.getOptionValue('k');
        String msgCount = commandLine.getOptionValue('c');

        DefaultMQProducer producer = new DefaultMQProducer(group);
        producer.setInstanceName(Long.toString(System.currentTimeMillis()));

        producer.start();

        for (int i = 0; i < Integer.parseInt(msgCount); i++) {
            try {
                Message msg = new Message(//
                    topic,// topic
                    tags,// tag
                    keys,// key
                    ("Hello RocketMQ " + i).getBytes());// body
                SendResult sendResult = producer.send(msg);

                System.out.printf("%-8d %s\n", i, sendResult);
            }
            catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }

        producer.shutdown();
    }
}
 

/**
 * 获取含有单元化订阅组的非单元化 Topic 列表
 * 
 * @param containRetry
 * @param timeoutMillis
 * @return
 * @throws RemotingException
 * @throws MQClientException
 * @throws InterruptedException
 */
public TopicList getHasUnitSubUnUnitTopicList(final boolean containRetry, final long timeoutMillis)
        throws RemotingException, MQClientException, InterruptedException {
    RemotingCommand request =
            RemotingCommand.createRequestCommand(RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST, null);

    RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);
    assert response != null;
    switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        byte[] body = response.getBody();
        if (body != null) {
            TopicList topicList = TopicList.decode(response.getBody(), TopicList.class);
            if (!UtilAll.isBlank(projectGroupPrefix)) {
                HashSet<String> newTopicSet = new HashSet<String>();
                for (String topic : topicList.getTopicList()) {
                    newTopicSet.add(VirtualEnvUtil.clearProjectGroup(topic, projectGroupPrefix));
                }
                topicList.setTopicList(newTopicSet);
            }
            if (!containRetry) {
                Iterator<String> it = topicList.getTopicList().iterator();
                while (it.hasNext()) {
                    String topic = it.next();
                    if (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX))
                        it.remove();
                }
            }
            return topicList;
        }
    }
    default:
        break;
    }

    throw new MQClientException(response.getCode(), response.getRemark());
}
 

public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag)
        throws MQClientException {
    this.makeSureStateOK();
    Validators.checkTopic(newTopic);

    this.mQClientFactory.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag);
}
 

public TopicList getHasUnitSubUnUnitTopicList(final boolean containRetry, final long timeoutMillis) throws RemotingException,
        MQClientException, InterruptedException {
    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST, null);

    RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);
    assert response != null;
    switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        byte[] body = response.getBody();
        if (body != null) {
            TopicList topicList = TopicList.decode(response.getBody(), TopicList.class);
            if (!containRetry) {
                Iterator<String> it = topicList.getTopicList().iterator();
                while (it.hasNext()) {
                    String topic = it.next();
                    if (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX))
                        it.remove();
                }
            }
            return topicList;
        }
    }
    default:
        break;
    }

    throw new MQClientException(response.getCode(), response.getRemark());
}
 

/**
 *
 * 通知 broker 调用 offset 重置处理
 * 
 * @param namespace
 * @param projectGroup
 * @param timeoutMillis
 * @throws RemotingException
 * @throws MQClientException
 * @throws InterruptedException
 */
public void deleteKVConfigByValue(final String namespace, final String projectGroup,
        final long timeoutMillis) throws RemotingException, MQClientException, InterruptedException {
    DeleteKVConfigRequestHeader requestHeader = new DeleteKVConfigRequestHeader();
    requestHeader.setNamespace(namespace);
    requestHeader.setKey(projectGroup);

    RemotingCommand request =
            RemotingCommand.createRequestCommand(RequestCode.DELETE_KV_CONFIG_BY_VALUE, requestHeader);

    List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();

    if (nameServerAddressList != null) {
        RemotingCommand errResponse = null;
        for (String namesrvAddr : nameServerAddressList) {
            RemotingCommand response =
                    this.remotingClient.invokeSync(namesrvAddr, request, timeoutMillis);
            assert response != null;
            switch (response.getCode()) {
            case ResponseCode.SUCCESS: {
                break;
            }
            default:
                errResponse = response;
            }
        }
        if (errResponse != null) {
            throw new MQClientException(errResponse.getCode(), errResponse.getRemark());
        }
    }
}
 

@Override
public List<QueueTimeSpan> queryConsumeTimeSpan(final String topic, final String group) throws InterruptedException, MQBrokerException,
        RemotingException, MQClientException {
    List<QueueTimeSpan> spanSet = new ArrayList<QueueTimeSpan>();
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            spanSet.addAll(this.mqClientInstance.getMQClientAPIImpl().queryConsumeTimeSpan(addr, topic, group, timeoutMillis));
        }
    }
    return spanSet;
}
 

private String computPullFromWhichFilterServer(final String topic, final String brokerAddr)
        throws MQClientException {
    ConcurrentHashMap<String, TopicRouteData> topicRouteTable = this.mQClientFactory.getTopicRouteTable();
    if (topicRouteTable != null) {
        TopicRouteData topicRouteData = topicRouteTable.get(topic);
        List<String> list = topicRouteData.getFilterServerTable().get(brokerAddr);

        if (list != null && !list.isEmpty()) {
            return list.get(randomNum() % list.size());
        }
    }

    throw new MQClientException("Find Filter Server Failed, Broker Addr: " + brokerAddr + " topic: "
            + topic, null);
}
 

/**
 * 通过Broker直接向某个Consumer发送一条消息，并立刻消费，返回结果给broker，再返回给调用方
 *
 * @param addr
 * @param consumerGroup
 * @param clientId
 * @param msgId
 * @param timeoutMillis
 * @return
 * @throws RemotingException
 * @throws MQClientException
 * @throws InterruptedException
 */
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String addr, String consumerGroup,
        String clientId, String msgId, final long timeoutMillis)
                throws RemotingException, MQClientException, InterruptedException {
    ConsumeMessageDirectlyResultRequestHeader requestHeader =
            new ConsumeMessageDirectlyResultRequestHeader();
    requestHeader.setConsumerGroup(consumerGroup);
    requestHeader.setClientId(clientId);
    requestHeader.setMsgId(msgId);

    RemotingCommand request =
            RemotingCommand.createRequestCommand(RequestCode.CONSUME_MESSAGE_DIRECTLY, requestHeader);

    RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
    assert response != null;
    switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        byte[] body = response.getBody();
        if (body != null) {
            ConsumeMessageDirectlyResult info =
                    ConsumeMessageDirectlyResult.decode(body, ConsumeMessageDirectlyResult.class);
            return info;
        }
    }
    default:
        break;
    }

    throw new MQClientException(response.getCode(), response.getRemark());
}
 

public PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums, long timeout)
        throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
    return this.pullSyncImpl(mq, subExpression, offset, maxNums, false, timeout);
}
 

void createAndUpdateKvConfig(String namespace, String key, String value) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;
 

public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
    createTopic(key, newTopic, queueNum, 0);
}
 

public static void main(String[] args) throws InterruptedException, MQClientException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_1");

    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

    consumer.setMessageModel(MessageModel.BROADCASTING);

    consumer.subscribe("TopicTest", "TagA || TagC || TagD");

    consumer.registerMessageListener(new MessageListenerConcurrently() {

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
            System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });

    consumer.start();

    System.out.println("Broadcast Consumer Started.");
}
 

@Override
public ConsumeStats examineConsumeStats(String consumerGroup, String topic)
        throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
    return MQAdminInstance.threadLocalMQAdminExt().examineConsumeStats(consumerGroup, topic);
}
 

@Override
public void deleteTopicInNameServer(Set<String> addrs, String topic)
        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
    defaultMQAdminExtImpl.deleteTopicInNameServer(addrs, topic);
}
 

public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag)
        throws MQClientException {
    this.makeSureStateOK();
    this.mQClientFactory.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag);
}
 

void sendOneway(final Message msg) throws MQClientException, RemotingException,
InterruptedException;
 

@Override
public TransactionSendResult sendMessageInTransaction(Message msg, LocalTransactionExecuter tranExecuter, final Object arg)
        throws MQClientException {
    throw new RuntimeException("sendMessageInTransaction not implement, please use TransactionMQProducer class");
}