جمع کننده های جریان: روشی جدید برای دستکاری جریان های جاوا

اولین تور با کد از اپراتورهای جریان قابل تنظیم جدید در رابط Java.util.stream.Gatherers جاوا ۲۲.

جاوا ۲۲ جمع‌آورنده‌های جریان را معرفی می‌کند، مکانیزم جدیدی برای دستکاری جریان‌های داده ها. جمع‌آورنده‌های جریان ویژگی ارائه‌شده برای JEP 461 هستند که به توسعه‌دهندگان اجازه می‌دهند اپراتورهای میانی سفارشی ایجاد کنند که عملیات پیچیده را ساده می‌کند. در نگاه اول، گردآورنده‌های جریان کمی پیچیده و مبهم به نظر می‌رسند، و ممکن است تعجب کنید که چرا به آنها نیاز دارید. اما هنگامی که با وضعیتی مواجه می شوید که به نوع خاصی از دستکاری جریان نیاز دارد، گردآورنده ها به یک افزونه واضح و خوشایند برای Stream API تبدیل می شوند.

API Stream و گردآورنده‌های جریان

جریان‌های جاوا مجموعه‌های پویا از عناصر را مدل‌سازی می‌کنند. همانطور که مشخصات می‌گوید، “جریان یک دنباله مقادیر بالقوه نامحدود است.”

این بدان معناست که می‌توانید به‌طور بی‌پایان جریان‌های داده را مصرف کرده و روی آن‌ها کار کنید. فکر کنید که در کنار رودخانه ای نشسته اید و جریان آب را تماشا می کنید. شما هرگز فکر نمی کنید منتظر بمانید تا رودخانه تمام شود. با نهرها، شما تازه کار را با رودخانه و هر چیزی که در آن وجود دارد شروع می کنید. وقتی کارتان تمام شد، کنار می‌روید.

API جریان چندین روش داخلی دارد برای کار بر روی عناصر در یک دنباله از مقادیر. اینها اپراتورهای عملکردی مانند فیلتر هستند. و نقشه. 

در Stream API، جریان‌ها با منبع رویدادها شروع می‌شوند و عملیاتی مانند filter و map به عنوان عملیات “واسطه” شناخته می‌شوند. هر عملیات میانی جریان را برمی گرداند، بنابراین می توانید آنها را با هم بسازید. اما با استفاده از Stream API، جاوا شروع به اعمال هیچ یک از این عملیات نمی‌کند تا زمانی که جریان به عملیات «پایانه» برسد. این از پردازش کارآمد حتی با بسیاری از اپراتورهای زنجیر شده با هم پشتیبانی می کند.

اپراتورهای میانی داخلی Stream قدرتمند هستند، اما نمی‌توانند کل حوزه مورد نیاز قابل تصور را پوشش دهند. برای موقعیت هایی که خارج از جعبه هستند، ما به راهی برای تعریف عملیات سفارشی نیاز داریم. گردآورندگان این راه را به ما می دهند.

چه کاری می توانید با جمع کننده های جریان انجام دهید

بگویید که در کنار رودخانه هستید و برگها با اعدادی که روی آنها نوشته شده است شناور هستند. اگر می‌خواهید کار ساده‌ای انجام دهید، مانند ایجاد آرایه‌ای از تمام اعداد زوجی که می‌بینید، می‌توانید از روش filter داخلی استفاده کنید:

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6);
numbers.stream().filter(number -> number % 2 == 0).toArray()
// result: { 2, 4, 6 }

در مثال بالا، با آرایه‌ای از اعداد صحیح (منبع) شروع می‌کنیم و سپس آن را به یک جریان تبدیل می‌کنیم و فیلتری را اعمال می‌کنیم که فقط آن اعدادی را که تقسیم بر دو آنها باقی نمی‌ماند، برمی‌گرداند. تماس toArray() تماس ترمینال است. این معادل بررسی یکنواختی هر برگ و کنار گذاشتن آن در صورت عبور است.

روش های داخلی گردآورندگان جریانی

رابط java.util.stream.Gatherers دارای تعداد زیادی توابع داخلی است که به شما امکان می دهد عملیات میانی سفارشی بسازید. بیایید نگاهی به کارهایی که هر کدام انجام می دهند بیاندازیم.

روش windowFixed

اگر بخواهید تمام برگ های شناور را بردارید و آنها را در سطل های دوتایی جمع کنید چه؟ این به‌طور شگفت‌انگیزی سخت‌تر است که با داخلی انجام می‌شود عملگرهای عملکردی این نیاز به تبدیل یک آرایه تک رقمی به آرایه ای از آرایه ها دارد. 

روش windowFixed راه ساده‌تری برای جمع‌آوری برگ‌های شما در سطل‌ها است:

Stream.iterate(0, i -> i + 1)
  .gather(Gatherers.windowFixed(2))
  .limit(5)
  .collect(Collectors.toList());

