add deal for fail msg

Author: zoudaokoulife

src/main/java/com/dtstack/logstash/assembly/FilterAndOutputThread.java

@@ -26,10 +26,6 @@ public class FilterAndOutputThread implements Runnable {
 	private List<BaseFilter> filterProcessors;
 	private List<BaseOutput> outputProcessors;
 	private int batchSize;
-	private List<Map<String, Object>> batchEvent = Lists
-			.newCopyOnWriteArrayList();
-
-	private Map<Long, Integer> repeatOffest = Maps.newLinkedHashMap();
 
 	public FilterAndOutputThread(
 			LinkedBlockingQueue<Map<String, Object>> inputQueue,
@@ -46,6 +42,19 @@ public void run() {
 		A: while (true) {
 			Map<String, Object> event = null;
 			try {
+				
+				//优先处理失败信息
+				boolean dealFailMsg = false;
+				for (BaseOutput bo : outputProcessors) {
+					if (bo.isConsistency()) {
+						dealFailMsg = dealFailMsg || bo.dealFailedMsg();
+					}
+				}
+				
+				if(dealFailMsg){
+					continue A;
+				}
+				
 				event = inputQueue.take();
 				if (this.filterProcessors != null) {
 					for (BaseFilter bf : filterProcessors) {
@@ -55,12 +64,7 @@ public void run() {
 				}
 				if (event != null && event.size() > 0) {
 					for (BaseOutput bo : outputProcessors) {
-						if (!bo.isConsistency()) {
-							bo.process(event);
-						} else {
-							repeatEvent(bo, event, true, 0);
-							repeatOffest.clear();
-						}
+						bo.process(event);
 					}
 				}
 			} catch (Exception e) {
@@ -70,37 +74,4 @@ public void run() {
 		}
 	}
 
-	public void repeatEvent(BaseOutput bo, Map<String, Object> event,
-			boolean source, int index) {
-		if (bo.getAto().get() == 0) {
-			bo.process(event);
-			if (source) {
-				batchEvent.add(event);
-			}
-		} else if (bo.getAto().get() == 1) {
-			bo.getAto().getAndSet(0);
-			bo.process(event);
-			if (source) {
-				batchEvent.clear();
-				batchEvent.add(event);
-			} else {
-				long size = repeatOffest.size();
-				repeatOffest.put(size, index);
-			}
-		} else if (bo.getAto().get() == 2) {
-			bo.getAto().getAndSet(0);
-			long nested = repeatOffest.size() + 1;
-			int lastTimeIndex = nested == 1 ? 0 : repeatOffest.get(nested - 1);
-			repeatOffest.put(nested, lastTimeIndex);
-			int eventSize = batchEvent.size();
-			for (int i = lastTimeIndex; i < eventSize; i++) {
-				if (nested < repeatOffest.size()) {
-					break;
-				}
-				repeatEvent(bo, batchEvent.get(i), false, i);
-			}
-			logger.warn(bo.getClass().getName() + "--->repeat event size:"
-					+ batchEvent.size());
-		}
-	}
 }

src/main/java/com/dtstack/logstash/log/LogbackComponent.java

@@ -19,7 +19,7 @@
  */
 public class LogbackComponent extends LogComponent{
 
-	private static String formatePattern ="%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n";
+	private static String formatePattern ="%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} [%file:%line] - %msg%n";
 
 	private static int day = 7;
 

src/main/java/com/dtstack/logstash/outputs/BaseOutput.java

@@ -4,11 +4,16 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 import com.dtstack.logstash.render.FreeMarkerRender;
 import com.dtstack.logstash.render.TemplateRender;
+import com.google.common.collect.Queues;
 
 /**
  * 
@@ -32,7 +37,9 @@ public abstract class BaseOutput implements Cloneable, java.io.Serializable{
 	protected AtomicInteger ato = new AtomicInteger(0);
 
 	//数据强一致性是否开启
-	protected static boolean consistency =false;
+	protected boolean consistency =false;
+	
+	public BlockingQueue<Object> failedMsgQueue = Queues.newLinkedBlockingDeque();
 
 	public BaseOutput(Map config) {
 		this.config = config;
@@ -50,6 +57,10 @@ public BaseOutput(Map config) {
 		} else {
 			IF = null;
 		}
+		
+		if(this.config.containsKey("consistency")){
+			consistency = (boolean) this.config.get("consistency");
+		}
 	}
 
 	public abstract void prepare();
@@ -78,10 +89,36 @@ public AtomicInteger getAto() {
 		return ato;
 	}
 
-
-	public static boolean isConsistency() {
+	public boolean isConsistency() {
 		return consistency;
 	}
+	
+	public boolean dealFailedMsg(){
+		if(failedMsgQueue.size() == 0){
+    		return false;
+    	}
+		
+    	Object msg = null;
+    	while(true){
+    		msg = failedMsgQueue.poll();
+    		if(msg == null){
+    			break;
+    		}
+    		
+    		sendFailedMsg(msg);
+    	}
+    	
+    	return true;
+	}
+	
+	public void addFailedMsg(Object msg){
+		if(consistency){
+			failedMsgQueue.offer(msg);
+		}
+	}
+	
+	public void sendFailedMsg(Object msg){
+	}
 
 
 	@Override