Как обеспечить возвращение данных потоком

Как обеспечить возвращение данных потоком (Stream) в Java

В Java 8+ Stream API позволяет функционально обрабатывать коллекции данных. Рассмотрю различные способы возвращения данных через потоки.

1. Базовое создание Stream

// Из коллекции
List<String> list = Arrays.asList("A", "B", "C");
Stream<String> stream = list.stream();

// Из массива
String[] array = {"A", "B", "C"};
Stream<String> stream = Arrays.stream(array);

// Создание пустого Stream
Stream<String> emptyStream = Stream.empty();

// Создание Stream с элементами
Stream<Integer> stream = Stream.of(1, 2, 3, 4, 5);

2. Метод, возвращающий Stream

public Stream<String> getNames() {
    List<String> names = Arrays.asList("John", "Jane", "Bob", "Alice");
    return names.stream();
}

// Использование
getNames()
    .filter(name -> name.startsWith("J"))
    .map(String::toUpperCase)
    .forEach(System.out::println);

3. Stream из базы данных

// Spring Data JPA
@Repository
public interface UserRepository extends JpaRepository<User, Long> {
    Stream<User> findAllByActive(boolean active);
}

// Использование
@Service
public class UserService {
    @Autowired
    private UserRepository userRepository;
    
    public void processActiveUsers() {
        try (Stream<User> userStream = userRepository.findAllByActive(true)) {
            userStream
                .map(User::getEmail)
                .forEach(this::sendNotification);
        }
    }
}

4. Stream с фильтрацией и преобразованием

public Stream<ProcessedData> getProcessedData(List<RawData> data) {
    return data.stream()
        .filter(item -> item.isValid())
        .map(item -> transform(item))
        .filter(processed -> processed.getValue() > 0);
}

private ProcessedData transform(RawData data) {
    return new ProcessedData(data.getValue() * 2);
}

// Использование
List<RawData> rawData = Arrays.asList(/* ... */);
getProcessedData(rawData)
    .collect(Collectors.toList());

5. Операции Intermediate (промежуточные)

public Stream<Integer> getProcessedNumbers(List<Integer> numbers) {
    return numbers.stream()
        .filter(n -> n > 0)                    // Фильтр
        .map(n -> n * 2)                       // Преобразование
        .distinct()                            // Уникальные значения
        .sorted()                              // Сортировка
        .limit(10);                            // Ограничение
}

6. Операции Terminal (терминальные)

public class StreamOperations {
    // collect - собрать в коллекцию
    public List<String> toList(Stream<String> stream) {
        return stream.collect(Collectors.toList());
    }
    
    // forEach - применить операцию к каждому
    public void printAll(Stream<String> stream) {
        stream.forEach(System.out::println);
    }
    
    // reduce - свернуть в одно значение
    public int sum(Stream<Integer> stream) {
        return stream.reduce(0, Integer::sum);
    }
    
    // count - количество элементов
    public long countElements(Stream<String> stream) {
        return stream.count();
    }
    
    // findFirst - первый элемент
    public Optional<String> getFirst(Stream<String> stream) {
        return stream.findFirst();
    }
    
    // anyMatch - хоть один соответствует условию
    public boolean hasLongString(Stream<String> stream) {
        return stream.anyMatch(s -> s.length() > 5);
    }
}

7. FlatMap для вложенных Stream

public Stream<String> getAllEmails(List<User> users) {
    return users.stream()
        .flatMap(user -> user.getEmails().stream())
        .distinct();
}

// Пример
List<User> users = Arrays.asList(
    new User("John", Arrays.asList("john@a.com", "john@b.com")),
    new User("Jane", Arrays.asList("jane@a.com"))
);

getAllEmails(users).forEach(System.out::println);
// Вывод:
// john@a.com
// john@b.com
// jane@a.com

8. GroupingBy и другие Collectors