این می گوید: یک جریان بر اساس تکرار اعداد صحیح با ۱ به من بدهید. هر دو عنصر را به یک آرایه جدید تبدیل کنید. پنج بار انجام دهید. در نهایت، جریان را به List تبدیل کنید. نتیجه این است:

[[۰, ۱], [۲, ۳], [۴, ۵], [۶, ۷], [۸, ۹]]

پنجره مانند حرکت دادن یک قاب بر روی جریان است. به شما امکان می دهد عکس های فوری بگیرید. 

روش windowSliding

یکی دیگر از عملکردهای پنجره، windowSliding، که مانند windowFixed() کار می‌کند، به جز اینکه هر پنجره به‌جای پایان آخرین پنجره، از عنصر بعدی در آرایه مبدأ شروع می‌شود. این یک مثال است:

Stream.iterate(0, i -> i + 1)
   .gather(Gatherers.windowSliding(2))
   .limit(5)
   .collect(Collectors.toList());

خروجی این است:

[[۰, ۱], [۱, ۲], [۲, ۳], [۳, ۴], [۴, ۵]]

خروجی windowSliding را با خروجی windowFixed مقایسه کنید و تفاوت را خواهید دید. هر زیرآرایه در windowSliding حاوی آخرین عنصر زیرآرایه قبلی است، برخلاف windowFixed.

روش Gatherers.fold

Gatherers.fold مانند نسخه اصلاح شده Stream.reduce. دیدن اینکه در کجای fold() نسبت به reduce() مفید است، کمی ظریف است. یک بحث خوب در این مقاله یافت می شود. در اینجا چیزی است که نویسنده، ویکتور کلانگ، درباره تفاوت‌های بین fold و reduce می‌گوید:

تاکردن تعمیم کاهش است. با کاهش، نوع نتیجه با نوع عنصر یکسان است، ترکیب کننده تداعی کننده است و مقدار اولیه یک هویت برای ترکیب کننده است. برای یک فولد، این شرایط لازم نیست، اگرچه ما از موازی‌پذیری صرف نظر می‌کنیم.

بنابراین می بینیم که reduce نوعی fold است. کاهش یک جریان را می گیرد و آن را به یک مقدار واحد تبدیل می کند. Folding نیز این کار را انجام می دهد، اما الزامات را کاهش می دهد: ۱) اینکه نوع برگشتی از همان نوع عناصر جریان باشد. ۲) اینکه ترکیب کننده تداعی کننده است. و ۳) اینکه مقدار اولیه موجود در fold یک تابع مولد واقعی است، نه یک مقدار استاتیک.

نیاز دوم مربوط به موازی سازی است که به زودی در مورد آن با جزئیات بیشتر صحبت خواهم کرد. فراخوانی Stream.parallel در یک جریان به این معنی است که موتور می‌تواند کار را به چند رشته تقسیم کند. این فقط در صورتی کار می کند که اپراتور انجمنی باشد. یعنی اگر ترتیب عملیات بر نتیجه تأثیری نداشته باشد، کار می کند.

در اینجا یک استفاده ساده از fold وجود دارد:

Stream.of("hello","world","how","are","you?")
  .gather(
    Gatherers.fold(() -> "", 
      (acc, element) -> acc.isEmpty() ? element : acc + "," + element
    )
   )
  .findFirst()
  .get();

این مثال مجموعه ای از رشته ها را می گیرد و آنها را با کاما ترکیب می کند. همان کار انجام شده توسط reduce:

String result = Stream.of("hello", "world", "how", "are", "you?")
  .reduce("", (acc, element) -> acc.isEmpty() ? element : acc + "," + element);

می توانید ببینید که با fold، یک تابع (() -> "") را به جای مقدار اولیه ("").  این بدان معناست که اگر به مدیریت پیچیده تری از آغازگر نیاز دارید، می توانید از تابع closure استفاده کنید. 

var result = Stream.of(1,"hello", true).gather(Gatherers.fold(() -> 0, (acc, el) -> acc + 1));
// result.findFirst().get() = 3

// bad, throws exception:
var result = Stream.of(1, "hello", true).reduce(0, (acc, el) -> acc + 1);
// Error: bad operand types for binary operator '+'

var result2 = Stream.of("apple", "banana", "apple", "orange")
  .collect(Collectors.toMap(word -> word, word -> 1, Integer::sum, HashMap::new));

Stream.of(1,2,3,4,5,6,7,8,9)
  .gather(
    Gatherers.scan(() -> "", (string, number) -> string + number)
  )
  .toList();

["۱", "۱۲", "۱۲۳", "۱۲۳۴", "۱۲۳۴۵", "۱۲۳۴۵۶", "۱۲۳۴۵۶۷", "۱۲۳۴۵۶۷۸", "۱۲۳۴۵۶۷۸۹"]

Stream.of(1,2,3,4,5).gather(Gatherers.mapConcurrent(4, x -> x * x)).collect(Collectors.toList());
// Result: [1, 4, 9, 16, 25]