Обработка потоков в Java

Question

Я пытаюсь разработать эффективный компонент потоков для моего проекта.Мой код Java читает 1000 файлов фидов, используя потоки (пул потоков).каждый файл имеет 1000 строк, и строка может дублироваться в нескольких файлах.Есть вероятность, что половина информации доступна в одном файле, а другая половина в каком-то другом файле (но у нас есть своя собственная бизнес-логика для корреляции).Мы читаем эти файлы и вставляем в базу данных.в настоящее время мы вставляем одну за другой (используя блокировку для вставки в БД, чтобы удостовериться, что только одна строка на запись из всех файлов), но я хотел создать такую ​​блокировку, чтобы блокировка свойств выполнялась, скажем, данные emp должны быть последовательнымии данные компании вместе, но данные emp и компании должны идти параллельно.Еще одна проблема, с которой мы не знаем тип записи: она динамическая.Завтра я могу получить данные о зарплате.

Answer 1

Я думаю, что вы хотите работать с кучей данных параллельно и иметь разные операции блокировки общих блокировок во время их работы.

Вот пример этого: хранить несколько блокировок на карте, которую выпосмотрите на некоторые сопоставления, которые вы черпаете из своих данных (например, данные сотрудников или компании).

import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class MultipleLocks {

    static Map<Operation.Type, Lock> locks = new ConcurrentHashMap<>();

    public static void main(String... args) {
        Collection<Operation> operations = new HashSet<>();
        operations.add(new Operation(Operation.Type.ALPHA, () -> System.out.println(1)));
        operations.add(new Operation(Operation.Type.ALPHA, () -> System.out.println(2)));
        operations.add(new Operation(Operation.Type.ALPHA, () -> System.out.println(3)));
        operations.add(new Operation(Operation.Type.BRAVO, () -> System.out.println(4)));
        operations.add(new Operation(Operation.Type.BRAVO, () -> System.out.println(5)));
        operations.add(new Operation(Operation.Type.CHARLIE, () -> System.out.println(5)));

        operations.parallelStream().forEach(operation -> {
            Lock lock = locks.computeIfAbsent(operation.type, type -> new ReentrantLock());
            lock.lock();
            operation.op.run();
            lock.unlock();
        });
    }

    static class Operation {
        enum Type {
            ALPHA, BRAVO, CHARLIE;
        }

        Type type;
        Runnable op;

        Operation(Type alpha, Runnable op) {
            type = alpha;
            this.op = op;
        }
    }
}