更新日志配置

src/main/java/com/tidecloud/dataacceptance/config/LogBackExtDiscriminator.java

@@ -0,0 +1,118 @@
+//package com.tidecloud.dataacceptance.config;
+//
+//import java.util.concurrent.Callable;
+//import java.util.concurrent.ExecutionException;
+//
+//import org.springframework.beans.BeansException;
+//import org.springframework.context.ApplicationContext;
+//import org.springframework.context.ApplicationContextAware;
+//import org.springframework.stereotype.Service;
+//
+//import com.google.common.cache.Cache;
+//import com.google.common.cache.CacheBuilder;
+//import com.tidecloud.dataacceptance.service.AcceptanceInboundHandlerAdapter;
+//
+//import ch.qos.logback.classic.spi.ILoggingEvent;
+//import ch.qos.logback.core.sift.AbstractDiscriminator;
+//
+//@Service
+//public class LogBackExtDiscriminator extends AbstractDiscriminator<ILoggingEvent> implements ApplicationContextAware {
+//	private static final String KEY = "strategyName";
+//	private String defaultValue;
+//	private static ApplicationContext context;
+//
+//	static Cache<String, loggerInfo> loggerNameCache = CacheBuilder.newBuilder()
+//			// 设置cache的初始大小为10，要合理设置该值
+//			.initialCapacity(10)
+//			// 设置并发数为5，即同一时间最多只能有5个线程往cache执行写入操作
+//			.concurrencyLevel(10)
+//			// 构建cache实例
+//			.build();
+//	static Cache<String, String> loggerThreadCache = CacheBuilder.newBuilder()
+//			// 设置cache的初始大小为10，要合理设置该值
+//			.initialCapacity(10)
+//			// 设置并发数为5，即同一时间最多只能有5个线程往cache执行写入操作
+//			.concurrencyLevel(10)
+//			// 构建cache实例
+//			.build();
+//
+//	class loggerInfo{
+//		String loggerName ;
+//		String loggerThread ;
+//		
+//		public String getLoggerName() {
+//			return loggerName;
+//		}
+//		public void setLoggerName(String loggerName) {
+//			this.loggerName = loggerName;
+//		}
+//		public String getLoggerThread() {
+//			return loggerThread;
+//		}
+//		public void setLoggerThread(String loggerThread) {
+//			this.loggerThread = loggerThread;
+//		}
+//		
+//	}
+//	@Override
+//	public String getDiscriminatingValue(ILoggingEvent event) {
+//		String loggerName = event.getLoggerName();
+//		String loggerThread = event.getThreadName();
+//		loggerInfo info = new loggerInfo();
+//		info.setLoggerName(loggerName);
+//		info.setLoggerThread(loggerThread);
+//		String acceptClass = AcceptanceInboundHandlerAdapter.class.getPackage().getName();
+//
+//		boolean isTrue = loggerName.startsWith(acceptClass);
+//		if (!isTrue) {
+//			return defaultValue;
+//		}
+//
+//		String cacheloggerName = loggerThreadCache.getIfPresent(loggerThread);
+//		 loggerNameCache.getIfPresent(loggerName);
+//		
+//		if (cacheloggerName == null) {
+//
+//			try {
+//				Class handlerClass = Class.forName(loggerName);
+//				AcceptanceInboundHandlerAdapter handler = (AcceptanceInboundHandlerAdapter) context
+//						.getBean(handlerClass);
+//
+//				String name = handler.getPrefixName();
+//				loggerThreadCache.put(loggerThread, name);
+//				return name;
+//			} catch (Exception e) {
+//			
+//			}
+//
+//			int dotIndex = loggerName.lastIndexOf('.');
+//			if (dotIndex != -1 && dotIndex < loggerName.length() - 1) {
+//				loggerName= loggerName.substring(dotIndex + 1);
+//			}
+//			
+//			
+//		}
+//		loggerNameCache.put(loggerName, info);
+//		return cacheloggerName;
+//	}
+//
+//	@Override
+//	public String getKey() {
+//		return KEY;
+//	}
+//
+//	public String getDefaultValue() {
+//		return defaultValue;
+//	}
+//
+//	public void setDefaultValue(String defaultValue) {
+//		this.defaultValue = defaultValue;
+//	}
+//
+//	@Override
+//	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+//		context = applicationContext;
+//
+//	}
+//
+//}

