lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化traceid

Browse Source
This commit is contained in:
redkale
2023-10-20 15:40:16 +08:00
parent 482a3313c4
commit 09596e9b4b
5 changed files with 26 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/org/redkale/mq/MessageServlet.java
View File

@@ -72,6 +72,8 @@ public abstract class MessageServlet implements MessageProcessor {
onError(response, message, ex);
}
logger.log(Level.SEVERE, getClass().getSimpleName() + " process error, message=" + message, ex instanceof CompletionException ? ((CompletionException) ex).getCause() : ex);
} finally {
Traces.removeTraceid();
}
}

1
src/main/java/org/redkale/net/Context.java
View File

@@ -123,6 +123,7 @@ public class Context {
response.context.logger.log(Level.WARNING, "Execute servlet occur exception. request = " + request, t);
response.finishError(t);
}
Traces.removeTraceid();
});
} else {
try {

12
src/main/java/org/redkale/net/client/ClientCodec.java
View File

@@ -141,21 +141,25 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
readThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.complete(rs);
Traces.removeTraceid();
});
} else if (workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE
if (workThread.inIO()) {
Traces.computeIfAbsent(request.traceid);
respFuture.complete(rs);
Traces.removeTraceid();
} else {
workThread.execute(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.complete(rs);
Traces.removeTraceid();
});
}
} else {
workThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.complete(rs);
Traces.removeTraceid();
});
}
} else { //异常
@@ -163,21 +167,25 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
readThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
Traces.removeTraceid();
});
} else if (workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE
if (workThread.inIO()) {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
Traces.removeTraceid();
} else {
workThread.execute(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
Traces.removeTraceid();
});
}
} else {
workThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
Traces.removeTraceid();
});
}
}
@@ -186,21 +194,25 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
readThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(t);
Traces.removeTraceid();
});
} else if (workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE
if (workThread.inIO()) {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(t);
Traces.removeTraceid();
} else {
workThread.execute(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(t);
Traces.removeTraceid();
});
}
} else {
workThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(t);
Traces.removeTraceid();
});
}
connection.client.logger.log(Level.INFO, "Complete result error, request: " + respFuture.request, t);

1
src/main/java/org/redkale/net/http/HttpServlet.java
View File

@@ -89,6 +89,7 @@ public class HttpServlet extends Servlet<HttpContext, HttpRequest, HttpResponse>
response.getContext().getLogger().log(Level.WARNING, "Servlet occur exception. request = " + request, t);
response.finishError(t);
}
Traces.removeTraceid();
});
}
} else {

54
src/main/java/org/redkale/util/Traces.java
View File

@@ -3,6 +3,7 @@
package org.redkale.util;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
/**
@@ -20,15 +21,12 @@ public class Traces {
private static final String PROCESS_ID = UUID.randomUUID().toString().replaceAll("-", "");
private static final ThreadLocal<IdSequence> THREAD_SEQUENCE = ThreadLocal.withInitial(IdSequence::new);
private static final AtomicLong sequence = new AtomicLong(System.currentTimeMillis());
private static final Supplier<String> tidSupplier = () -> PROCESS_ID + sequence.incrementAndGet();
private static final ThreadLocal<String> localTrace = new ThreadLocal<>();
//借用Skywalking的GlobalIdGenerator生成ID的规则
private static final Supplier<String> tidSupplier = () -> PROCESS_ID
+ "." + Thread.currentThread().getId()
+ "." + THREAD_SEQUENCE.get().nextSeq();
public static boolean enable() {
return enable;
}
@@ -41,6 +39,12 @@ public class Traces {
return enable ? localTrace.get() : null;
}
public static void removeTraceid() {
if (enable) {
localTrace.remove();
}
}
public static String computeIfAbsent(String requestTraceid) {
if (enable) {
String rs = requestTraceid;
@@ -52,42 +56,4 @@ public class Traces {
}
return requestTraceid;
}
private static class IdSequence {
private long lastTimestamp;
private short threadSeq;
private long lastShiftTimestamp;
private int lastShiftValue;
public IdSequence() {
this.lastTimestamp = System.currentTimeMillis();
}
public long nextSeq() {
long rs = timestamp() * 10000;
if (threadSeq == 10000) {
threadSeq = 0;
}
return rs + threadSeq++;
}
private long timestamp() {
long currentTimeMillis = System.currentTimeMillis();
if (currentTimeMillis < lastTimestamp) {
// Just for considering time-shift-back by Ops or OS. @hanahmily 's suggestion.
if (lastShiftTimestamp != currentTimeMillis) {
lastShiftValue++;
lastShiftTimestamp = currentTimeMillis;
}
return lastShiftValue;
} else {
lastTimestamp = currentTimeMillis;
return lastTimestamp;
}
}
}
}