本文整理汇总了Java中com.alibaba.rocketmq.common.admin.OffsetWrapper类的典型用法代码示例。如果您正苦于以下问题:Java OffsetWrapper类的具体用法?Java OffsetWrapper怎么用?Java OffsetWrapper使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
OffsetWrapper类属于com.alibaba.rocketmq.common.admin包,在下文中一共展示了OffsetWrapper类的9个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Java代码示例。
示例1: getConsumeStats
import com.alibaba.rocketmq.common.admin.OffsetWrapper; //导入依赖的package包/类
public ConsumeStats getConsumeStats(final String addr, final String consumerGroup,
final long timeoutMillis) throws InterruptedException, RemotingTimeoutException,
RemotingSendRequestException, RemotingConnectException, MQBrokerException {
// 添加虚拟运行环境相关的projectGroupPrefix
String consumerGroupWithProjectGroup = consumerGroup;
if (!UtilAll.isBlank(projectGroupPrefix)) {
consumerGroupWithProjectGroup =
VirtualEnvUtil.buildWithProjectGroup(consumerGroup, projectGroupPrefix);
}
GetConsumeStatsRequestHeader requestHeader = new GetConsumeStatsRequestHeader();
requestHeader.setConsumerGroup(consumerGroupWithProjectGroup);
RemotingCommand request =
RemotingCommand.createRequestCommand(MQRequestCode.GET_CONSUME_STATS_VALUE, requestHeader);
RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
switch (response.getCode()) {
case ResponseCode.SUCCESS_VALUE: {
ConsumeStats consumeStats = ConsumeStats.decode(response.getBody(), ConsumeStats.class);
// 清除虚拟运行环境相关的projectGroupPrefix
if (!UtilAll.isBlank(projectGroupPrefix)) {
HashMap<MessageQueue, OffsetWrapper> newTopicOffsetMap =
new HashMap<MessageQueue, OffsetWrapper>();
for (Map.Entry<MessageQueue, OffsetWrapper> messageQueue : consumeStats.getOffsetTable()
.entrySet()) {
MessageQueue key = messageQueue.getKey();
key.setTopic(VirtualEnvUtil.clearProjectGroup(key.getTopic(), projectGroupPrefix));
newTopicOffsetMap.put(key, messageQueue.getValue());
}
consumeStats.setOffsetTable(newTopicOffsetMap);
}
return consumeStats;
}
default:
break;
}
throw new MQBrokerException(response.getCode(), response.getRemark());
}
开发者ID:brucechan0921,项目名称:RocketMQ-3.0.8,代码行数:41,代码来源:MQClientAPIImpl.java
示例2: computeUndoneMsgs
import com.alibaba.rocketmq.common.admin.OffsetWrapper; //导入依赖的package包/类
private void computeUndoneMsgs(final UndoneMsgs undoneMsgs, final ConsumeStats consumeStats) {
long total = 0;
long singleMax = 0;
long delayMax = 0;
Iterator<Entry<MessageQueue, OffsetWrapper>> it = consumeStats.getOffsetTable().entrySet().iterator();
while (it.hasNext()) {
Entry<MessageQueue, OffsetWrapper> next = it.next();
MessageQueue mq = next.getKey();
OffsetWrapper ow = next.getValue();
long diff = ow.getBrokerOffset() - ow.getConsumerOffset();
if (diff > singleMax) {
singleMax = diff;
}
if (diff > 0) {
total += diff;
}
// Delay
if (ow.getLastTimestamp() > 0) {
try {
long maxOffset = this.defaultMQPullConsumer.maxOffset(mq);
if (maxOffset > 0) {
PullResult pull = this.defaultMQPullConsumer.pull(mq, "*", maxOffset - 1, 1);
switch (pull.getPullStatus()) {
case FOUND:
long delay =
pull.getMsgFoundList().get(0).getStoreTimestamp() - ow.getLastTimestamp();
if (delay > delayMax) {
delayMax = delay;
}
break;
case NO_MATCHED_MSG:
case NO_NEW_MSG:
case OFFSET_ILLEGAL:
break;
default:
break;
}
}
}
catch (Exception e) {
}
}
}
undoneMsgs.setUndoneMsgsTotal(total);
undoneMsgs.setUndoneMsgsSingleMQ(singleMax);
undoneMsgs.setUndoneMsgsDelayTimeMills(delayMax);
}
开发者ID:y123456yz,项目名称:reading-and-annotate-rocketmq-3.4.6,代码行数:52,代码来源:MonitorService.java
示例3: getConsumeStats
import com.alibaba.rocketmq.common.admin.OffsetWrapper; //导入依赖的package包/类
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().getMaxOffsetInQuque(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;
}
开发者ID:y123456yz,项目名称:reading-and-annotate-rocketmq-3.4.6,代码行数:77,代码来源:AdminBrokerProcessor.java
示例4: fromOffsetTableEntry
import com.alibaba.rocketmq.common.admin.OffsetWrapper; //导入依赖的package包/类
public static QueueStatInfo fromOffsetTableEntry(MessageQueue key, OffsetWrapper value){
QueueStatInfo queueStatInfo = new QueueStatInfo();
BeanUtils.copyProperties(key,queueStatInfo);
BeanUtils.copyProperties(value,queueStatInfo);
return queueStatInfo;
}
开发者ID:didapinchegit,项目名称:rocket-console,代码行数:7,代码来源:QueueStatInfo.java
示例5: getConsumeStats
import com.alibaba.rocketmq.common.admin.OffsetWrapper; //导入依赖的package包/类
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().getMaxOffsetInQuque(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;
}
开发者ID:medusar,项目名称:rocketmq-commet,代码行数:76,代码来源:AdminBrokerProcessor.java
示例6: getConsumeStats
import com.alibaba.rocketmq.common.admin.OffsetWrapper; //导入依赖的package包/类
public ConsumeStats getConsumeStats(final String addr, final String consumerGroup,
final long timeoutMillis) throws InterruptedException, RemotingTimeoutException,
RemotingSendRequestException, RemotingConnectException, MQBrokerException {
// 添加虚拟运行环境相关的projectGroupPrefix
String consumerGroupWithProjectGroup = consumerGroup;
if (!UtilAll.isBlank(projectGroupPrefix)) {
consumerGroupWithProjectGroup =
VirtualEnvUtil.buildWithProjectGroup(consumerGroup, projectGroupPrefix);
}
GetConsumeStatsRequestHeader requestHeader = new GetConsumeStatsRequestHeader();
requestHeader.setConsumerGroup(consumerGroupWithProjectGroup);
RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.GET_CONSUME_STATS, requestHeader);
RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
ConsumeStats consumeStats = ConsumeStats.decode(response.getBody(), ConsumeStats.class);
// 清除虚拟运行环境相关的projectGroupPrefix
if (!UtilAll.isBlank(projectGroupPrefix)) {
HashMap<MessageQueue, OffsetWrapper> newTopicOffsetMap =
new HashMap<MessageQueue, OffsetWrapper>();
for (Map.Entry<MessageQueue, OffsetWrapper> messageQueue : consumeStats.getOffsetTable()
.entrySet()) {
MessageQueue key = messageQueue.getKey();
key.setTopic(VirtualEnvUtil.clearProjectGroup(key.getTopic(), projectGroupPrefix));
newTopicOffsetMap.put(key, messageQueue.getValue());
}
consumeStats.setOffsetTable(newTopicOffsetMap);
}
return consumeStats;
}
default:
break;
}
throw new MQBrokerException(response.getCode(), response.getRemark());
}
开发者ID:diwayou,项目名称:rocketmq-all-trans,代码行数:42,代码来源:MQClientAPIImpl.java
示例7: resetOffsetByTimestampOld
import com.alibaba.rocketmq.common.admin.OffsetWrapper; //导入依赖的package包/类
@Override
public List<RollbackStats> resetOffsetByTimestampOld(String consumerGroup, String topic, long timestamp,
boolean force) throws RemotingException, MQBrokerException, InterruptedException,
MQClientException {
TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
List<RollbackStats> rollbackStatsList = new ArrayList<RollbackStats>();
for (BrokerData bd : topicRouteData.getBrokerDatas()) {
String addr = bd.selectBrokerAddr();
if (addr != null) {
// 根据 consumerGroup 查找对应的 mq
ConsumeStats consumeStats =
this.mQClientFactory.getMQClientAPIImpl().getConsumeStats(addr, consumerGroup, 3000);
// 根据 topic 过滤不需要的 mq
for (Map.Entry<MessageQueue, OffsetWrapper> entry : consumeStats.getOffsetTable().entrySet()) {
MessageQueue queue = entry.getKey();
OffsetWrapper offsetWrapper = entry.getValue();
if (topic.equals(queue.getTopic())) {
// 根据 timestamp 查找对应的offset
long offset =
this.mQClientFactory.getMQClientAPIImpl().searchOffset(addr, topic,
queue.getQueueId(), timestamp, 3000);
// 构建按时间回溯消费进度
RollbackStats rollbackStats = new RollbackStats();
rollbackStats.setBrokerName(bd.getBrokerName());
rollbackStats.setQueueId(queue.getQueueId());
rollbackStats.setBrokerOffset(offsetWrapper.getBrokerOffset());
rollbackStats.setConsumerOffset(offsetWrapper.getConsumerOffset());
rollbackStats.setTimestampOffset(offset);
rollbackStats.setRollbackOffset(offsetWrapper.getConsumerOffset());
// 更新 offset
if (force || offset <= offsetWrapper.getConsumerOffset()) {
rollbackStats.setRollbackOffset(offset);
UpdateConsumerOffsetRequestHeader requestHeader =
new UpdateConsumerOffsetRequestHeader();
requestHeader.setConsumerGroup(consumerGroup);
requestHeader.setTopic(topic);
requestHeader.setQueueId(queue.getQueueId());
requestHeader.setCommitOffset(offset);
this.mQClientFactory.getMQClientAPIImpl().updateConsumerOffset(addr,
requestHeader, 3000);
}
rollbackStatsList.add(rollbackStats);
}
}
}
}
return rollbackStatsList;
}
开发者ID:diwayou,项目名称:rocketmq-all-trans,代码行数:50,代码来源:DefaultMQAdminExtImpl.java
示例8: getConsumeStats
import com.alibaba.rocketmq.common.admin.OffsetWrapper; //导入依赖的package包/类
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 =
this.brokerController.getConsumerOffsetManager().whichTopicByConsumer(
requestHeader.getConsumerGroup());
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) {
log.warn("consumeStats, the consumer group[{}], topic[{}] not exist",
requestHeader.getConsumerGroup(), topic);
continue;
}
for (int i = 0; i < topicConfig.getWriteQueueNums(); 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().getMaxOffsetInQuque(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);
}
long consumeTps =
this.brokerController.getConsumerOffsetManager().computePullTPS(topic,
requestHeader.getConsumerGroup());
consumeTps += consumeStats.getConsumeTps();
consumeStats.setConsumeTps(consumeTps);
}
byte[] body = consumeStats.encode();
response.setBody(body);
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
return response;
}
开发者ID:diwayou,项目名称:rocketmq-all-trans,代码行数:80,代码来源:AdminBrokerProcessor.java
示例9: getConsumeStats
import com.alibaba.rocketmq.common.admin.OffsetWrapper; //导入依赖的package包/类
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 =
this.brokerController.getConsumerOffsetManager().whichTopicByConsumer(
requestHeader.getConsumerGroup());
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) {
log.warn("consumeStats, the consumer group[{}], topic[{}] not exist",
requestHeader.getConsumerGroup(), topic);
continue;
}
for (int i = 0; i < topicConfig.getWriteQueueNums(); 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().getMaxOffsetInQuque(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);
}
long consumeTps =
this.brokerController.getConsumerOffsetManager().computePullTPS(topic,
requestHeader.getConsumerGroup());
consumeTps += consumeStats.getConsumeTps();
consumeStats.setConsumeTps(consumeTps);
}
byte[] body = consumeStats.encode();
response.setBody(body);
response.setCode(ResponseCode.SUCCESS_VALUE);
response.setRemark(null);
return response;
}
开发者ID:brucechan0921,项目名称:RocketMQ-3.0.8,代码行数:80,代码来源:AdminBrokerProcessor.java
注:本文中的com.alibaba.rocketmq.common.admin.OffsetWrapper类示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论