Java源码示例：org.apache.rocketmq.common.admin.ConsumeStats
示例1
public ConsumeStats getConsumeStats(final String addr, final String consumerGroup, final String topic,
final long timeoutMillis)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException,
MQBrokerException {
GetConsumeStatsRequestHeader requestHeader = new GetConsumeStatsRequestHeader();
requestHeader.setConsumerGroup(consumerGroup);
requestHeader.setTopic(topic);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUME_STATS, requestHeader);
RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),
request, timeoutMillis);
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
ConsumeStats consumeStats = ConsumeStats.decode(response.getBody(), ConsumeStats.class);
return consumeStats;
}
default:
break;
}
throw new MQBrokerException(response.getCode(), response.getRemark());
}
示例2
@Override
public ConsumeStats examineConsumeStats(String consumerGroup,
String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException {
String queryTopic = topic == null ? MixAll.getRetryTopic(consumerGroup) : topic;
TopicRouteData topicRouteData = this.examineTopicRouteInfo(queryTopic);
ConsumeStats result = new ConsumeStats();
for (BrokerData bd : topicRouteData.getBrokerDatas()) {
String addr = bd.selectBrokerAddr();
if (addr != null) {
ConsumeStats consumeStats =
this.mqClientInstance.getMQClientAPIImpl().getConsumeStats(addr, consumerGroup, topic, timeoutMillis * 3);
result.getOffsetTable().putAll(consumeStats.getOffsetTable());
double value = result.getConsumeTps() + consumeStats.getConsumeTps();
result.setConsumeTps(value);
}
}
if (result.getOffsetTable().isEmpty()) {
throw new MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
"Not found the consumer group consume stats, because return offset table is empty, maybe the consumer not consume any message");
}
return result;
}
示例3
public ConsumeStats getConsumeStats(final String addr, final String consumerGroup, final String topic, final long timeoutMillis)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException,
MQBrokerException {
GetConsumeStatsRequestHeader requestHeader = new GetConsumeStatsRequestHeader();
requestHeader.setConsumerGroup(consumerGroup);
requestHeader.setTopic(topic);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUME_STATS, requestHeader);
RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),
request, timeoutMillis);
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
ConsumeStats consumeStats = ConsumeStats.decode(response.getBody(), ConsumeStats.class);
return consumeStats;
}
default:
break;
}
throw new MQBrokerException(response.getCode(), response.getRemark());
}
示例4
@Override
public ConsumeStats examineConsumeStats(String consumerGroup,
String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException {
String queryTopic = topic == null ? MixAll.getRetryTopic(consumerGroup) : topic;
TopicRouteData topicRouteData = this.examineTopicRouteInfo(queryTopic);
ConsumeStats result = new ConsumeStats();
for (BrokerData bd : topicRouteData.getBrokerDatas()) {
String addr = bd.selectBrokerAddr();
if (addr != null) {
ConsumeStats consumeStats =
this.mqClientInstance.getMQClientAPIImpl().getConsumeStats(addr, consumerGroup, topic, timeoutMillis * 3);
result.getOffsetTable().putAll(consumeStats.getOffsetTable());
double value = result.getConsumeTps() + consumeStats.getConsumeTps();
result.setConsumeTps(value);
}
}
if (result.getOffsetTable().isEmpty()) {
throw new MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
"Not found the consumer group consume stats, because return offset table is empty, maybe the consumer not consume any message");
}
return result;
}
示例5
public ConsumeStats getConsumeStats(final String addr, final String consumerGroup, final String topic,
final long timeoutMillis)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException,
MQBrokerException {
GetConsumeStatsRequestHeader requestHeader = new GetConsumeStatsRequestHeader();
requestHeader.setConsumerGroup(consumerGroup);
requestHeader.setTopic(topic);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUME_STATS, requestHeader);
RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),
request, timeoutMillis);
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
ConsumeStats consumeStats = ConsumeStats.decode(response.getBody(), ConsumeStats.class);
return consumeStats;
}
default:
break;
}
throw new MQBrokerException(response.getCode(), response.getRemark());
}
示例6
public ConsumeStats getConsumeStats(final String addr, final String consumerGroup, final String topic,
final long timeoutMillis)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException,
MQBrokerException {
GetConsumeStatsRequestHeader requestHeader = new GetConsumeStatsRequestHeader();
requestHeader.setConsumerGroup(consumerGroup);
requestHeader.setTopic(topic);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUME_STATS, requestHeader);
String acturallyAddr = getActurallyBrokerAddr(addr);
RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), acturallyAddr),
request, timeoutMillis);
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
ConsumeStats consumeStats = ConsumeStats.decode(response.getBody(), ConsumeStats.class);
return consumeStats;
}
default:
break;
}
throw new MQBrokerException(response.getCode(), response.getRemark());
}
示例7
@BeforeClass
public static void init() throws NoSuchFieldException, IllegalAccessException, InterruptedException, RemotingTimeoutException, MQClientException, RemotingSendRequestException, RemotingConnectException {
mQClientAPIImpl = mock(MQClientAPIImpl.class);
defaultMQAdminExt = new DefaultMQAdminExt();
defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(defaultMQAdminExt, 1000);
Field field = DefaultMQAdminExtImpl.class.getDeclaredField("mqClientInstance");
field.setAccessible(true);
field.set(defaultMQAdminExtImpl, mqClientInstance);
field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
field.setAccessible(true);
field.set(mqClientInstance, mQClientAPIImpl);
field = DefaultMQAdminExt.class.getDeclaredField("defaultMQAdminExtImpl");
field.setAccessible(true);
field.set(defaultMQAdminExt, defaultMQAdminExtImpl);
ConsumeStatsList consumeStatsList = new ConsumeStatsList();
consumeStatsList.setBrokerAddr("127.0l.0.1:10911");
consumeStatsList.setConsumeStatsList(new ArrayList<Map<String, List<ConsumeStats>>>());
consumeStatsList.setTotalDiff(123);
when(mQClientAPIImpl.fetchConsumeStatsInBroker(anyString(), anyBoolean(), anyLong())).thenReturn(consumeStatsList);
}
示例8
@Override
public ConsumeStats examineConsumeStats(String consumerGroup,
String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException {
String retryTopic = MixAll.getRetryTopic(consumerGroup);
TopicRouteData topicRouteData = this.examineTopicRouteInfo(retryTopic);
ConsumeStats result = new ConsumeStats();
for (BrokerData bd : topicRouteData.getBrokerDatas()) {
String addr = bd.selectBrokerAddr();
if (addr != null) {
ConsumeStats consumeStats =
this.mqClientInstance.getMQClientAPIImpl().getConsumeStats(addr, consumerGroup, topic, timeoutMillis * 3);
result.getOffsetTable().putAll(consumeStats.getOffsetTable());
double value = result.getConsumeTps() + consumeStats.getConsumeTps();
result.setConsumeTps(value);
}
}
if (result.getOffsetTable().isEmpty()) {
throw new MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
"Not found the consumer group consume stats, because return offset table is empty, maybe the consumer not consume any message");
}
return result;
}
示例9
@BeforeClass
public static void init() throws NoSuchFieldException, IllegalAccessException, InterruptedException, RemotingTimeoutException, MQClientException, RemotingSendRequestException, RemotingConnectException {
mQClientAPIImpl = mock(MQClientAPIImpl.class);
defaultMQAdminExt = new DefaultMQAdminExt();
defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(defaultMQAdminExt, 1000);
Field field = DefaultMQAdminExtImpl.class.getDeclaredField("mqClientInstance");
field.setAccessible(true);
field.set(defaultMQAdminExtImpl, mqClientInstance);
field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
field.setAccessible(true);
field.set(mqClientInstance, mQClientAPIImpl);
field = DefaultMQAdminExt.class.getDeclaredField("defaultMQAdminExtImpl");
field.setAccessible(true);
field.set(defaultMQAdminExt, defaultMQAdminExtImpl);
ConsumeStatsList consumeStatsList = new ConsumeStatsList();
consumeStatsList.setBrokerAddr("127.0l.0.1:10911");
consumeStatsList.setConsumeStatsList(new ArrayList<Map<String, List<ConsumeStats>>>());
consumeStatsList.setTotalDiff(123);
when(mQClientAPIImpl.fetchConsumeStatsInBroker(anyString(), anyBoolean(), anyLong())).thenReturn(consumeStatsList);
}
示例10
public ConsumeStats getConsumeStats(final String addr, final String consumerGroup, final String topic, final long timeoutMillis)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException,
MQBrokerException {
GetConsumeStatsRequestHeader requestHeader = new GetConsumeStatsRequestHeader();
requestHeader.setConsumerGroup(consumerGroup);
requestHeader.setTopic(topic);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUME_STATS, requestHeader);
RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),
request, timeoutMillis);
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
ConsumeStats consumeStats = ConsumeStats.decode(response.getBody(), ConsumeStats.class);
return consumeStats;
}
default:
break;
}
throw new MQBrokerException(response.getCode(), response.getRemark());
}
示例11
@Override
public ConsumeStats examineConsumeStats(String consumerGroup, String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException {
String retryTopic = MixAll.getRetryTopic(consumerGroup);
TopicRouteData topicRouteData = this.examineTopicRouteInfo(retryTopic);
ConsumeStats result = new ConsumeStats();
for (BrokerData bd : topicRouteData.getBrokerDatas()) {
String addr = bd.selectBrokerAddr();
if (addr != null) {
ConsumeStats consumeStats =
this.mqClientInstance.getMQClientAPIImpl().getConsumeStats(addr, consumerGroup, topic, timeoutMillis * 3);
result.getOffsetTable().putAll(consumeStats.getOffsetTable());
double value = result.getConsumeTps() + consumeStats.getConsumeTps();
result.setConsumeTps(value);
}
}
if (result.getOffsetTable().isEmpty()) {
throw new MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
"Not found the consumer group consume stats, because return offset table is empty, maybe the consumer not consume any message");
}
return result;
}
示例12
public boolean consumed(final MessageExt msg, final String group) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
ConsumeStats cstats = this.examineConsumeStats(group);
ClusterInfo ci = this.examineBrokerClusterInfo();
Iterator<Entry<MessageQueue, OffsetWrapper>> it = cstats.getOffsetTable().entrySet().iterator();
while (it.hasNext()) {
Entry<MessageQueue, OffsetWrapper> next = it.next();
MessageQueue mq = next.getKey();
if (mq.getTopic().equals(msg.getTopic()) && mq.getQueueId() == msg.getQueueId()) {
BrokerData brokerData = ci.getBrokerAddrTable().get(mq.getBrokerName());
if (brokerData != null) {
String addr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
if (addr.equals(RemotingUtil.socketAddress2String(msg.getStoreHost()))) {
if (next.getValue().getConsumerOffset() > msg.getQueueOffset()) {
return true;
}
}
}
}
}
return false;
}
示例13
@Override
public ConsumeStats examineConsumeStats(String consumerGroup,
String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException {
String retryTopic = MixAll.getRetryTopic(consumerGroup);
TopicRouteData topicRouteData = this.examineTopicRouteInfo(retryTopic);
ConsumeStats result = new ConsumeStats();
for (BrokerData bd : topicRouteData.getBrokerDatas()) {
String addr = bd.selectBrokerAddr();
if (addr != null) {
ConsumeStats consumeStats =
this.mqClientInstance.getMQClientAPIImpl().getConsumeStats(addr, consumerGroup, topic, timeoutMillis * 3);
result.getOffsetTable().putAll(consumeStats.getOffsetTable());
double value = result.getConsumeTps() + consumeStats.getConsumeTps();
result.setConsumeTps(value);
}
}
if (result.getOffsetTable().isEmpty()) {
throw new MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
"Not found the consumer group consume stats, because return offset table is empty, maybe the consumer not consume any message");
}
return result;
}
示例14
@BeforeClass
public static void init() throws NoSuchFieldException, IllegalAccessException, InterruptedException, RemotingTimeoutException, MQClientException, RemotingSendRequestException, RemotingConnectException {
mQClientAPIImpl = mock(MQClientAPIImpl.class);
defaultMQAdminExt = new DefaultMQAdminExt();
defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(defaultMQAdminExt, 1000);
Field field = DefaultMQAdminExtImpl.class.getDeclaredField("mqClientInstance");
field.setAccessible(true);
field.set(defaultMQAdminExtImpl, mqClientInstance);
field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
field.setAccessible(true);
field.set(mqClientInstance, mQClientAPIImpl);
field = DefaultMQAdminExt.class.getDeclaredField("defaultMQAdminExtImpl");
field.setAccessible(true);
field.set(defaultMQAdminExt, defaultMQAdminExtImpl);
ConsumeStatsList consumeStatsList = new ConsumeStatsList();
consumeStatsList.setBrokerAddr("127.0l.0.1:10911");
consumeStatsList.setConsumeStatsList(new ArrayList<Map<String, List<ConsumeStats>>>());
consumeStatsList.setTotalDiff(123);
when(mQClientAPIImpl.fetchConsumeStatsInBroker(anyString(), anyBoolean(), anyLong())).thenReturn(consumeStatsList);
}
示例15
@BeforeClass
public static void init() throws NoSuchFieldException, IllegalAccessException, InterruptedException, RemotingTimeoutException, MQClientException, RemotingSendRequestException, RemotingConnectException {
mQClientAPIImpl = mock(MQClientAPIImpl.class);
defaultMQAdminExt = new DefaultMQAdminExt();
defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(defaultMQAdminExt, 1000);
Field field = DefaultMQAdminExtImpl.class.getDeclaredField("mqClientInstance");
field.setAccessible(true);
field.set(defaultMQAdminExtImpl, mqClientInstance);
field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
field.setAccessible(true);
field.set(mqClientInstance, mQClientAPIImpl);
field = DefaultMQAdminExt.class.getDeclaredField("defaultMQAdminExtImpl");
field.setAccessible(true);
field.set(defaultMQAdminExt, defaultMQAdminExtImpl);
ConsumeStatsList consumeStatsList = new ConsumeStatsList();
consumeStatsList.setBrokerAddr("127.0l.0.1:10911");
consumeStatsList.setConsumeStatsList(new ArrayList<Map<String, List<ConsumeStats>>>());
consumeStatsList.setTotalDiff(123);
when(mQClientAPIImpl.fetchConsumeStatsInBroker(anyString(), anyBoolean(), anyLong())).thenReturn(consumeStatsList);
}
示例16
@BeforeClass
public static void init() throws NoSuchFieldException, IllegalAccessException, InterruptedException, RemotingException, MQClientException, MQBrokerException {
mQClientAPIImpl = mock(MQClientAPIImpl.class);
defaultMQAdminExt = new DefaultMQAdminExt();
defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(defaultMQAdminExt, 1000);
Field field = DefaultMQAdminExtImpl.class.getDeclaredField("mqClientInstance");
field.setAccessible(true);
field.set(defaultMQAdminExtImpl, mqClientInstance);
field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
field.setAccessible(true);
field.set(mqClientInstance, mQClientAPIImpl);
field = DefaultMQAdminExt.class.getDeclaredField("defaultMQAdminExtImpl");
field.setAccessible(true);
field.set(defaultMQAdminExt, defaultMQAdminExtImpl);
TopicRouteData topicRouteData = new TopicRouteData();
List<BrokerData> brokerDatas = new ArrayList<>();
HashMap<Long, String> brokerAddrs = new HashMap<>();
brokerAddrs.put(1234l, "127.0.0.1:10911");
BrokerData brokerData = new BrokerData();
brokerData.setCluster("default-cluster");
brokerData.setBrokerName("default-broker");
brokerData.setBrokerAddrs(brokerAddrs);
brokerDatas.add(brokerData);
topicRouteData.setBrokerDatas(brokerDatas);
topicRouteData.setQueueDatas(new ArrayList<QueueData>());
topicRouteData.setFilterServerTable(new HashMap<String, List<String>>());
when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(topicRouteData);
ConsumeStats consumeStats = new ConsumeStats();
consumeStats.setConsumeTps(1234);
MessageQueue messageQueue = new MessageQueue();
OffsetWrapper offsetWrapper = new OffsetWrapper();
HashMap<MessageQueue, OffsetWrapper> stats = new HashMap<>();
stats.put(messageQueue, offsetWrapper);
consumeStats.setOffsetTable(stats);
when(mQClientAPIImpl.getConsumeStats(anyString(), anyString(), anyString(), anyLong())).thenReturn(consumeStats);
}
示例17
public boolean consumed(final MessageExt msg,
final String group) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
ConsumeStats cstats = this.examineConsumeStats(group);
ClusterInfo ci = this.examineBrokerClusterInfo();
Iterator<Entry<MessageQueue, OffsetWrapper>> it = cstats.getOffsetTable().entrySet().iterator();
while (it.hasNext()) {
Entry<MessageQueue, OffsetWrapper> next = it.next();
MessageQueue mq = next.getKey();
if (mq.getTopic().equals(msg.getTopic()) && mq.getQueueId() == msg.getQueueId()) {
BrokerData brokerData = ci.getBrokerAddrTable().get(mq.getBrokerName());
if (brokerData != null) {
String addr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
if (addr.equals(RemotingUtil.socketAddress2String(msg.getStoreHost()))) {
if (next.getValue().getConsumerOffset() > msg.getQueueOffset()) {
return true;
}
}
}
}
}
return false;
}
示例18
@Override
public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException {
String srcGroup = commandLine.getOptionValue("s").trim();
String destGroup = commandLine.getOptionValue("d").trim();
String topic = commandLine.getOptionValue("t").trim();
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName("admin-" + Long.toString(System.currentTimeMillis()));
try {
defaultMQAdminExt.start();
ConsumeStats consumeStats = defaultMQAdminExt.examineConsumeStats(srcGroup);
Set<MessageQueue> mqs = consumeStats.getOffsetTable().keySet();
if (!mqs.isEmpty()) {
TopicRouteData topicRoute = defaultMQAdminExt.examineTopicRouteInfo(topic);
for (MessageQueue mq : mqs) {
String addr = null;
for (BrokerData brokerData : topicRoute.getBrokerDatas()) {
if (brokerData.getBrokerName().equals(mq.getBrokerName())) {
addr = brokerData.selectBrokerAddr();
break;
}
}
long offset = consumeStats.getOffsetTable().get(mq).getBrokerOffset();
if (offset >= 0) {
defaultMQAdminExt.updateConsumeOffset(addr, destGroup, mq, offset);
}
}
}
System.out.printf("clone group offset success. srcGroup[%s], destGroup=[%s], topic[%s]",
srcGroup, destGroup, topic);
} catch (Exception e) {
throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
} finally {
defaultMQAdminExt.shutdown();
}
}
示例19
public boolean consumed(final MessageExt msg,
final String group) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
ConsumeStats cstats = this.examineConsumeStats(group);
ClusterInfo ci = this.examineBrokerClusterInfo();
Iterator<Entry<MessageQueue, OffsetWrapper>> it = cstats.getOffsetTable().entrySet().iterator();
while (it.hasNext()) {
Entry<MessageQueue, OffsetWrapper> next = it.next();
MessageQueue mq = next.getKey();
if (mq.getTopic().equals(msg.getTopic()) && mq.getQueueId() == msg.getQueueId()) {
BrokerData brokerData = ci.getBrokerAddrTable().get(mq.getBrokerName());
if (brokerData != null) {
String addr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
if (RemotingUtil.socketAddress2String(msg.getStoreHost()).equals(addr)) {
if (next.getValue().getConsumerOffset() > msg.getQueueOffset()) {
return true;
}
}
}
}
}
return false;
}
示例20
@Test
public void testNewGroup() throws Exception {
group = "cg_test_new_rmq_group";
topic = "test-2";
TopicStatsTable ts = mqAdmin.examineTopicStats(topic);
for (Map.Entry<MessageQueue, TopicOffset> messageQueueTopicOffsetEntry : ts.getOffsetTable().entrySet()) {
System.out.println(messageQueueTopicOffsetEntry);
}
TopicRouteData routeInfo = mqAdmin.examineTopicRouteInfo(topic);
System.out.println();
MQClientInstance instance = MQClientManager.getInstance().getAndCreateMQClientInstance(mqAdmin);
for (BrokerData brokerData : routeInfo.getBrokerDatas()) {
ConsumeStats r = instance.getMQClientAPIImpl().getConsumeStats(brokerData.selectBrokerAddr(), group, topic, 3000);
System.out.println(brokerData + "consumeStats:" + r.getOffsetTable());
}
// List<RollbackStats> r = mqAdmin.resetOffsetByTimestampOld(group, topic, -1, true);
// for (RollbackStats rollbackStats : r) {
// System.out.println(rollbackStats);
// }
// ConsumeStats cs = mqAdmin.examineConsumeStats(group, topic);
// for (Map.Entry<MessageQueue, OffsetWrapper> messageQueueOffsetWrapperEntry : cs.getOffsetTable().entrySet()) {
// System.out.println(messageQueueOffsetWrapperEntry);
// }
}
示例21
@BeforeClass
public static void init() throws NoSuchFieldException, IllegalAccessException, InterruptedException, RemotingException, MQClientException, MQBrokerException {
mQClientAPIImpl = mock(MQClientAPIImpl.class);
defaultMQAdminExt = new DefaultMQAdminExt();
defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(defaultMQAdminExt, 1000);
Field field = DefaultMQAdminExtImpl.class.getDeclaredField("mqClientInstance");
field.setAccessible(true);
field.set(defaultMQAdminExtImpl, mqClientInstance);
field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
field.setAccessible(true);
field.set(mqClientInstance, mQClientAPIImpl);
field = DefaultMQAdminExt.class.getDeclaredField("defaultMQAdminExtImpl");
field.setAccessible(true);
field.set(defaultMQAdminExt, defaultMQAdminExtImpl);
TopicRouteData topicRouteData = new TopicRouteData();
List<BrokerData> brokerDatas = new ArrayList<>();
HashMap<Long, String> brokerAddrs = new HashMap<>();
brokerAddrs.put(1234l, "127.0.0.1:10911");
BrokerData brokerData = new BrokerData();
brokerData.setCluster("default-cluster");
brokerData.setBrokerName("default-broker");
brokerData.setBrokerAddrs(brokerAddrs);
brokerDatas.add(brokerData);
topicRouteData.setBrokerDatas(brokerDatas);
topicRouteData.setQueueDatas(new ArrayList<QueueData>());
topicRouteData.setFilterServerTable(new HashMap<String, List<String>>());
when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(topicRouteData);
ConsumeStats consumeStats = new ConsumeStats();
consumeStats.setConsumeTps(1234);
MessageQueue messageQueue = new MessageQueue();
OffsetWrapper offsetWrapper = new OffsetWrapper();
HashMap<MessageQueue, OffsetWrapper> stats = new HashMap<>();
stats.put(messageQueue, offsetWrapper);
consumeStats.setOffsetTable(stats);
when(mQClientAPIImpl.getConsumeStats(anyString(), anyString(), anyString(), anyLong())).thenReturn(consumeStats);
}
示例22
public boolean consumed(final MessageExt msg,
final String group) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
ConsumeStats cstats = this.examineConsumeStats(group);
ClusterInfo ci = this.examineBrokerClusterInfo();
Iterator<Entry<MessageQueue, OffsetWrapper>> it = cstats.getOffsetTable().entrySet().iterator();
while (it.hasNext()) {
Entry<MessageQueue, OffsetWrapper> next = it.next();
MessageQueue mq = next.getKey();
if (mq.getTopic().equals(msg.getTopic()) && mq.getQueueId() == msg.getQueueId()) {
BrokerData brokerData = ci.getBrokerAddrTable().get(mq.getBrokerName());
if (brokerData != null) {
String addr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
if (addr.equals(RemotingUtil.socketAddress2String(msg.getStoreHost()))) {
if (next.getValue().getConsumerOffset() > msg.getQueueOffset()) {
return true;
}
}
}
}
}
return false;
}
示例23
@Test
public void testNewGroup() throws Exception {
group = "cg_test_new_rmq_group";
topic = "test-2";
TopicStatsTable ts = mqAdmin.examineTopicStats(topic);
for (Map.Entry<MessageQueue, TopicOffset> messageQueueTopicOffsetEntry : ts.getOffsetTable().entrySet()) {
System.out.println(messageQueueTopicOffsetEntry);
}
TopicRouteData routeInfo = mqAdmin.examineTopicRouteInfo(topic);
System.out.println();
MQClientInstance instance = MQClientManager.getInstance().getAndCreateMQClientInstance(mqAdmin);
for (BrokerData brokerData : routeInfo.getBrokerDatas()) {
ConsumeStats r = instance.getMQClientAPIImpl().getConsumeStats(brokerData.selectBrokerAddr(), group, topic, 3000);
System.out.println(brokerData + "consumeStats:" + r.getOffsetTable());
}
// List<RollbackStats> r = mqAdmin.resetOffsetByTimestampOld(group, topic, -1, true);
// for (RollbackStats rollbackStats : r) {
// System.out.println(rollbackStats);
// }
// ConsumeStats cs = mqAdmin.examineConsumeStats(group, topic);
// for (Map.Entry<MessageQueue, OffsetWrapper> messageQueueOffsetWrapperEntry : cs.getOffsetTable().entrySet()) {
// System.out.println(messageQueueOffsetWrapperEntry);
// }
}
示例24
public boolean consumed(final MessageExt msg,
final String group) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
ConsumeStats cstats = this.examineConsumeStats(group);
ClusterInfo ci = this.examineBrokerClusterInfo();
Iterator<Entry<MessageQueue, OffsetWrapper>> it = cstats.getOffsetTable().entrySet().iterator();
while (it.hasNext()) {
Entry<MessageQueue, OffsetWrapper> next = it.next();
MessageQueue mq = next.getKey();
if (mq.getTopic().equals(msg.getTopic()) && mq.getQueueId() == msg.getQueueId()) {
BrokerData brokerData = ci.getBrokerAddrTable().get(mq.getBrokerName());
if (brokerData != null) {
String addr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
if (RemotingUtil.socketAddress2String(msg.getStoreHost()).equals(addr)) {
if (next.getValue().getConsumerOffset() > msg.getQueueOffset()) {
return true;
}
}
}
}
}
return false;
}
示例25
@BeforeClass
public static void init() throws NoSuchFieldException, IllegalAccessException, InterruptedException, RemotingException, MQClientException, MQBrokerException {
mQClientAPIImpl = mock(MQClientAPIImpl.class);
defaultMQAdminExt = new DefaultMQAdminExt();
defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(defaultMQAdminExt, 1000);
Field field = DefaultMQAdminExtImpl.class.getDeclaredField("mqClientInstance");
field.setAccessible(true);
field.set(defaultMQAdminExtImpl, mqClientInstance);
field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
field.setAccessible(true);
field.set(mqClientInstance, mQClientAPIImpl);
field = DefaultMQAdminExt.class.getDeclaredField("defaultMQAdminExtImpl");
field.setAccessible(true);
field.set(defaultMQAdminExt, defaultMQAdminExtImpl);
TopicRouteData topicRouteData = new TopicRouteData();
List<BrokerData> brokerDatas = new ArrayList<>();
HashMap<Long, String> brokerAddrs = new HashMap<>();
brokerAddrs.put(1234l, "127.0.0.1:10911");
BrokerData brokerData = new BrokerData();
brokerData.setCluster("default-cluster");
brokerData.setBrokerName("default-broker");
brokerData.setBrokerAddrs(brokerAddrs);
brokerDatas.add(brokerData);
topicRouteData.setBrokerDatas(brokerDatas);
topicRouteData.setQueueDatas(new ArrayList<QueueData>());
topicRouteData.setFilterServerTable(new HashMap<String, List<String>>());
when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(topicRouteData);
ConsumeStats consumeStats = new ConsumeStats();
consumeStats.setConsumeTps(1234);
MessageQueue messageQueue = new MessageQueue();
OffsetWrapper offsetWrapper = new OffsetWrapper();
HashMap<MessageQueue, OffsetWrapper> stats = new HashMap<>();
stats.put(messageQueue, offsetWrapper);
consumeStats.setOffsetTable(stats);
when(mQClientAPIImpl.getConsumeStats(anyString(), anyString(), anyString(), anyLong())).thenReturn(consumeStats);
}
示例26
@BeforeClass
public static void init() throws NoSuchFieldException, IllegalAccessException, InterruptedException, RemotingException, MQClientException, MQBrokerException {
mQClientAPIImpl = mock(MQClientAPIImpl.class);
defaultMQAdminExt = new DefaultMQAdminExt();
defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(defaultMQAdminExt, 1000);
Field field = DefaultMQAdminExtImpl.class.getDeclaredField("mqClientInstance");
field.setAccessible(true);
field.set(defaultMQAdminExtImpl, mqClientInstance);
field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
field.setAccessible(true);
field.set(mqClientInstance, mQClientAPIImpl);
field = DefaultMQAdminExt.class.getDeclaredField("defaultMQAdminExtImpl");
field.setAccessible(true);
field.set(defaultMQAdminExt, defaultMQAdminExtImpl);
TopicRouteData topicRouteData = new TopicRouteData();
List<BrokerData> brokerDatas = new ArrayList<>();
HashMap<Long, String> brokerAddrs = new HashMap<>();
brokerAddrs.put(1234l, "127.0.0.1:10911");
BrokerData brokerData = new BrokerData();
brokerData.setCluster("default-cluster");
brokerData.setBrokerName("default-broker");
brokerData.setBrokerAddrs(brokerAddrs);
brokerDatas.add(brokerData);
topicRouteData.setBrokerDatas(brokerDatas);
topicRouteData.setQueueDatas(new ArrayList<QueueData>());
topicRouteData.setFilterServerTable(new HashMap<String, List<String>>());
when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(topicRouteData);
ConsumeStats consumeStats = new ConsumeStats();
consumeStats.setConsumeTps(1234);
MessageQueue messageQueue = new MessageQueue();
OffsetWrapper offsetWrapper = new OffsetWrapper();
HashMap<MessageQueue, OffsetWrapper> stats = new HashMap<>();
stats.put(messageQueue, offsetWrapper);
consumeStats.setOffsetTable(stats);
when(mQClientAPIImpl.getConsumeStats(anyString(), anyString(), anyString(), anyLong())).thenReturn(consumeStats);
}
示例27
public ConsumeStats getConsumeStats(String group, String topic) throws Exception {
return defaultMQAdminExt.examineConsumeStats(group, topic);
}
示例28
public ConsumeStats getConsumeStats(String group, String topic) throws Exception {
return defaultMQAdminExt.examineConsumeStats(group, topic);
}
示例29
public void setConsumeStatsList(List<Map<String, List<ConsumeStats>>> consumeStatsList) {
this.consumeStatsList = consumeStatsList;
}
示例30
private RemotingCommand getConsumeStats(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
final GetConsumeStatsRequestHeader requestHeader =
(GetConsumeStatsRequestHeader) request.decodeCommandCustomHeader(GetConsumeStatsRequestHeader.class);
ConsumeStats consumeStats = new ConsumeStats();
Set<String> topics = new HashSet<String>();
if (UtilAll.isBlank(requestHeader.getTopic())) {
topics = this.brokerController.getConsumerOffsetManager().whichTopicByConsumer(requestHeader.getConsumerGroup());
} else {
topics.add(requestHeader.getTopic());
}
for (String topic : topics) {
TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(topic);
if (null == topicConfig) {
log.warn("consumeStats, topic config not exist, {}", topic);
continue;
}
{
SubscriptionData findSubscriptionData =
this.brokerController.getConsumerManager().findSubscriptionData(requestHeader.getConsumerGroup(), topic);
if (null == findSubscriptionData
&& this.brokerController.getConsumerManager().findSubscriptionDataCount(requestHeader.getConsumerGroup()) > 0) {
log.warn("consumeStats, the consumer group[{}], topic[{}] not exist", requestHeader.getConsumerGroup(), topic);
continue;
}
}
for (int i = 0; i < topicConfig.getReadQueueNums(); i++) {
MessageQueue mq = new MessageQueue();
mq.setTopic(topic);
mq.setBrokerName(this.brokerController.getBrokerConfig().getBrokerName());
mq.setQueueId(i);
OffsetWrapper offsetWrapper = new OffsetWrapper();
long brokerOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, i);
if (brokerOffset < 0)
brokerOffset = 0;
long consumerOffset = this.brokerController.getConsumerOffsetManager().queryOffset(
requestHeader.getConsumerGroup(),
topic,
i);
if (consumerOffset < 0)
consumerOffset = 0;
offsetWrapper.setBrokerOffset(brokerOffset);
offsetWrapper.setConsumerOffset(consumerOffset);
long timeOffset = consumerOffset - 1;
if (timeOffset >= 0) {
long lastTimestamp = this.brokerController.getMessageStore().getMessageStoreTimeStamp(topic, i, timeOffset);
if (lastTimestamp > 0) {
offsetWrapper.setLastTimestamp(lastTimestamp);
}
}
consumeStats.getOffsetTable().put(mq, offsetWrapper);
}
double consumeTps = this.brokerController.getBrokerStatsManager().tpsGroupGetNums(requestHeader.getConsumerGroup(), topic);
consumeTps += consumeStats.getConsumeTps();
consumeStats.setConsumeTps(consumeTps);
}
byte[] body = consumeStats.encode();
response.setBody(body);
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
return response;
}
