需求
自从公司采用微服务架构以后,日志不好看了, 虽然已经把所有日志都收集到了elk, 但是因为一个前端的请求通过N个服务,导致日志分布很零乱,没有办法显示这一个请求相关的日志,所以需要开发一点小功能来实现单个请求的跟踪.
设计
基本的想法是, 对每个请求设置一个RequestId, 然后让这个id跟随请求传输到每个服务,进而打印到日志中. 需要解决以下问题:
1.请求id的隔离；
2.将自定义变量打印到日志中, 不能修改日志输出函数；
3.让id跟随请求到达每个服务.
实现
请求id的隔离
每个请求都是单个线程处理的, 那么可以跟据这点来做请求id隔离, 即线程id->请求id, 对于每个新的请求,都生成一个请求id,然后将id保存到Map<线程id,请求id>中.
自定义变量打印到日志
当前用的日志框架是log4j2, 通过查询文档和搜索资料, 发现log4j2的插件能实现这个功能: 具体文档在 Log4j2 Converters
具体实现代码如下:
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.core.LogEvent;
import org.apache.logging.log4j.core.config.plugins.Plugin;
import org.apache.logging.log4j.core.pattern.ConverterKeys;
import org.apache.logging.log4j.core.pattern.LogEventPatternConverter;
import org.apache.logging.log4j.core.pattern.PatternConverter;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
@Plugin(name = "LogRequestIdPlugin", category = PatternConverter.CATEGORY)
@ConverterKeys({"RequestId"})
public class LogRequestIdPlugin extends LogEventPatternConverter {
private static ConcurrentHashMap<Long, String> requestIdMap = new ConcurrentHashMap<>();
private static LogRequestIdPlugin instance = new LogRequestIdPlugin("RequestId", "RequestId");
/**
* Constructs an instance of LoggingEventPatternConverter.
*
* @param name name of converter.
* @param style CSS style for output.
*/
public LogRequestIdPlugin(String name, String style) {
super(name, style);
}
public static LogRequestIdPlugin newInstance(String[] params) {
return instance;
}
/**
* Formats an event into a string buffer.
*
* @param event event to format, may not be null.
* @param toAppendTo string buffer to which the formatted event will be appended. May not be null.
*/
@Override
public void format(LogEvent event, StringBuilder toAppendTo) {
toAppendTo.append(getRequestId());
}
/**
* 设置当前线程的请求id
*
* @param requestId
*/
public static void setRequestId(String requestId) {
if (StringUtils.isEmpty(requestId)) {
requestIdMap.remove(Thread.currentThread().getId());
} else {
requestIdMap.put(Thread.currentThread().getId(), requestId);
}
}
/**
* 获取当前线程请求id
* 如果请求id不存在,则生成一个
*
* @return
*/
public static String getRequestId() {
String requestId = requestIdMap.get(Thread.currentThread().getId());
if (StringUtils.isEmpty(requestId)) {
requestId = newRequestId();
requestIdMap.put(Thread.currentThread().getId(), requestId);
}
return requestId;
}
private final static String charMap = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
private static String randomString(int length) {
Random random = new Random();
StringBuilder stringBuilder = new StringBuilder();
for (int i = 0; i < length; i++) {
stringBuilder.append(charMap.charAt(random.nextInt(charMap.length())));
}
return stringBuilder.toString();
}
/**
* 生成一个请求id
*
* @return
*/
public static String newRequestId() {
String timeHex = String.format("%x", System.currentTimeMillis() - 1513728000000L);
return randomString(6) + timeHex;
}
}
然后修改日志配置文件layout:
<PatternLayout pattern="[%RequestId] %d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
请求id跟随请求到达每个服务
在当前架构中, 客户的一个请求到达网关之后, 在每个服务之间的调用方式有:
controller层的http请求, dubbo请求和rocketmq消息；
在不能影响当前业务的前提条件下, 怎么传输RequestId呢?
- controller层http请求:
在网关将RequestId放到header里面, 在接收端将RequestId取出来放到Map<线程id,请求id>中, 需要加一个web filter, 具体实现如下:
import cn.com.xxx.common.log.LogRequestIdPlugin;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.annotation.Configuration;
import javax.servlet.*;
import javax.servlet.annotation.WebFilter;
import javax.servlet.http.HttpServletRequest;
import java.io.IOException;
@Configuration
@WebFilter(urlPatterns = "/", filterName = "RequestIdLogFilter")
public class WebLogFilter implements Filter {
@Override
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
if (request instanceof HttpServletRequest) {
String requestId = ((HttpServletRequest) request).getHeader("RequestId");
if (StringUtils.isEmpty(requestId)) {
requestId = LogRequestIdPlugin.newRequestId();
}
LogRequestIdPlugin.setRequestId(requestId);
}
chain.doFilter(request, response);
}
@Override
public void init(FilterConfig filterConfig) throws ServletException {
}
@Override
public void destroy() {
}
}
- dubbo请求
dubbo请求可以使用过滤器,在消费端放入RequestId, 在提供端接收RequestId, 具体实现如下:
import cn.com.xxx.common.log.LogRequestIdPlugin;
import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.extension.Activate;
import com.alibaba.dubbo.rpc.*;
import com.alibaba.dubbo.rpc.protocol.dubbo.DecodeableRpcInvocation;
import org.springframework.util.StringUtils;
@Activate(group = {Constants.PROVIDER, Constants.CONSUMER})
public class DubboLogFilter implements Filter {
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
if (invocation instanceof DecodeableRpcInvocation) {
String requestId = invocation.getAttachment("RequestId");
if (!StringUtils.isEmpty(requestId)) {
LogRequestIdPlugin.setRequestId(requestId);
}
} else {
invocation.getAttachments().put("RequestId", LogRequestIdPlugin.getRequestId());
}
return invoker.invoke(invocation);
}
}
3.rocketmq消息
rocketmq可以对每个消费增加自定义变量,用来携带RequestId. 可以自定义公共发送和消费类中处理RequestId, 因为我们当前没有公共发送类 , 只有公共消费类, 所以使用了aop来做处理, 具体实现如下:
import cn.com.xxx.common.log.LogRequestIdPlugin;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.aspectj.lang.annotation.Pointcut;
import org.springframework.stereotype.Component;
@Aspect
@Component
public class RocketMQLogFilter {
@Pointcut("execution(public * org.apache.rocketmq.client.producer.DefaultMQProducer.send(" +
"org.apache.rocketmq.common.message.Message,..)) " +
"&& args(msg,..)")
private void sendMessage(Message msg) {
}
@Before(value = "sendMessage(msg)", argNames = "msg")
public void onSendMessage(Message msg) {
msg.putUserProperty("RequestId", LogRequestIdPlugin.getRequestId());
}
/**
* 消费时调用, 放到公共库中处理
*
* @param msg
*/
public static void onConsumeMessage(MessageExt msg) {
String requestId = msg.getUserProperty("RequestId");
if (!StringUtils.isEmpty(requestId)) {
LogRequestIdPlugin.setRequestId(requestId);
}
}
}
总结
一个简单的基于请求的日志跟踪功能就这样实现了, 虽然很简单, 但是对于查找问题还是非常有用的.
异步处理
日志处理异步输出模式时, 可以参考<<微服务基于请求的日志跟踪2>>