public Map<String, List<User>> groupUsersByCity(Stream<User> users) {
    return users.collect(Collectors.groupingBy(User::getCity));
}

public Map<String, Long> countByCity(Stream<User> users) {
    return users.collect(Collectors.groupingBy(
        User::getCity,
        Collectors.counting()
    ));
}

public Map<String, Optional<User>> oldestByCity(Stream<User> users) {
    return users.collect(Collectors.groupingBy(
        User::getCity,
        Collectors.minBy(Comparator.comparingInt(User::getAge))
    ));
}

public String getUserNames(Stream<User> users) {
    return users.map(User::getName)
        .collect(Collectors.joining(", "));
}

9. Parallel Stream для больших данных

public long countLargeData(List<LargeData> data) {
    return data.parallelStream()  // Параллельный Stream
        .filter(this::isValid)
        .map(this::process)
        .count();
}

// Сравнение производительности
public void comparePerformance() {
    List<Integer> numbers = IntStream.range(0, 1_000_000)
        .boxed()
        .collect(Collectors.toList());
    
    // Последовательный
    long start1 = System.currentTimeMillis();
    long count1 = numbers.stream()
        .filter(n -> n % 2 == 0)
        .count();
    System.out.println("Sequential: " + (System.currentTimeMillis() - start1));
    
    // Параллельный
    long start2 = System.currentTimeMillis();
    long count2 = numbers.parallelStream()
        .filter(n -> n % 2 == 0)
        .count();
    System.out.println("Parallel: " + (System.currentTimeMillis() - start2));
}

10. Полный пример: обработка данных потоком

@Service
public class OrderService {
    @Autowired
    private OrderRepository orderRepository;
    
    // Вернуть Stream заказов с преобразованием
    public Stream<OrderDTO> getActiveOrdersAsStream() {
        return orderRepository.findAllByStatus(OrderStatus.ACTIVE)
            .stream()
            .map(this::convertToDTO)
            .filter(dto -> dto.getTotalAmount().compareTo(BigDecimal.ZERO) > 0);
    }
    
    // Использование Stream для анализа
    public Map<String, BigDecimal> getTotalByUser() {
        return getActiveOrdersAsStream()
            .collect(Collectors.groupingBy(
                OrderDTO::getUserId,
                Collectors.reducing(
                    BigDecimal.ZERO,
                    OrderDTO::getTotalAmount,
                    BigDecimal::add
                )
            ));
    }
    
    // Обработка больших наборов данных
    public void processAllOrders() {
        try (Stream<OrderDTO> orders = getActiveOrdersAsStream()) {
            orders
                .parallel()
                .forEach(this::processOrder);
        }
    }
    
    private OrderDTO convertToDTO(Order order) {
        return new OrderDTO(
            order.getId(),
            order.getUserId(),
            order.getTotalAmount()
        );
    }
    
    private void processOrder(OrderDTO order) {
        // Логика обработки
    }
}

11. Важные замечания

Ленивость Stream:

Stream<Integer> stream = Stream.of(1, 2, 3)
    .map(n -> {
        System.out.println("Map: " + n);
        return n * 2;
    });
// Ничего не выводится до terminal операции!

stream.forEach(System.out::println);
// Теперь выполняется вся цепочка

Stream одноразовый:

Stream<Integer> stream = Stream.of(1, 2, 3);
stream.forEach(System.out::println);
stream.forEach(System.out::println);  // IllegalStateException!

Resource управление:

try (Stream<User> users = userRepository.findAllByActive(true).stream()) {
    users.forEach(System.out::println);
}  // Stream закрывается автоматически

Лучшие практики

1. Используй Stream для функциональной обработки — вместо loops
2. Закрывай Stream из БД — используй try-with-resources
3. Careful с parallelStream — имеет overhead
4. Используй collect для терминирования — вместо forEach где возможно
5. Избегай side effects — функции в Stream должны быть чистыми
6. Ленивость — Stream вычисляются только при terminal операции