@Test public void testRegisterProducer() { boolean flag = mqClientInstance.registerProducer(group, mock(DefaultMQProducerImpl.class)); assertThat(flag).isTrue(); flag = mqClientInstance.registerProducer(group, mock(DefaultMQProducerImpl.class)); assertThat(flag).isFalse(); mqClientInstance.unregisterProducer(group); flag = mqClientInstance.registerProducer(group, mock(DefaultMQProducerImpl.class)); assertThat(flag).isTrue(); }
public boolean registerProducer(final String group, final DefaultMQProducerImpl producer) { if (null == group || null == producer) { return false; } MQProducerInner prev = this.producerTable.putIfAbsent(group, producer); if (prev != null) { log.warn("the producer group[{}] exist already.", group); return false; } return true; }
MQClientInstance { public boolean registerProducer(final String group, final DefaultMQProducerImpl producer) { if (null == group || null == producer) { return false; } MQProducerInner prev = this.producerTable.putIfAbsent(group, producer); if (prev != null) { log.warn("the producer group[{}] exist already.", group); return false; } return true; } }
MQClientInstance { public boolean registerProducer(final String group, final DefaultMQProducerImpl producer) { if (null == group || null == producer) { return false; } MQProducerInner prev = this.producerTable.putIfAbsent(group, producer); if (prev != null) { log.warn("the producer group[{}] exist already.", group); return false; } return true; } MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId);  MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook); }
MQClientInstance { public boolean registerProducer(final String group, final DefaultMQProducerImpl producer) { if (null == group || null == producer) { return false; } MQProducerInner prev = this.producerTable.putIfAbsent(group, producer); if (prev != null) { log.warn("the producer group[{}] exist already.", group); return false; } return true; } MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId);  MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook); static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route); static Set<MessageQueue> topicRouteData2TopicSubscribeInfo(final String topic, final TopicRouteData route); void start(); String getClientId(); void updateTopicRouteInfoFromNameServer(); Map<MessageQueue, Long> parseOffsetTableFromBroker(Map<MessageQueue, Long> offsetTable, String namespace); void checkClientInBroker(); void sendHeartbeatToAllBrokerWithLock(); void adjustThreadPool(); boolean updateTopicRouteInfoFromNameServer(final String topic); boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
        DefaultMQProducer defaultMQProducer); void shutdown(); boolean registerConsumer(final String group, final MQConsumerInner consumer); void unregisterConsumer(final String group); boolean registerProducer(final String group, final DefaultMQProducerImpl producer); void unregisterProducer(final String group); boolean registerAdminExt(final String group, final MQAdminExtInner admin); void unregisterAdminExt(final String group); void rebalanceImmediately(); void doRebalance(); MQProducerInner selectProducer(final String group); MQConsumerInner selectConsumer(final String group); FindBrokerResult findBrokerAddressInAdmin(final String brokerName); String findBrokerAddressInPublish(final String brokerName); FindBrokerResult findBrokerAddressInSubscribe(
        final String brokerName,
        final long brokerId,
        final boolean onlyThisBroker
    ); int findBrokerVersion(String brokerName, String brokerAddr); List<String> findConsumerIdList(final String topic, final String group); String findBrokerAddrByTopic(final String topic); void resetOffset(String topic, String group, Map<MessageQueue, Long> offsetTable); Map<MessageQueue, Long> getConsumerStatus(String topic, String group); TopicRouteData getAnExistTopicRouteData(final String topic); MQClientAPIImpl getMQClientAPIImpl(); MQAdminImpl getMQAdminImpl(); long getBootTimestamp(); ScheduledExecutorService getScheduledExecutorService(); PullMessageService getPullMessageService(); DefaultMQProducer getDefaultMQProducer(); ConcurrentMap<String, TopicRouteData> getTopicRouteTable(); ConsumeMessageDirectlyResult consumeMessageDirectly(final MessageExt msg,
        final String consumerGroup,
        final String brokerName); ConsumerRunningInfo consumerRunningInfo(final String consumerGroup); ConsumerStatsManager getConsumerStatsManager(); NettyClientConfig getNettyClientConfig(); ClientConfig getClientConfig(); }
MQClientInstance { public boolean registerProducer(final String group, final DefaultMQProducerImpl producer) { if (null == group || null == producer) { return false; } MQProducerInner prev = this.producerTable.putIfAbsent(group, producer); if (prev != null) { log.warn("the producer group[{}] exist already.", group); return false; } return true; } MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId);  MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook); static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route); static Set<MessageQueue> topicRouteData2TopicSubscribeInfo(final String topic, final TopicRouteData route); void start(); String getClientId(); void updateTopicRouteInfoFromNameServer(); Map<MessageQueue, Long> parseOffsetTableFromBroker(Map<MessageQueue, Long> offsetTable, String namespace); void checkClientInBroker(); void sendHeartbeatToAllBrokerWithLock(); void adjustThreadPool(); boolean updateTopicRouteInfoFromNameServer(final String topic); boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
        DefaultMQProducer defaultMQProducer); void shutdown(); boolean registerConsumer(final String group, final MQConsumerInner consumer); void unregisterConsumer(final String group); boolean registerProducer(final String group, final DefaultMQProducerImpl producer); void unregisterProducer(final String group); boolean registerAdminExt(final String group, final MQAdminExtInner admin); void unregisterAdminExt(final String group); void rebalanceImmediately(); void doRebalance(); MQProducerInner selectProducer(final String group); MQConsumerInner selectConsumer(final String group); FindBrokerResult findBrokerAddressInAdmin(final String brokerName); String findBrokerAddressInPublish(final String brokerName); FindBrokerResult findBrokerAddressInSubscribe(
        final String brokerName,
        final long brokerId,
        final boolean onlyThisBroker
    ); int findBrokerVersion(String brokerName, String brokerAddr); List<String> findConsumerIdList(final String topic, final String group); String findBrokerAddrByTopic(final String topic); void resetOffset(String topic, String group, Map<MessageQueue, Long> offsetTable); Map<MessageQueue, Long> getConsumerStatus(String topic, String group); TopicRouteData getAnExistTopicRouteData(final String topic); MQClientAPIImpl getMQClientAPIImpl(); MQAdminImpl getMQAdminImpl(); long getBootTimestamp(); ScheduledExecutorService getScheduledExecutorService(); PullMessageService getPullMessageService(); DefaultMQProducer getDefaultMQProducer(); ConcurrentMap<String, TopicRouteData> getTopicRouteTable(); ConsumeMessageDirectlyResult consumeMessageDirectly(final MessageExt msg,
        final String consumerGroup,
        final String brokerName); ConsumerRunningInfo consumerRunningInfo(final String consumerGroup); ConsumerStatsManager getConsumerStatsManager(); NettyClientConfig getNettyClientConfig(); ClientConfig getClientConfig();  }