tiennm99/zfoo
1
0 Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-10-14 20:17:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf[runnable]: 使用Runnable函数式编程接口

Browse Source
This commit is contained in:
godotg committed 2022-07-28 21:47:33 +08:00
1 parent a637db260a
commit c4cdf272eb
42 files changed
+105 -142
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

net/src/main/java/com/zfoo/net/consumer/registry/ZookeeperRegistry.java
+2 -13
View File
@@ -29,7 +29,6 @@ import com.zfoo.protocol.util.IOUtils;
import com.zfoo.protocol.util.JsonUtils;
import com.zfoo.protocol.util.StringUtils;
import com.zfoo.scheduler.manager.SchedulerBus;
import com.zfoo.util.SafeRunnable;
import com.zfoo.util.ThreadUtils;
import com.zfoo.util.net.HostAndPort;
import io.netty.util.concurrent.FastThreadLocalThread;
@@ -361,12 +360,7 @@ public class ZookeeperRegistry implements IRegistry {
} catch (Exception e) {
//
logger.error("zookeeper初始化失败，等待[{}]秒，重新初始化", RETRY_SECONDS, e);
SchedulerBus.schedule(new SafeRunnable() {
@Override
public void doRun() {
initZookeeper();
}
}, RETRY_SECONDS, TimeUnit.SECONDS);
SchedulerBus.schedule(() -> initZookeeper(), RETRY_SECONDS, TimeUnit.SECONDS);
}
});
}
@@ -522,12 +516,7 @@ public class ZookeeperRegistry implements IRegistry {
}
if (recheckFlag) {
SchedulerBus.schedule(new SafeRunnable() {
@Override
public void doRun() {
checkConsumer();
}
}, RETRY_SECONDS, TimeUnit.SECONDS);
SchedulerBus.schedule(() -> checkConsumer(), RETRY_SECONDS, TimeUnit.SECONDS);
}
}
net/src/main/java/com/zfoo/net/task/TaskBus.java
+5 -5
View File
@@ -73,9 +73,9 @@ public final class TaskBus {
}
public static class TaskThreadFactory implements ThreadFactory {
private int poolNumber;
private AtomicInteger threadNumber = new AtomicInteger(1);
private ThreadGroup group;
private final int poolNumber;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final ThreadGroup group;
public TaskThreadFactory(int poolNumber) {
var s = System.getSecurityManager();
@@ -124,8 +124,8 @@ public final class TaskBus {
return Math.abs(executorConsistentHash % EXECUTOR_SIZE);
}
public static void execute(int executorConsistentHash, SafeRunnable runnable) {
executors[executorIndex(executorConsistentHash)].execute(runnable);
public static void execute(int executorConsistentHash, Runnable runnable) {
executors[executorIndex(executorConsistentHash)].execute(SafeRunnable.valueOf(runnable));
}
// 在task，event，scheduler线程执行的异步请求，请求成功过后依然在相同的线程执行回调任务
net/src/main/java/com/zfoo/net/util/SimpleCache.java
+18 -25
View File
@@ -20,7 +20,6 @@ import com.zfoo.event.manager.EventBus;
import com.zfoo.protocol.collection.CollectionUtils;
import com.zfoo.protocol.model.Pair;
import com.zfoo.scheduler.manager.SchedulerBus;
import com.zfoo.util.SafeRunnable;
import org.checkerframework.checker.nullness.qual.NonNull;
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -94,31 +93,25 @@ public class SimpleCache<K, V> {
});
SchedulerBus.scheduleAtFixedRate(new SafeRunnable() {
@Override
public void doRun() {
// 不在任务调度线程中执行耗时任务，因为任务调度线程只有一个线程池
EventBus.asyncExecute(new SafeRunnable() {
@Override
public void doRun() {
var list = new ArrayList<K>();
while (!linkedQueue.isEmpty()) {
var key = linkedQueue.poll();
list.add(key);
if (list.size() >= BATCH_RELOAD_SIZE) {
var result = batchLoadCallback.apply(list);
result.forEach(it -> cache.put(it.getKey(), it.getValue()));
list.clear();
}
}
if (CollectionUtils.isNotEmpty(list)) {
var result = batchLoadCallback.apply(list);
result.forEach(it -> cache.put(it.getKey(), it.getValue()));
}
SchedulerBus.scheduleAtFixedRate(() -> {
// 不在任务调度线程中执行耗时任务，因为任务调度线程只有一个线程池
EventBus.asyncExecute(() -> {
var list = new ArrayList<K>();
while (!linkedQueue.isEmpty()) {
var key = linkedQueue.poll();
list.add(key);
if (list.size() >= BATCH_RELOAD_SIZE) {
var result = batchLoadCallback.apply(list);
result.forEach(it -> cache.put(it.getKey(), it.getValue()));
list.clear();
}
});
}
}
if (CollectionUtils.isNotEmpty(list)) {
var result = batchLoadCallback.apply(list);
result.forEach(it -> cache.put(it.getKey(), it.getValue()));
}
});
}, refreshDuration, TimeUnit.MILLISECONDS);
net/src/main/java/com/zfoo/net/util/SingleCache.java
+1 -7
View File
@@ -15,7 +15,6 @@ package com.zfoo.net.util;
import com.zfoo.event.manager.EventBus;
import com.zfoo.scheduler.util.TimeUtils;
import com.zfoo.util.SafeRunnable;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -62,12 +61,7 @@ public class SingleCache<V> {
try {
if (now > refreshTime) {
refreshTime = now + refreshDuration;
EventBus.asyncExecute(new SafeRunnable() {
@Override
public void doRun() {
cache = supplier.get();
}
});
EventBus.asyncExecute(() -> cache = supplier.get());
}
} finally {
lock.unlock();
net/src/test/java/com/zfoo/net/router/SignalBridgeTest.java
+4 -5
View File
@@ -15,7 +15,6 @@ package com.zfoo.net.router;
import com.zfoo.event.manager.EventBus;
import com.zfoo.net.router.route.SignalBridge;
import com.zfoo.scheduler.util.TimeUtils;
import com.zfoo.util.SafeRunnable;
import com.zfoo.util.ThreadUtils;
import org.junit.Ignore;
import org.junit.Test;
@@ -55,9 +54,9 @@ public class SignalBridgeTest {
var countDownLatch = new CountDownLatch(executorSize);
for (var i = 0; i < executorSize; i++) {
EventBus.execute(i, new SafeRunnable() {
EventBus.execute(i, new Runnable() {
@Override
public void doRun() {
public void run() {
addAndRemoveArray();
countDownLatch.countDown();
}
@@ -73,9 +72,9 @@ public class SignalBridgeTest {
var countDownLatch = new CountDownLatch(executorSize);
for (int i = 0; i < executorSize; i++) {
EventBus.execute(i, new SafeRunnable() {
EventBus.execute(i, new Runnable() {
@Override
public void doRun() {
public void run() {
addAndRemoveMap();
countDownLatch.countDown();
}