DefaultMQProducer是RocketMQ中默认的生产者实现，DefaultMQProducer的类之间的继承关系如下所示：
image
可以看到这个生产者在实现时包含生产者的操作和配置属性，这是典型的类对象设计。
以下是一些核心属性：
namesrvAddr：继承自ClientConfig，表示RocketMQ集群的Namesrv地址，如果是多个则用分号分开。
clientIP：使用的客户端程序所在机器的IP地址，支持IPv4和IPv6，IPv4排除了本地的环回地址和私有内网地址。这了需要注意的是，如果Client运行在Docker容器中，获取的IP地址是容器所在得IP地址，而非宿主机的IP地址。
instanceName：实例名，每个实例都需要取唯一的名字，因为有时我们会在同一个机器上部署多个程序进程，如果名字有重复就会导致启动失败。
vipChannelEnabled：这是一个boolean值，表示是否开启VIP通道。VIP通道和非VIP通道区别是：在通信过程中使用的端口号不同。
clientCallBackExecutorThreads：客户端回调线程数。该参数表示Netty通信层回调线程的个数，默认值Runtime.getRuntime().availableProcessors() 表示当前CPU的有效个数。
pollNameServerInterval：获取Topic路由信息的间隔时长，单位为ms，默认为30000ms。
heartbeatBrokerInterval：与Broker心跳间隔的时长，单位为ms，默认为30000ms。
defaultMQProducerImpl：默认生产者的实现类，其中封装了Broker的各种API。如果你想自己实现一个生产者，可以添加一个新的实现，保持DefaultMQProducer对外接口不变，用户完全没有感知。
producerGroup：生产者组名，这是一个必须传递的参数。RocketMQ-way表示同一个生产者组中的生产者实例行为需要一致。
sendMsgTimeout：发送超时时间，单位为ms。
compressMsgBodyOverHowmuch：消息体的容量上限，超过该上限时消息体会通过ZIP进行压缩，该值默认为4MB。该参数在Client中是如何生效的呢？具体实现代码如下：
private boolean tryToCompressMessage(final Message msg) {
if (msg instanceof MessageBatch) {
//batch dose not support compressing right now
return false;
}
byte[] body = msg.getBody();
if (body != null) {
if (body.length >= this.defaultMQProducer.getCompressMsgBodyOverHowmuch()) {
try {
byte[] data = UtilAll.compress(body, zipCompressLevel);
if (data != null) {
msg.setBody(data);
return true;
}
} catch (IOException e) {
log.error("tryToCompressMessage exception", e);
log.warn(msg.toString());
}
}
}
return false;
}
image.gif
retryTimesWhenSendAsyncFailed：异步发送失败后重试的次数。默认是2次。异步重试是有条件的重试，并不是每次发送失败后都重试。
以下是一些核心方法：
start()：这是启动整个生产者实例的入口，主要负责校验生产者的配置参数是否正确，并启动通信通道、各种定时计划任务、Pull服务、Rebalance服务、注册生产者到Broker等操作。
shutdown()：关闭本地已注册的生产者，关闭已注册到Broker的客户端。
fetchPublishMessageQueue(Topic)：获取一个Topic有哪些Queue。在发送消息、Pull消息时需要调用。
send(Messgae msg)：同步发送普通消息。
send(Message msg, long timeout)：同步发送普通消息（超时设置）
send(Message msg,SendCallBack sendCallBack)：异步发送普通消息
send(Message msg,SendCallBack sendCallBack, long timeout)：异步发送普通消息（超时设置）
sendOneway(Message msg)：发送单向消息。只负责发送消息，不管发送结果
send(Message msg,MessageQueue mq)：同步向指定队列发送消息
send(Message msg,MessageQueue mq, long timeout)：同步向指定队列发送消息（超时设置）
同步向指定队列发送消息时，如果只有一个发送线程，在发送到某个指定队列中时，这个指定队列中的消息是有顺序的，那么就按照发送时间排序；如果某个Topic的队列都是这种情况，那么我们称该Topic的全部消息是分区有序的。
send(Message msg,MessageQueue mq, SendCallBack sendCallBack)：异步发送消到指定队列
send(Message msg,MessageQueue mq, SendCallBack sendCallBack, long timeout)：异步发送消到指定队列（超时设置）
send(Message msg,MessageQueueSelector selector,Object arg,SendCallBack sendCallBack)：自定义消息发送到指定队列。通过实现MessageQueueSelector接口来选择消息发送到哪个队列。
send(Collection<Message> msgs)：批量发送消息
下面介绍两个核心管理接口
createTopic(String key,String newTopic,int queueName)：创建Topic
viewMessage(String offsetMsgId)：根据消息id查询消息内容
生产者启动的流程比消费者启动的流程更加简单，一般用户使用DefaultMQProducer的构造函数构造一个生产者实例，并设置各种参数。比如Namesrv地址、生产者组等，调用start()方法启动生产者实例，start()方法调用了生产者默认实现类的start()方法启动，这里我们主要讲实现类的start()方法的实现：
// 第一步 调用 DefaultMQProducer 默认构造方法
public DefaultMQProducer(final String namespace, final String producerGroup, RPCHook rpcHook) {
// 设置 namespace 和生产者组
this.namespace = namespace;
this.producerGroup = producerGroup;
// 创建生产者默认实现
defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
}
// 第二步调用start方法启动生产者实例
public void start() throws MQClientException {
this.setProducerGroup(withNamespace(this.producerGroup));
// 调用默认实现的 start() 方法
this.defaultMQProducerImpl.start();
if (null != traceDispatcher) {
try {
traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
} catch (MQClientException e) {
log.warn("trace dispatcher start failed ", e);
}
}
}
public void start(final boolean startFactory) throws MQClientException {
// 服务状态 默认是创建状态，如果不是创建状态直接抛出异常
switch (this.serviceState) {
case CREATE_JUST:
// 将生产者状态设置为启动失败
this.serviceState = ServiceState.START_FAILED;
// 参数校验
this.checkConfig();
// 默认生产者组名转换为 进程id
if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
this.defaultMQProducer.changeInstanceNameToPID();
}
// 创建mq工厂
this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
// 注册生产者
boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
if (!registerOK) {
this.serviceState = ServiceState.CREATE_JUST;
throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
+ "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
null);
}
this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
// 注册本地路由信息
if (startFactory) {
mQClientFactory.start();
}
log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
this.defaultMQProducer.isSendMessageWithVIPChannel());
// 生产者状态设置为运行状态
this.serviceState = ServiceState.RUNNING;
break;
case RUNNING:
case START_FAILED:
case SHUTDOWN_ALREADY:
throw new MQClientException("The producer service state not OK, maybe started once, "
+ this.serviceState
+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
null);
default:
break;
}
// 给所有Broker发送心跳信息
this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
// 启用定时任务，每秒扫描过期任务并移除
this.startScheduledTask();
}
// mQClientFactory.start();
public void start() throws MQClientException {
synchronized (this) {
switch (this.serviceState) {
case CREATE_JUST:
// 设置服务状态为启动失败
this.serviceState = ServiceState.START_FAILED;
// If not specified,looking address from name server
// 如果没有配置 namesrv 则远程获取
if (null == this.clientConfig.getNamesrvAddr()) {
this.mQClientAPIImpl.fetchNameServerAddr();
}
// Start request-response channel
// 启动通信模块服务
this.mQClientAPIImpl.start();
// Start various schedule tasks
// 启动各种定时任务
this.startScheduledTask();
// Start pull service
// 启动消息拉取服务
this.pullMessageService.start();
// Start rebalance service
// 启动 rebalanceService 服务
this.rebalanceService.start();
// Start push service
// 启动消息推送服务
this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
log.info("the client factory [{}] start OK", this.clientId);
// 设置服务状态为运行
this.serviceState = ServiceState.RUNNING;
break;
case START_FAILED:
throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
default:
break;
}
}
}
image.gif
第一步：通过switch-case判断当前生产者的服务状态，创建时默认状态是CREATE_JUST，设置默认启动状态为启动失败。
第二步：执行checkConfig()方法。校验生产者实例设置的各种参数。比如生产者组名是否为空，是否满足命名规则、长度是否满足等。
第三步：执行changeInstanceNameToPID()方法。校验 instance name，如果是默认名字则将其修改为进程id。
第四步：执行getAndCreateMQClientInstance()方法，根据生产者组名获取获者初始化一个MQClientInstance。初始化代码如下：
public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
String clientId = clientConfig.buildMQClientId();
MQClientInstance instance = this.factoryTable.get(clientId);
if (null == instance) {
// 创建客户端实例
instance =
new MQClientInstance(clientConfig.cloneClientConfig(),
this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
if (prev != null) {
instance = prev;
log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
} else {
log.info("Created new MQClientInstance for clientId:[{}]", clientId);
}
}
return instance;
}
public String buildMQClientId() {
StringBuilder sb = new StringBuilder();
sb.append(this.getClientIP());
sb.append("@");
sb.append(this.getInstanceName());
if (!UtilAll.isBlank(this.unitName)) {
sb.append("@");
sb.append(this.unitName);
}
return sb.toString();
}
image.gif
MQClientInstance实例的功能是管理本实例中全部生产者与消费者的生产和消费行为。
下面对该实例中变量进行说明
public class MQClientInstance {
......
// 当前client实例的全部生产者的内部实例
private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>();
// 当前client实例的全部消费者的内部实例
private final ConcurrentMap<String/* group */, MQConsumerInner> consumerTable = new ConcurrentHashMap<String, MQConsumerInner>();
// 当前client实例的全部管理实例
private final ConcurrentMap<String/* group */, MQAdminExtInner> adminExtTable = new ConcurrentHashMap<String, MQAdminExtInner>();
private final NettyClientConfig nettyClientConfig;
// 其实每个client也是一个netty server,也会支持Broker访问，这里实现了全部client支持的接口
private final MQClientAPIImpl mQClientAPIImpl;
// 管理接口的本地实现类
private final MQAdminImpl mQAdminImpl;
// 当前生产者、消费者中全部Topic的本地缓存路由信息
private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();
private final Lock lockNamesrv = new ReentrantLock();
private final Lock lockHeartbeat = new ReentrantLock();
private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */,
String/* address */>> brokerAddrTable =
new ConcurrentHashMap<String, HashMap<Long, String>>();
private final ConcurrentMap<String/* Broker Name */, HashMap<String/* address */, Integer>> brokerVersionTable =
new ConcurrentHashMap<String, HashMap<String, Integer>>();
// 本地定时任务，比如定期获取当前 Namesrv 地址，定期同步Namesrv信息，定期更新Topic路由信息，定期发送心跳信息给Broker、定期清理已下线的Broker、定期持久化消费位点、定期调整消费线程数
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "MQClientFactoryScheduledThread");
}
});
// 请求的处理器，从处理方法 processRequest() 中我们可以知道目前支持哪些功能接口
private final ClientRemotingProcessor clientRemotingProcessor;
// Pull服务
private final PullMessageService pullMessageService;
// 重新平衡服务。定期指定重新平衡方法，
private final RebalanceService rebalanceService;
private final DefaultMQProducer defaultMQProducer;
// 消费监控
private final ConsumerStatsManager consumerStatsManager;
public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId) {
this(clientConfig, instanceIndex, clientId, null);
}
public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) {
this.clientConfig = clientConfig;
this.instanceIndex = instanceIndex;
this.nettyClientConfig = new NettyClientConfig();
this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads());
this.nettyClientConfig.setUseTLS(clientConfig.isUseTLS());
this.clientRemotingProcessor = new ClientRemotingProcessor(this);
this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook, clientConfig);
if (this.clientConfig.getNamesrvAddr() != null) {
this.mQClientAPIImpl.updateNameServerAddressList(this.clientConfig.getNamesrvAddr());
log.info("user specified name server address: {}", this.clientConfig.getNamesrvAddr());
}
this.clientId = clientId;
this.mQAdminImpl = new MQAdminImpl(this);
this.pullMessageService = new PullMessageService(this);
this.rebalanceService = new RebalanceService(this);
this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);
this.defaultMQProducer.resetClientConfig(clientConfig);
this.consumerStatsManager = new ConsumerStatsManager(this.scheduledExecutorService);
log.info("Created a new client Instance, InstanceIndex:{}, ClientID:{}, ClientConfig:{}, ClientVersion:{}, SerializerType:{}",
this.instanceIndex,
this.clientId,
this.clientConfig,
MQVersion.getVersionDesc(MQVersion.CURRENT_VERSION), RemotingCommand.getSerializeTypeConfigInThisServer());
}
// 从多个Namesrv中获取最新Topic路由信息，更新本地缓存
public void updateTopicRouteInfoFromNameServer() {
Set<String> topicList = new HashSet<String>();
// Consumer
{
Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, MQConsumerInner> entry = it.next();
MQConsumerInner impl = entry.getValue();
if (impl != null) {
Set<SubscriptionData> subList = impl.subscriptions();
if (subList != null) {
for (SubscriptionData subData : subList) {
topicList.add(subData.getTopic());
}
}
}
}
}
// Producer
{
Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, MQProducerInner> entry = it.next();
MQProducerInner impl = entry.getValue();
if (impl != null) {
Set<String> lst = impl.getPublishTopicList();
topicList.addAll(lst);
}
}
}
for (String topic : topicList) {
this.updateTopicRouteInfoFromNameServer(topic);
}
}
/**
* Remove offline broker 检查Client是在Broker中有效
*/
private void cleanOfflineBroker() {
try {
if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS))
try {
ConcurrentHashMap<String, HashMap<Long, String>> updatedTable = new ConcurrentHashMap<String, HashMap<Long, String>>();
Iterator<Entry<String, HashMap<Long, String>>> itBrokerTable = this.brokerAddrTable.entrySet().iterator();
while (itBrokerTable.hasNext()) {
Entry<String, HashMap<Long, String>> entry = itBrokerTable.next();
String brokerName = entry.getKey();
HashMap<Long, String> oneTable = entry.getValue();
HashMap<Long, String> cloneAddrTable = new HashMap<Long, String>();
cloneAddrTable.putAll(oneTable);
Iterator<Entry<Long, String>> it = cloneAddrTable.entrySet().iterator();
while (it.hasNext()) {
Entry<Long, String> ee = it.next();
String addr = ee.getValue();
if (!this.isBrokerAddrExistInTopicRouteTable(addr)) {
it.remove();
log.info("the broker addr[{} {}] is offline, remove it", brokerName, addr);
}
}
if (cloneAddrTable.isEmpty()) {
itBrokerTable.remove();
log.info("the broker[{}] name's host is offline, remove it", brokerName);
} else {
updatedTable.put(brokerName, cloneAddrTable);
}
}
if (!updatedTable.isEmpty()) {
this.brokerAddrTable.putAll(updatedTable);
}
} finally {
this.lockNamesrv.unlock();
}
} catch (InterruptedException e) {
log.warn("cleanOfflineBroker Exception", e);
}
}
// 发送客户端的心跳信息给所有的Broker
public void sendHeartbeatToAllBrokerWithLock() {
if (this.lockHeartbeat.tryLock()) {
try {
this.sendHeartbeatToAllBroker();
this.uploadFilterClassSource();
} catch (final Exception e) {
log.error("sendHeartbeatToAllBroker exception", e);
} finally {
this.lockHeartbeat.unlock();
}
} else {
log.warn("lock heartBeat, but failed. [{}]", this.clientId);
}
}
// 在本地注册一个消费者
public synchronized boolean registerConsumer(final String group, final MQConsumerInner consumer) {
if (null == group || null == consumer) {
return false;
}
MQConsumerInner prev = this.consumerTable.putIfAbsent(group, consumer);
if (prev != null) {
log.warn("the consumer group[" + group + "] exist already.");
return false;
}
return true;
}
// 取消本地注册的消费者
public synchronized void unregisterConsumer(final String group) {
this.consumerTable.remove(group);
this.unregisterClient(null, group);
}
// 在本地注册一个生产者
public synchronized 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;
}
// 取消本地注册的生产者
public synchronized void unregisterProducer(final String group) {
this.producerTable.remove(group);
this.unregisterClient(group, null);
}
// 注册一个管理实例
public boolean registerAdminExt(final String group, final MQAdminExtInner admin) {
if (null == group || null == admin) {
return false;
}
MQAdminExtInner prev = this.adminExtTable.putIfAbsent(group, admin);
if (prev != null) {
log.warn("the admin group[{}] exist already.", group);
return false;
}
return true;
}
// 立即执行一次 rebalance
public void rebalanceImmediately() {
this.rebalanceService.wakeup();
}
// 对于所有已经注册的消费者实例，执行一次 rebalance
public void doRebalance() {
for (Map.Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) {
MQConsumerInner impl = entry.getValue();
if (impl != null) {
try {
impl.doRebalance();
} catch (Throwable e) {
log.error("doRebalance exception", e);
}
}
}
}
// 在本地缓存中查询master或salve信息
public FindBrokerResult findBrokerAddressInAdmin(final String brokerName) {
String brokerAddr = null;
boolean slave = false;
boolean found = false;
HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
if (map != null && !map.isEmpty()) {
for (Map.Entry<Long, String> entry : map.entrySet()) {
Long id = entry.getKey();
brokerAddr = entry.getValue();
if (brokerAddr != null) {
found = true;
if (MixAll.MASTER_ID == id) {
slave = false;
} else {
slave = true;
}
break;
}
} // end of for
}
if (found) {
return new FindBrokerResult(brokerAddr, slave, findBrokerVersion(brokerName, brokerAddr));
}
return null;
}
// 在本地缓存中查询master信息
public String findBrokerAddressInPublish(final String brokerName) {
HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
if (map != null && !map.isEmpty()) {
return map.get(MixAll.MASTER_ID);
}
return null;
}
// 在本地缓存中查询salve信息
public FindBrokerResult findBrokerAddressInSubscribe(
final String brokerName,
final long brokerId,
final boolean onlyThisBroker
) {
String brokerAddr = null;
boolean slave = false;
boolean found = false;
HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
if (map != null && !map.isEmpty()) {
brokerAddr = map.get(brokerId);
slave = brokerId != MixAll.MASTER_ID;
found = brokerAddr != null;
if (!found && slave) {
brokerAddr = map.get(brokerId + 1);
found = brokerAddr != null;
}
if (!found && !onlyThisBroker) {
Entry<Long, String> entry = map.entrySet().iterator().next();
brokerAddr = entry.getValue();
slave = entry.getKey() != MixAll.MASTER_ID;
found = true;
}
}
if (found) {
return new FindBrokerResult(brokerAddr, slave, findBrokerVersion(brokerName, brokerAddr));
}
return null;
}
// 查找消费者id列表
public List<String> findConsumerIdList(final String topic, final String group) {
String brokerAddr = this.findBrokerAddrByTopic(topic);
if (null == brokerAddr) {
this.updateTopicRouteInfoFromNameServer(topic);
brokerAddr = this.findBrokerAddrByTopic(topic);
}
if (null != brokerAddr) {
try {
return this.mQClientAPIImpl.getConsumerIdListByGroup(brokerAddr, group, clientConfig.getMqClientApiTimeout());
} catch (Exception e) {
log.warn("getConsumerIdListByGroup exception, " + brokerAddr + " " + group, e);
}
}
return null;
}
// 通过 Topic 名字查找 Broker 地址
public String findBrokerAddrByTopic(final String topic) {
TopicRouteData topicRouteData = this.topicRouteTable.get(topic);
if (topicRouteData != null) {
List<BrokerData> brokers = topicRouteData.getBrokerDatas();
if (!brokers.isEmpty()) {
int index = random.nextInt(brokers.size());
BrokerData bd = brokers.get(index % brokers.size());
return bd.selectBrokerAddr();
}
}
return null;
}
// 重置消费位点
public synchronized void resetOffset(String topic, String group, Map<MessageQueue, Long> offsetTable) {
DefaultMQPushConsumerImpl consumer = null;
try {
MQConsumerInner impl = this.consumerTable.get(group);
if (impl != null && impl instanceof DefaultMQPushConsumerImpl) {
consumer = (DefaultMQPushConsumerImpl) impl;
} else {
log.info("[reset-offset] consumer dose not exist. group={}", group);
return;
}
consumer.suspend();
ConcurrentMap<MessageQueue, ProcessQueue> processQueueTable = consumer.getRebalanceImpl().getProcessQueueTable();
for (Map.Entry<MessageQueue, ProcessQueue> entry : processQueueTable.entrySet()) {
MessageQueue mq = entry.getKey();
if (topic.equals(mq.getTopic()) && offsetTable.containsKey(mq)) {
ProcessQueue pq = entry.getValue();
pq.setDropped(true);
pq.clear();
}
}
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException e) {
}
Iterator<MessageQueue> iterator = processQueueTable.keySet().iterator();
while (iterator.hasNext()) {
MessageQueue mq = iterator.next();
Long offset = offsetTable.get(mq);
if (topic.equals(mq.getTopic()) && offset != null) {
try {
consumer.updateConsumeOffset(mq, offset);
consumer.getRebalanceImpl().removeUnnecessaryMessageQueue(mq, processQueueTable.get(mq));
iterator.remove();
} catch (Exception e) {
log.warn("reset offset failed. group={}, {}", group, mq, e);
}
}
}
} finally {
if (consumer != null) {
consumer.resume();
}
}
}
// 获取一个订阅关系中每个队列的消费进度
public Map<MessageQueue, Long> getConsumerStatus(String topic, String group) {
MQConsumerInner impl = this.consumerTable.get(group);
if (impl != null && impl instanceof DefaultMQPushConsumerImpl) {
DefaultMQPushConsumerImpl consumer = (DefaultMQPushConsumerImpl) impl;
return consumer.getOffsetStore().cloneOffsetTable(topic);
} else if (impl != null && impl instanceof DefaultMQPullConsumerImpl) {
DefaultMQPullConsumerImpl consumer = (DefaultMQPullConsumerImpl) impl;
return consumer.getOffsetStore().cloneOffsetTable(topic);
} else {
return Collections.EMPTY_MAP;
}
}
// 获取本地缓存Topic路由
public ConcurrentMap<String, TopicRouteData> getTopicRouteTable() {
return topicRouteTable;
}
// 直接将消息发送给指定的消费者消费，和正常投递不同的是，指定了已经订阅的消费者组中的一个，而不是全部。一般适用于在消费消息后，某一个消费者组想再消费一次的场景
public ConsumeMessageDirectlyResult consumeMessageDirectly(final MessageExt msg,
final String consumerGroup,
final String brokerName) {
MQConsumerInner mqConsumerInner = this.consumerTable.get(consumerGroup);
if (null != mqConsumerInner) {
DefaultMQPushConsumerImpl consumer = (DefaultMQPushConsumerImpl) mqConsumerInner;
ConsumeMessageDirectlyResult result = consumer.getConsumeMessageService().consumeMessageDirectly(msg, brokerName);
return result;
}
return null;
}
// 获取消费者的消费统计信息，包含消费RT，消费tps等
public ConsumerRunningInfo consumerRunningInfo(final String consumerGroup) {
MQConsumerInner mqConsumerInner = this.consumerTable.get(consumerGroup);
if (mqConsumerInner == null) {
return null;
}
ConsumerRunningInfo consumerRunningInfo = mqConsumerInner.consumerRunningInfo();
List<String> nsList = this.mQClientAPIImpl.getRemotingClient().getNameServerAddressList();
StringBuilder strBuilder = new StringBuilder();
if (nsList != null) {
for (String addr : nsList) {
strBuilder.append(addr).append(";");
}
}
String nsAddr = strBuilder.toString();
consumerRunningInfo.getProperties().put(ConsumerRunningInfo.PROP_NAMESERVER_ADDR, nsAddr);
consumerRunningInfo.getProperties().put(ConsumerRunningInfo.PROP_CONSUME_TYPE, mqConsumerInner.consumeType().name());
consumerRunningInfo.getProperties().put(ConsumerRunningInfo.PROP_CLIENT_VERSION,
MQVersion.getVersionDesc(MQVersion.CURRENT_VERSION));
return consumerRunningInfo;
}
}
image.gif
