Log4j2并行流线程上下文管理：确保日志完整性

import org.apache.logging.log4j.ThreadContext;

public class ParallelStreamLogging {

    private static final String BATCH_LOGGER_THREAD_CONTEXT_KEY = "batchConfig";
    private static final String MAIN_THREAD_PREFIX = "main"; // 通常主线程名为"main"

    public void processDataParallel(String configuration, List<String> data) {
        // 在主线程中设置上下文（如果需要，例如在方法外部已经设置）
        // ThreadContext.put(BATCH_LOGGER_THREAD_CONTEXT_KEY, configuration); 

        data.parallelStream().forEach(item -> {
            // 检查当前线程是否为主线程，如果不是，则设置ThreadContext
            if (!Thread.currentThread().getName().startsWith(MAIN_THREAD_PREFIX)) {
                // 将主线程的配置信息传播到工作线程
                ThreadContext.put(BATCH_LOGGER_THREAD_CONTEXT_KEY, configuration);
            }
            // 执行具体的业务逻辑，其中会产生日志
            log.info("Processing item {} in thread {}. Config: {}", item, Thread.currentThread().getName(), ThreadContext.get(BATCH_LOGGER_THREAD_CONTEXT_KEY));
            // ... 其他处理逻辑 ...
        });

        // 在所有并行处理完成后，清理主线程的ThreadContext（如果之前设置过）
        // ThreadContext.remove(BATCH_LOGGER_THREAD_CONTEXT_KEY);
    }
}

import org.apache.logging.log4j.ThreadContext;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ParallelStreamLoggingWithCleanup {

    private static final Logger log = LoggerFactory.getLogger(ParallelStreamLoggingWithCleanup.class);
    private static final String BATCH_LOGGER_THREAD_CONTEXT_KEY = "batchConfig";
    private static final String MAIN_THREAD_PREFIX = "main";

    public void processDataParallel(String configuration, List<String> data) {
        // 在主线程中设置上下文，如果需要
        ThreadContext.put(BATCH_LOGGER_THREAD_CONTEXT_KEY, configuration); 

        try {
            data.parallelStream().forEach(item -> {
                boolean isWorkerThread = !Thread.currentThread().getName().startsWith(MAIN_THREAD_PREFIX);
                if (isWorkerThread) {
                    ThreadContext.put(BATCH_LOGGER_THREAD_CONTEXT_KEY, configuration);
                }
                try {
                    // 执行具体的业务逻辑
                    log.info("Processing item {} in thread {}. Config: {}", item, Thread.currentThread().getName(), ThreadContext.get(BATCH_LOGGER_THREAD_CONTEXT_KEY));
                    // ... 其他处理逻辑 ...
                } finally {
                    // 确保在工作线程完成任务后清理ThreadContext
                    if (isWorkerThread) {
                        ThreadContext.remove(BATCH_LOGGER_THREAD_CONTEXT_KEY);
                    }
                }
            });
        } finally {
            // 在所有并行处理完成后，清理主线程的ThreadContext
            ThreadContext.remove(BATCH_LOGGER_THREAD_CONTEXT_KEY);
        }
    }

    public static void main(String[] args) {
        ParallelStreamLoggingWithCleanup processor = new ParallelStreamLoggingWithCleanup();
        List<String> sampleData = List.of("data1", "data2", "data3", "data4", "data5", "data6", "data7", "data8", "data9", "data10");
        String batchConfig = "Batch_20231027";
        log.info("Starting parallel processing with config: {}", batchConfig);
        processor.processDataParallel(batchConfig, sampleData);
        log.info("Parallel processing finished.");
        // 验证ThreadContext是否已清理
        log.info("ThreadContext after processing: {}", ThreadContext.get(BATCH_LOGGER_THREAD_CONTEXT_KEY));
    }
}