src/main/java/com/tidecloud/dataacceptance/service/AcceptanceDeviceService.java

@@ -77,7 +77,7 @@ public class AcceptanceDeviceService implements ApplicationContextAware {
 			for (int i = 0; i < workDeviceList.size(); i++) {
 				DeviceConfig config = workDeviceList.get(i);
 				singleThreadPool.execute(() -> {
-					MDC.put("strategyName", config.getName());
+					
 					try {
 						Class handlerClass = Class.forName(config.getHandlerClass());
 						AcceptanceInboundHandlerAdapter handler = (AcceptanceInboundHandlerAdapter) context

src/main/java/com/tidecloud/dataacceptance/service/AcceptanceInboundHandlerAdapter.java

@@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang.ArrayUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
@@ -94,9 +95,13 @@ public  class AcceptanceInboundHandlerAdapter extends ChannelInboundHandlerAdapt
 
 	private String topic;
 
-	
+	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception{
+		
+		MDC.put("strategyName", this.getPrefixName());
+	}
 
 	protected void receiveMsg(String msg, String deviceId, Channel channel) {
+		
 		manageChannel(channel, deviceId);
 		kafkaSendThreadPool.execute(() -> sendKakfaMsg(msg, deviceId));
 		//singleThreadPool.execute(() -> dataStorage(msg));
@@ -130,12 +135,14 @@ public  class AcceptanceInboundHandlerAdapter extends ChannelInboundHandlerAdapt
 
 	protected void sendKakfaMsg(String msg, String deviceId) {
 		Integer key = new Integer(deviceId.hashCode());
+		MDC.put("strategyName", this.getPrefixName());
 		ListenableFuture<SendResult<Integer, String>> listenableFuture = kafkaTemplate.send(topic, key, msg);
 
 		// 发送成功回调
 		SuccessCallback<SendResult<Integer, String>> successCallback = new SuccessCallback<SendResult<Integer, String>>() {
 			@Override
 			public void onSuccess(SendResult<Integer, String> result) {
+				MDC.put("strategyName", AcceptanceInboundHandlerAdapter.this.getPrefixName());
 				// 成功业务逻辑
 				logger.info("发送kafka成功.deviceId:{}, msg:{}", key, msg);
 			}
@@ -144,6 +151,7 @@ public  class AcceptanceInboundHandlerAdapter extends ChannelInboundHandlerAdapt
 		FailureCallback failureCallback = new FailureCallback() {
 			@Override
 			public void onFailure(Throwable ex) {
+				MDC.put("strategyName", AcceptanceInboundHandlerAdapter.this.getPrefixName());
 				// 失败业务逻辑
 				logger.error("发送kafka失败.deviceId:{}, msg:{}", key, msg);
 				singleThreadPool.execute(() -> dataStorage(msg));

src/main/java/com/tidecloud/dataacceptance/service/HexBinaryAcceptanceHandlerAdapter.java

@@ -13,6 +13,7 @@ public abstract class HexBinaryAcceptanceHandlerAdapter extends AcceptanceInboun
 	private static final Logger logger = LoggerFactory.getLogger(HexBinaryAcceptanceHandlerAdapter.class);
 	@Override
 	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+		super.channelRead(ctx, msg);
 		ByteBuf in = (ByteBuf) msg;
 		printAcceptanceData(in,ctx);
 		

src/main/java/com/tidecloud/dataacceptance/service/StringAcceptanceHandlerAdapter.java

@@ -12,6 +12,7 @@ public abstract class StringAcceptanceHandlerAdapter extends AcceptanceInboundHa
 	
 	@Override
 	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+		super.channelRead(ctx, msg);
 		ByteBuf byteBuf = (ByteBuf) msg;
 		String str = byteBufferToString(byteBuf.nioBuffer());
 		logger.info("接入数据：{}", str);