电子技术论坛_中国专业的电子工程师学习交流社区-中电网技术论坛

标题: 分布式集群环境下调用链路追踪（6）创建 demoe-message 工程 [打印本页]

作者: look_w 时间: 2018-11-29 18:05 标题: 分布式集群环境下调用链路追踪（6）创建 demoe-message 工程

创建 demoe-message 工程

新建一个基于 Spring Boot Web 的 Gradle 项目，项目名称为：demo-message。

项目 build.gradle 配置如下：清单 13. demo-message Gradle 脚本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

repositories {
}

dependencies {
compile project(':demo-model')
compile project(':demo-util')

compile('org.springframework.boot:spring-boot-starter-aop')
compile('org.springframework.cloud:spring-cloud-starter-sleuth')
compile('org.springframework.cloud:spring-cloud-sleuth-stream')
compile('org.springframework.cloud:spring-cloud-starter-stream-kafka')
compile('org.springframework.boot:spring-boot-starter-data-redis')
compile('org.springframework.kafka:spring-kafka')
compile('org.springframework.boot:spring-boot-starter-web')
compile('org.springframework.cloud:spring-cloud-sleuth-zipkin')
}

demoe-message Redis 消息解包处理，（如清单 14 所示），基于 Aspect 实现，以此来实现对接受到的消息进行解包处理，同时保证代码透明注入，请在项目中查看完整源代码。清单 14. demo-message Message 封包处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

@Pointcut("execution(* onMessage(..)) & &
target(org.springframework.data.redis.connection.MessageListener)")
public void messageListener() {
      // This func is intentionally empty. Nothing special is needed here.
}

@Around("messageListener()")
public void interceptMessage(ProceedingJoinPoint joinPoint) throws Throwable {
      try {
         Object[] joinPointArgs = joinPoint.getArgs();
         Object[] result = new Object[joinPointArgs.length];
         for (int i = 0; i < joinPointArgs.length; i++) {
            Object object = joinPointArgs;
            ObjectMapper objectMapper = new ObjectMapper();
            if (object instanceof Message && jsonParse(object)) {
                  Message message = (Message) object;
                  RedisMessage redisMessage = objectMapper.readValue((String)
object.toString(), RedisMessage.class);
                  logger.debug("Received >> redisMessage {} ", redisMessage);
                  Map<String, String> spanMap = redisMessage.getHeader();
                  String traceIdString =
Span.idToHex(Long.parseLong(spanMap.get(TraceMessageHeaders.TRACE_ID_NAME)));
                  String spanId =
Span.idToHex(Long.parseLong(spanMap.get(TraceMessageHeaders.SPAN_ID_NAME)));
                  Span parentSpan = SleuthHelper.fromMap(spanMap);
                  logger.debug("Received >> trace {} span {}", traceIdString,
spanId);
                  if (sleuthProperties != null &&
sleuthProperties.isSupportsJoin()) {
                     SleuthHelper.joinSpan(tracer, spanMap);
                     logger.debug("Join trace span {}", spanMap);
                  } else {
                     SleuthHelper.continueSpan(tracer, spanMap);
                     logger.debug("Continue trace span {}", spanMap);
                  }
                  slf4jSpanLogger.logStartedSpan(parentSpan,
tracer.getCurrentSpan());
                  logger.debug("Received >> trace {} span {}",
Span.idToHex(tracer.getCurrentSpan().getTraceId()),
                        Span.idToHex(tracer.getCurrentSpan().getSpanId()));
                  byte[] channel = message.getChannel();
                  byte[] body =
SerializationUtils.serialize(redisMessage.getMessage());
                  logger.debug("Received >> message {}",
SerializationUtils.deserialize(body));
                  result = new RedisDefaultMessage(channel, body);
            } else {
                  result = object;
            }
         }
         SpanBuilder spanBuilder =
Span.builder().from(this.tracer.getCurrentSpan());
         Span currentSpan =
spanBuilder.parents(this.tracer.getCurrentSpan().getParents()).name(joinPoint.getSi
gnature().getName()).remote(false).build();
         currentSpan.logEvent("sr");
         this.tracer.close(currentSpan);
         this.tracer.continueSpan(currentSpan);
         joinPoint.proceed(result);
      } catch (JsonParseException e) {
         logger.error("Join trace span with JsonParseException {} ", e);
      } catch (JsonMappingException e) {
         logger.error("Join trace span with JsonMappingException {} ", e);
      } catch (IOException e) {
         logger.error("Join trace span with IOException {}", e);
      }
}

private boolean jsonParse(Object object) {
      boolean falg = true;
      ObjectMapper objectMapper = new ObjectMapper();
      try {
         objectMapper.readValue((String) object.toString(), RedisMessage.class);
         falg = true;
      } catch (Exception e) {
         falg = false;
      }
      return falg;
}

欢迎光临电子技术论坛_中国专业的电子工程师学习交流社区-中电网技术论坛 (http://bbs.eccn.com/)