Skip to content

R
Rocketmq

Apache RocketMQ / Rocketmq
上一次同步 2 年多

通知 240
Star 16136
Fork 68
R
Rocketmq

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

未验证 提交 a83cdc5f 编写于 作者: G Git_Yang 提交者： GitHub
浏览文件
操作

[format] Fix formatting errors (#3380) 

Signed-off-by: Nzhangyang <Git_Yang@163.com>
上级 4dbdbf0a
隐藏空白更改
内联 并排
Showing with 94 addition and 93 deletion
acl/src/main/java/org/apache/rocketmq/acl/common/AclUtils.java
浏览文件 @ a83cdc5f
...@@ -23,7 +23,6 @@ import java.io.FileNotFoundException; ...@@ -23,7 +23,6 @@ import java.io.FileNotFoundException;
import java.io.FileWriter; import java.io.FileWriter;
import java.io.IOException; import java.io.IOException;
import java.io.PrintWriter; import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.Map; import java.util.Map;
import java.util.SortedMap; import java.util.SortedMap;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
...@@ -210,9 +209,9 @@ public class AclUtils { ...@@ -210,9 +209,9 @@ public class AclUtils {
// expand netaddress // expand netaddress
int separatorCount = StringUtils.countMatches(netaddress, ":"); int separatorCount = StringUtils.countMatches(netaddress, ":");
int padCount = part - separatorCount; int padCount = part - separatorCount;
if(padCount > 0){ if (padCount > 0) {
StringBuilder padStr = new StringBuilder(":"); StringBuilder padStr = new StringBuilder(":");
for(int i = 0; i < padCount; i++){ for (int i = 0; i < padCount; i++) {
padStr.append(":"); padStr.append(":");
} }
netaddress = StringUtils.replace(netaddress, "::", padStr.toString()); netaddress = StringUtils.replace(netaddress, "::", padStr.toString());
...@@ -221,7 +220,7 @@ public class AclUtils { ...@@ -221,7 +220,7 @@ public class AclUtils {
// pad netaddress // pad netaddress
String[] strArray = StringUtils.splitPreserveAllTokens(netaddress, ":"); String[] strArray = StringUtils.splitPreserveAllTokens(netaddress, ":");
for (int i = 0; i < strArray.length; i++) { for (int i = 0; i < strArray.length; i++) {
if(strArray[i].length() < 4){ if (strArray[i].length() < 4) {
strArray[i] = StringUtils.leftPad(strArray[i], 4, '0'); strArray[i] = StringUtils.leftPad(strArray[i], 4, '0');
} }
} }
......
example/src/main/java/org/apache/rocketmq/example/simple/PullConsumer.java
浏览文件 @ a83cdc5f
...@@ -36,7 +36,7 @@ import org.apache.rocketmq.remoting.exception.RemotingException; ...@@ -36,7 +36,7 @@ import org.apache.rocketmq.remoting.exception.RemotingException;
public class PullConsumer { public class PullConsumer {
public static void main(String[] args) throws MQClientException { public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5"); DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
consumer.setNamesrvAddr("127.0.0.1:9876"); consumer.setNamesrvAddr("127.0.0.1:9876");
Set<String> topics = new HashSet<>(); Set<String> topics = new HashSet<>();
...@@ -46,99 +46,101 @@ public class PullConsumer { ...@@ -46,99 +46,101 @@ public class PullConsumer {
consumer.start(); consumer.start();
ExecutorService executors = Executors.newFixedThreadPool(topics.size(), new ThreadFactory() { ExecutorService executors = Executors.newFixedThreadPool(topics.size(), new ThreadFactory() {
@Override @Override
public Thread newThread(Runnable r) { public Thread newThread(Runnable r) {
return new Thread(r, "PullConsumerThread"); return new Thread(r, "PullConsumerThread");
} }
}); });
for(String topic : consumer.getRegisterTopics()){ for (String topic : consumer.getRegisterTopics()) {
executors.execute(new Runnable() { executors.execute(new Runnable() {
public void doSomething(List<MessageExt> msgs){ public void doSomething(List<MessageExt> msgs) {
//do you business //do you business
System.out.println(msgs);
} }
@Override
public void run() { @Override
while(true){ public void run() {
try { while (true) {
Set<MessageQueue> messageQueues = consumer.fetchMessageQueuesInBalance(topic); try {
if(messageQueues == null || messageQueues.isEmpty()){ Set<MessageQueue> messageQueues = consumer.fetchMessageQueuesInBalance(topic);
Thread.sleep(1000); if (messageQueues == null || messageQueues.isEmpty()) {
continue; Thread.sleep(1000);
} continue;
PullResult pullResult = null; }
for(MessageQueue messageQueue : messageQueues){ PullResult pullResult = null;
try { for (MessageQueue messageQueue : messageQueues) {
long offset = this.consumeFromOffset(messageQueue); try {
pullResult = consumer.pull(messageQueue, "*", offset, 32); long offset = this.consumeFromOffset(messageQueue);
switch (pullResult.getPullStatus()) { pullResult = consumer.pull(messageQueue, "*", offset, 32);
case FOUND: switch (pullResult.getPullStatus()) {
List<MessageExt> msgs = pullResult.getMsgFoundList(); case FOUND:
List<MessageExt> msgs = pullResult.getMsgFoundList();
if(msgs != null && !msgs.isEmpty()){
this.doSomething(msgs); if (msgs != null && !msgs.isEmpty()) {
//update offset to broker this.doSomething(msgs);
consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset()); //update offset to broker
//print pull tps consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset());
this.incPullTPS(topic, pullResult.getMsgFoundList().size()); //print pull tps
} this.incPullTPS(topic, pullResult.getMsgFoundList().size());
break; }
case OFFSET_ILLEGAL: break;
consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset()); case OFFSET_ILLEGAL:
break; consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset());
case NO_NEW_MSG: break;
Thread.sleep(1); case NO_NEW_MSG:
consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset()); Thread.sleep(1);
break; consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset());
case NO_MATCHED_MSG: break;
consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset()); case NO_MATCHED_MSG:
break; consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset());
default: break;
} default:
} catch (RemotingException e) { }
e.printStackTrace(); } catch (RemotingException e) {
} catch (MQBrokerException e) { e.printStackTrace();
e.printStackTrace(); } catch (MQBrokerException e) {
} catch (Exception e){ e.printStackTrace();
e.printStackTrace(); } catch (Exception e) {
} e.printStackTrace();
} }
} catch (MQClientException e) { }
//reblance error } catch (MQClientException e) {
e.printStackTrace(); //reblance error
} catch (InterruptedException e) { e.printStackTrace();
e.printStackTrace(); } catch (InterruptedException e) {
} catch (Exception e){ e.printStackTrace();
e.printStackTrace(); } catch (Exception e) {
} e.printStackTrace();
} }
} }
}
public long consumeFromOffset(MessageQueue messageQueue) throws MQClientException{
//-1 when started public long consumeFromOffset(MessageQueue messageQueue) throws MQClientException {
long offset = consumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY); //-1 when started
if(offset < 0){ long offset = consumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY);
//query from broker if (offset < 0) {
offset = consumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_STORE); //query from broker
} offset = consumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_STORE);
if (offset < 0){ }
//first time start from last offset if (offset < 0) {
offset = consumer.maxOffset(messageQueue); //first time start from last offset
offset = consumer.maxOffset(messageQueue);
} }
//make sure //make sure
if (offset < 0){ if (offset < 0) {
offset = 0; offset = 0;
} }
return offset; return offset;
} }
public void incPullTPS(String topic, int pullSize) {
consumer.getDefaultMQPullConsumerImpl().getRebalanceImpl().getmQClientFactory() public void incPullTPS(String topic, int pullSize) {
.getConsumerStatsManager().incPullTPS(consumer.getConsumerGroup(), topic, pullSize); consumer.getDefaultMQPullConsumerImpl().getRebalanceImpl().getmQClientFactory()
} .getConsumerStatsManager().incPullTPS(consumer.getConsumerGroup(), topic, pullSize);
}); }
});
} }
// executors.shutdown(); // executors.shutdown();
// consumer.shutdown(); // consumer.shutdown();
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或