Как посчитать записи обработанные несколькими потоками

Как посчитать записи, обработанные несколькими потоками

При работе с многопоточностью счётчик должен быть потокобезопасным. Java предоставляет несколько способов достичь этого.

1. Использование synchronized

Самый простой способ — синхронизация метода:

public class ThreadSafeCounter {
    private int count = 0;
    
    public synchronized void increment() {
        count++;
    }
    
    public synchronized int getCount() {
        return count;
    }
}

// Использование
ThreadSafeCounter counter = new ThreadSafeCounter();
ExecutorService executor = Executors.newFixedThreadPool(4);

for (int i = 0; i < 100; i++) {
    executor.submit(() -> {
        counter.increment();
    });
}

executor.shutdown();
executor.awaitTermination(1, TimeUnit.MINUTES);
System.out.println("Обработано: " + counter.getCount());

Однако synchronized может привести к снижению производительности при высокой конкуренции.

2. AtomicInteger (рекомендуется)

Более эффективный способ используя lock-free алгоритмы:

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicCounterExample {
    private AtomicInteger count = new AtomicInteger(0);
    
    public void processRecord() {
        count.incrementAndGet();
    }
    
    public int getCount() {
        return count.get();
    }
}

// Использование
AtomicCounterExample processor = new AtomicCounterExample();
ExecutorService executor = Executors.newFixedThreadPool(10);

for (int i = 0; i < 1000; i++) {
    executor.submit(() -> processor.processRecord());
}

executor.shutdown();
executor.awaitTermination(1, TimeUnit.MINUTES);
System.out.println("Обработано записей: " + processor.getCount());

AtomicInteger использует CAS (Compare-And-Swap) операции, избегая блокировок.

3. AtomicLong для больших значений

public class LongCounterExample {
    private AtomicLong recordsProcessed = new AtomicLong(0);
    
    public void processRecord(String data) {
        // Обработка записи
        recordsProcessed.incrementAndGet();
    }
    
    public long getTotalProcessed() {
        return recordsProcessed.get();
    }
}

4. LongAdder для высокой конкуренции

Для очень нагруженных систем:

import java.util.concurrent.atomic.LongAdder;

public class HighPerformanceCounter {
    private LongAdder counter = new LongAdder();
    
    public void processRecord() {
        counter.increment();
    }
    
    public long getCount() {
        return counter.sum();
    }
}

// Тест производительности
HighPerformanceCounter counter = new HighPerformanceCounter();
ExecutorService executor = Executors.newFixedThreadPool(16);
Long startTime = System.nanoTime();

for (int i = 0; i < 100000; i++) {
    executor.submit(() -> counter.processRecord());
}

executor.shutdown();
executor.awaitTermination(2, TimeUnit.MINUTES);
Long endTime = System.nanoTime();

System.out.println("Обработано: " + counter.getCount());
System.out.println("Время: " + (endTime - startTime) / 1_000_000 + "ms");

5. Практический пример: обработка файла несколькими потоками

public class FileProcessor {
    private final LongAdder processedRecords = new LongAdder();
    private final ExecutorService executor;
    
    public FileProcessor(int threadCount) {
        executor = Executors.newFixedThreadPool(threadCount);
    }
    
    public void processFile(Path filePath) throws IOException {
        Files.lines(filePath).forEach(line -> {
            executor.submit(() -> {
                try {
                    processLine(line);
                    processedRecords.increment();
                } catch (Exception e) {
                    System.err.println("Ошибка обработки: " + e.getMessage());
                }
            });
        });
    }
    
    private void processLine(String line) {
        // Логика обработки строки
    }
    
    public long getProcessedCount() {
        return processedRecords.sum();
    }
    
    public void shutdown() {
        executor.shutdown();
    }
}

Сравнение подходов

Заключение

Для большинства случаев используйте AtomicInteger или AtomicLong. Если проект требует максимальной производительности при высокой конкуренции, выбирайте LongAdder. Избегайте synchronized для счётчиков — это устарелый подход.