Что такое Exchanger?

Exchanger: синхронизация между потоками

Exchanger — это класс из пакета java.util.concurrent, который предоставляет механизм для синхронизации между двумя потоками путём обмена данными. Он реализует паттерн producer-consumer, но с уникальной особенностью: оба потока обмениваются данными в точке встречи.

Основная идея

Exchanger работает как встречная точка двух потоков:

Поток A ──────────────┐
                       │ Встреча и обмен
Поток B ──────────────┘

Оба потока блокируются в методе exchange() до тех пор, пока оба не придут в эту точку. Затем они обменивают данные и продолжают работу.

Декларация и использование

// Создание Exchanger для обмена строками
Exchanger<String> exchanger = new Exchanger<>();

// Поток А передаёт "Hello" и получает от потока В
String received = exchanger.exchange("Hello");

// Поток В передаёт "World" и получает от потока А
String received = exchanger.exchange("World");

// После встречи:
// Поток А: переменная received = "World"
// Поток В: переменная received = "Hello"

Пример 1: Простой обмен данными между потоками

public class SimpleExchangerExample {
    public static void main(String[] args) {
        Exchanger<String> exchanger = new Exchanger<>();
        
        // Поток 1
        Thread thread1 = new Thread(() -> {
            try {
                String data1 = "Данные от потока 1";
                System.out.println("Поток 1 отправляет: " + data1);
                
                String received = exchanger.exchange(data1);
                System.out.println("Поток 1 получил: " + received);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        // Поток 2
        Thread thread2 = new Thread(() -> {
            try {
                String data2 = "Данные от потока 2";
                System.out.println("Поток 2 отправляет: " + data2);
                
                String received = exchanger.exchange(data2);
                System.out.println("Поток 2 получил: " + received);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        thread1.start();
        thread2.start();
        
        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

// Вывод:
// Поток 1 отправляет: Данные от потока 1
// Поток 2 отправляет: Данные от потока 2
// Поток 1 получил: Данные от потока 2
// Поток 2 получил: Данные от потока 1

Пример 2: Обмен сложными объектами

public class BufferExchangerExample {
    static class Buffer {
        int[] data;
        int position;
        
        public Buffer(int size) {
            this.data = new int[size];
            this.position = 0;
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        Exchanger<Buffer> exchanger = new Exchanger<>();
        Buffer buffer1 = new Buffer(5);
        Buffer buffer2 = new Buffer(5);
        
        // Производитель
        Thread producer = new Thread(() -> {
            Buffer currentBuffer = buffer1;
            try {
                for (int i = 0; i < 10; i++) {
                    // Заполняем буфер
                    for (int j = 0; j < 5; j++) {
                        currentBuffer.data[j] = i * 5 + j;
                    }
                    System.out.println("Производитель заполнил буфер с данными: " + 
                        Arrays.toString(currentBuffer.data));
                    
                    // Обмениваемся буфером с потребителем
                    currentBuffer = exchanger.exchange(currentBuffer);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        // Потребитель
        Thread consumer = new Thread(() -> {
            Buffer currentBuffer = buffer2;
            try {
                for (int i = 0; i < 10; i++) {
                    // Получаем буфер с данными
                    currentBuffer = exchanger.exchange(currentBuffer);
                    System.out.println("Потребитель получил буфер: " + 
                        Arrays.toString(currentBuffer.data));
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        producer.start();
        consumer.start();
        
        producer.join();
        consumer.join();
    }
}

Пример 3: Обмен с timeout

public class ExchangerWithTimeoutExample {
    public static void main(String[] args) {
        Exchanger<String> exchanger = new Exchanger<>();
        
        Thread thread1 = new Thread(() -> {
            try {
                String data = "Данные потока 1";
                System.out.println("Поток 1 ждёт на обмен...");
                
                // Ждём 2 секунды для встречи
                String received = exchanger.exchange(data, 2, TimeUnit.SECONDS);
                System.out.println("Поток 1 получил: " + received);
            } catch (InterruptedException e) {
                System.out.println("Поток 1 прерван");
                Thread.currentThread().interrupt();
            } catch (TimeoutException e) {
                System.out.println("Поток 1: timeout - другой поток не пришёл");
            }
        });
        
        Thread thread2 = new Thread(() -> {
            try {
                // Подождём 5 секунд перед обменом
                Thread.sleep(5000);
                
                String data = "Данные потока 2";
                System.out.println("Поток 2 ждёт на обмен...");
                
                String received = exchanger.exchange(data, 2, TimeUnit.SECONDS);
                System.out.println("Поток 2 получил: " + received);
            } catch (InterruptedException e) {
                System.out.println("Поток 2 прерван");
                Thread.currentThread().interrupt();
            } catch (TimeoutException e) {
                System.out.println("Поток 2: timeout - другой поток не пришёл");
            }
        });
        
        thread1.start();
        thread2.start();
        
        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

// Вывод:
// Поток 1 ждёт на обмен...
// Поток 1: timeout - другой поток не пришёл
// Поток 2 ждёт на обмен...
// Поток 2: timeout - другой поток не пришёл

Пример 4: Pipeline обработки данных

public class PipelineExample {
    static class Task {
        String name;
        int value;
        
        public Task(String name, int value) {
            this.name = name;
            this.value = value;
        }
        
        @Override
        public String toString() {
            return name + ": " + value;
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        Exchanger<Task> exchanger = new Exchanger<>();
        
        // Этап 1: Парсер
        Thread parser = new Thread(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    Task task = new Task("parsed", i);
                    System.out.println("Парсер: " + task);
                    task = exchanger.exchange(task);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        // Этап 2: Обработчик
        Thread processor = new Thread(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    Task task = exchanger.exchange(null);
                    task.value *= 2; // Обработка
                    task.name = "processed";
                    System.out.println("Обработчик: " + task);
                    task = exchanger.exchange(task);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        parser.start();
        processor.start();
        
        parser.join();
        processor.join();
    }
}

Отличие от других механизмов синхронизации

Основные методы

// Блокирует поток до встречи с другим потоком
V exchange(V x) throws InterruptedException;

// Блокирует с timeout
V exchange(V x, long timeout, TimeUnit unit) 
    throws InterruptedException, TimeoutException;

Важные особенности

1. Двусторонность — оба потока должны вызвать exchange()
2. Синхронность — обмен происходит атомарно
3. Безопасность потоков — thread-safe
4. Blocking — потоки блокируются до встречи
5. InterruptedException — поддерживает прерывание

Когда использовать Exchanger?

1. Обмен буферами — Producer заполняет буфер А, Consumer обрабатывает буфер В
2. Pipeline обработки — несколько этапов передают друг другу данные
3. Синхронизация две-потоковых операций — ровно два потока нужны
4. Genetic algorithms — обмен хромосомами между популяциями

Exchanger — это элегантный и простой инструмент для синхронизации между ровно двумя потоками, когда оба должны встретиться для обмена данными.