JAVA 와 .NET 에서 동시성 처리및 병렬처리 비교

import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class AsyncExample {
    public static void main(String[] args) {
        // 100개의 랜덤 문자를 생성하여 리스트에 저장
        List<Character> charList = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            charList.add((char) ('A' + (int) (Math.random() * 26)));
        }

        // 리스트의 문자를 2개씩 비동기적으로 출력
        List<CompletableFuture<Void>> futures = new ArrayList<>();
        for (int i = 0; i < charList.size(); i += 2) {
            char finalChar1 = charList.get(i);
            char finalChar2 = (i + 1 < charList.size()) ? charList.get(i + 1) : ' ';
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                System.out.println(finalChar1 + " " + finalChar2);
            });
            futures.add(future);
        }

        // 모든 작업이 완료될 때까지 기다림
        CompletableFuture<Void> allDone = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
        try {
            allDone.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

using System;
using System.Collections.Generic;
using System.Threading.Tasks;

class AsyncExample
{
    static void Main()
    {
        // 100개의 랜덤 문자를 생성하여 리스트에 저장
        List<char> charList = new List<char>();
        Random random = new Random();
        for (int i = 0; i < 100; i++)
        {
            charList.Add((char)('A' + random.Next(26)));
        }

        // 리스트의 문자를 2개씩 비동기적으로 출력
        List<Task> tasks = new List<Task>();
        for (int i = 0; i < charList.Count; i += 2)
        {
            char char1 = charList[i];
            char char2 = (i + 1 < charList.Count) ? charList[i + 1] : ' ';
            Task task = Task.Run(() =>
            {
                Console.WriteLine(char1 + " " + char2);
            });
            tasks.Add(task);
        }

        // 모든 작업이 완료될 때까지 기다림
        Task.WhenAll(tasks).Wait();
    }
}

public class ThreadExample {
    public static void main(String[] args) {
        // 100개의 랜덤 문자를 생성하여 리스트에 저장
        List<Character> charList = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            charList.add((char) ('A' + (int) (Math.random() * 26)));
        }

        // 리스트의 문자를 2개씩 스레드를 사용해 출력
        for (int i = 0; i < charList.size(); i += 2) {
            char finalChar1 = charList.get(i);
            char finalChar2 = (i + 1 < charList.size()) ? charList.get(i + 1) : ' ';
            Thread thread = new Thread(() -> {
                System.out.println(finalChar1 + " " + finalChar2);
            });
            thread.start();
        }
    }
}

using System;
using System.Collections.Generic;
using System.Threading;

class ThreadExample
{
    static void Main()
    {
        // 100개의 랜덤 문자를 생성하여 리스트에 저장
        List<char> charList = new List<char>();
        Random random = new Random();
        for (int i = 0; i < 100; i++)
        {
            charList.Add((char)('A' + random.Next(26)));
        }

        // 리스트의 문자를 2개씩 스레드를 사용해 출력
        for (int i = 0; i < charList.Count; i += 2)
        {
            char char1 = charList[i];
            char char2 = (i + 1 < charList.Count) ? charList[i + 1] : ' ';
            Thread thread = new Thread(() =>
            {
                Console.WriteLine(char1 + " " + char2);
            });
            thread.Start();
        }
    }
}

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolExample {
    public static void main(String[] args) {
        // 스레드풀 생성
        ExecutorService executor = Executors.newFixedThreadPool(10); // 10개의 스레드를 가진 풀

        // 100개의 랜덤 문자를 생성하여 리스트에 저장
        List<Character> charList = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            charList.add((char) ('A' + (int) (Math.random() * 26)));
        }

        // 리스트의 문자를 2개씩 스레드풀을 사용해 출력
        for (int i = 0; i < charList.size(); i += 2) {
            char finalChar1 = charList.get(i);
            char finalChar2 = (i + 1 < charList.size()) ? charList.get(i + 1) : ' ';
            executor.execute(() -> {
                System.out.println(finalChar1 + " " + finalChar2);
            });
        }

        // 스레드풀 종료
        executor.shutdown();
    }
}

using System;
using System.Collections.Generic;
using System.Threading;

class ThreadPoolExample
{
    static void Main()
    {
        // 100개의 랜덤 문자를 생성하여 리스트에 저장
        List<char> charList = new List<char>();
        Random random = new Random();
        for (int i = 0; i < 100; i++)
        {
            charList.Add((char)('A' + random.Next(26)));
        }

        // 리스트의 문자를 2개씩 스레드풀을 사용해 출력
        for (int i = 0; i < charList.Count; i += 2)
        {
            char char1 = charList[i];
            char char2 = (i + 1 < charList.Count) ? charList[i + 1] : ' ';
            ThreadPool.QueueUserWorkItem(_ =>
            {
                Console.WriteLine(char1 + " " + char2);
            });
        }
    }
}

import java.util.stream.IntStream;
import java.util.List;
import java.util.stream.Collectors;

public class PrimeNumbers {
    public static void main(String[] args) {
        List<Integer> primeNumbers = IntStream.rangeClosed(1, 100)
                                              .filter(PrimeNumbers::isPrime)
                                              .boxed()
                                              .collect(Collectors.toList());
        System.out.println(primeNumbers);
    }

    private static boolean isPrime(int number) {
        if (number <= 1) {
            return false;
        }
        return IntStream.rangeClosed(2, (int) Math.sqrt(number))
                        .noneMatch(i -> number % i == 0);
    }
}

using System;
using System.Linq;
using System.Collections.Generic;

public class PrimeNumbers
{
    public static void Main()
    {
        var primeNumbers = Enumerable.Range(1, 100)
                                     .Where(IsPrime)
                                     .ToList();
        Console.WriteLine(String.Join(", ", primeNumbers));
    }

    private static bool IsPrime(int number)
    {
        if (number <= 1) return false;
        for (int i = 2; i <= Math.Sqrt(number); i++)
        {
            if (number % i == 0) return false;
        }
        return true;
    }
}

import java.util.List;
import java.util.stream.Collectors;
import java.util.Arrays;

public class FilterNames {
    public static void main(String[] args) {
        List<String> names = Arrays.asList("Emily", "Michael", "Olivia", "Ethan", "Sophia", "Daniel", "Ava", "Alexander", "Grace", "Lucas");
        List<String> filteredNames = names.stream()
                                          .filter(name -> name.startsWith("A"))
                                          .collect(Collectors.toList());
        System.out.println(filteredNames);
    }
}

using System;
using System.Linq;
using System.Collections.Generic;

public class FilterNames
{
    public static void Main()
    {
        List<string> names = new List<string> { "Emily", "Michael", "Olivia", "Ethan", "Sophia", "Daniel", "Ava", "Alexander", "Grace", "Lucas" };
        var filteredNames = names.Where(name => name.StartsWith("A"))
                                 .ToList();
        Console.WriteLine(String.Join(", ", filteredNames));
    }
}

import java.util.List;
import java.util.stream.Collectors;
import java.util.Arrays;

public class ParallelFilterNames {
    public static void main(String[] args) {
        List<String> names = Arrays.asList("Emily", "Michael", "Olivia", "Ethan", "Sophia", "Daniel", "Ava", "Alexander", "Grace", "Lucas");
        List<String> filteredNames = names.parallelStream()
                                          .filter(name -> name.startsWith("A"))
                                          .collect(Collectors.toList());
        System.out.println(filteredNames);
    }
}

using System;
using System.Linq;
using System.Collections.Generic;

public class ParallelFilterNames
{
    public static void Main()
    {
        List<string> names = new List<string> { "Emily", "Michael", "Olivia", "Ethan", "Sophia", "Daniel", "Ava", "Alexander", "Grace", "Lucas" };
        var filteredNames = names.AsParallel()
                                 .WithDegreeOfParallelism(2)
                                 .Where(name => name.StartsWith("A"))
                                 .ToList();
        Console.WriteLine(String.Join(", ", filteredNames));
    }
}

import io.reactivex.rxjava3.core.Observable;
import java.util.Arrays;
import java.util.List;

public class RxJavaFilterNames {
    public static void main(String[] args) {
        List<String> names = Arrays.asList("Emily", "Michael", "Olivia", "Ethan", "Sophia", "Daniel", "Ava", "Alexander", "Grace", "Lucas");
        
        Observable.fromIterable(names)
                  .filter(name -> name.startsWith("A"))
                  .subscribe(System.out::println);
    }
}

using System;
using System.Linq;
using System.Collections.Generic;
using System.Reactive.Linq;

public class RxNetFilterNames
{
    public static void Main()
    {
        List<string> names = new List<string> { "Emily", "Michael", "Olivia", "Ethan", "Sophia", "Daniel", "Ava", "Alexander", "Grace", "Lucas" };
        names.ToObservable()
             .Where(name => name.StartsWith("A"))
             .Subscribe(Console.WriteLine);
    }
}

import io.reactivex.rxjava3.core.Observable;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class RxJavaRateLimitedFilterNames {
    public static void main(String[] args) {
        List<String> names = Arrays.asList("Emily", "Michael", "Olivia", "Ethan", "Sophia", "Daniel", "Ava", "Alexander", "Grace", "Lucas");

        // Create an Observable that emits a long value every 500 milliseconds
        Observable<Long> interval = Observable.interval(500, TimeUnit.MILLISECONDS);

        Observable.fromIterable(names)
                  .filter(name -> name.startsWith("A"))
                  .zipWith(interval, (name, time) -> name) // Combine with the interval Observable
                  .subscribe(System.out::println);

        // Keep the main thread alive for a short while to see some results
        try {
            Thread.sleep(10000); // Adjust the sleep time as necessary
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

using System;
using System.Linq;
using System.Collections.Generic;
using System.Reactive.Linq;
using System.Reactive.Concurrency;

public class RxNetRateLimitedFilterNames
{
    public static void Main()
    {
        List<string> names = new List<string> { "Emily", "Michael", "Olivia", "Ethan", "Sophia", "Daniel", "Ava", "Alexander", "Grace", "Lucas" };
        
        // Create an Observable sequence that emits a value every 500 milliseconds (2 per second)
        var interval = Observable.Interval(TimeSpan.FromMilliseconds(500));

        names.ToObservable()
             .Where(name => name.StartsWith("A"))
             .Zip(interval, (name, _) => name) // Combine with the interval sequence
             .Subscribe(Console.WriteLine);

        Console.ReadLine(); // Wait for the computation to finish
    }
}

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.schedulers.Schedulers;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class RxJavaAsyncFunctionExample {
    public static void main(String[] args) {
        List<String> names = Arrays.asList("Emily", "Michael", "Olivia", "Ethan", "Sophia", "Daniel", "Ava", "Alexander", "Grace", "Lucas");

        Observable.fromIterable(names)
                  .filter(name -> name.startsWith("A"))
                  .zipWith(Observable.interval(500, TimeUnit.MILLISECONDS), (name, time) -> name)
                  .flatMap(name -> asyncFunction(name)
                                    .subscribeOn(Schedulers.io())) // Run the async function on the IO scheduler
                  .observeOn(Schedulers.single()) // Observe the results on a single thread
                  .subscribe(result -> System.out.println("Processed: " + result),
                             Throwable::printStackTrace);

        // Keep the main thread alive for a short while to see some results
        try {
            Thread.sleep(10000); // Adjust the sleep time as necessary
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static Observable<String> asyncFunction(String name) {
        return Observable.just(name)
                         .map(n -> {
                             // Simulate some asynchronous processing
                             try {
                                 Thread.sleep(1000); // Simulate delay
                             } catch (InterruptedException e) {
                                 e.printStackTrace();
                             }
                             return n.toUpperCase(); // Return some processed result
                         });
    }
}

using System;
using System.Linq;
using System.Collections.Generic;
using System.Reactive.Linq;
using System.Reactive.Concurrency;
using System.Threading.Tasks;

public class RxNetAsyncFunctionExample
{
    public static void Main()
    {
        List<string> names = new List<string> { "Emily", "Michael", "Olivia", "Ethan", "Sophia", "Daniel", "Ava", "Alexander", "Grace", "Lucas" };

        names.ToObservable()
             .Where(name => name.StartsWith("A"))
             .Zip(Observable.Interval(TimeSpan.FromMilliseconds(500)), (name, _) => name)
             .SelectMany(name => AsyncFunction(name)
                                 .ToObservable()
                                 .SubscribeOn(TaskPoolScheduler.Default)) // Execute the async function on the task pool
             .ObserveOn(Scheduler.Default) // Observe the results on the default scheduler
             .Subscribe(result => Console.WriteLine("Processed: " + result),
                        ex => Console.WriteLine($"Error: {ex.Message}"));

        Console.ReadLine(); // Wait for the computation to finish
    }

    private static async Task<string> AsyncFunction(string name)
    {
        // Simulate some asynchronous processing
        await Task.Delay(1000); // Simulate delay
        return name.ToUpper(); // Return some processed result
    }
}

import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.stream.ActorMaterializer;
import akka.stream.javadsl.Source;
import akka.stream.javadsl.Sink;
import java.time.Duration;
import java.util.Arrays;

public class AkkaStreamExample {
    public static void main(String[] args) {
        final ActorSystem system = ActorSystem.create("stream-example");
        final ActorMaterializer materializer = ActorMaterializer.create(system);

        // 배열에 저장된 이름들
        String[] names = {"Emily", "Michael", "Olivia", "Ethan", "Sophia", 
                          "Daniel", "Ava", "Alexander", "Grace", "Lucas"};

        // Source 생성
        Source<String, NotUsed> source = Source.from(Arrays.asList(names));

        // 초당 2개의 요소를 출력하는 Sink
        source.throttle(2, Duration.ofSeconds(1))
              .runWith(Sink.foreach(System.out::println), materializer);
    }
}

using Akka.Actor;
using Akka.Streams;
using Akka.Streams.Dsl;
using System;
using System.Linq;

class AkkaNetStreamExample
{
    static void Main(string[] args)
    {
        using (var system = ActorSystem.Create("stream-example"))
        using (var materializer = system.Materializer())
        {
            // 배열에 저장된 이름들
            string[] names = {"Emily", "Michael", "Olivia", "Ethan", "Sophia", 
                              "Daniel", "Ava", "Alexander", "Grace", "Lucas"};

            // Source 생성
            var source = Source.From(names);

            // 초당 2개의 요소를 출력하는 Sink
            source.Throttle(2, TimeSpan.FromSeconds(1))
                  .RunWith(Sink.ForEach<string>(Console.WriteLine), materializer)
                  .Wait();
        }
    }
}

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Flux;
import java.time.Duration;

import static org.springframework.web.reactive.function.server.RouterFunctions.route;

@SpringBootApplication
public class WebfluxExample {

    public static void main(String[] args) {
        SpringApplication.run(WebfluxExample.class, args);
    }

    @Bean
    public RouterFunction<ServerResponse> routerFunction() {
        return route()
            .GET("/stream-names", request -> {
                String[] names = {"Emily", "Michael", "Olivia", "Ethan", "Sophia",
                                  "Daniel", "Ava", "Alexander", "Grace", "Lucas"};

                Flux<String> nameStream = Flux.fromArray(names)
                                              .zipWith(Flux.interval(Duration.ofSeconds(1)))
                                              .map(tuple -> tuple.getT1());

                return ServerResponse.ok().body(nameStream, String.class);
            })
            .build();
    }
}

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import reactor.core.publisher.Flux;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

public class ReactiveKafkaToS3 {

    private final S3Client s3Client;
    private final ReceiverOptions<String, String> receiverOptions;

    public ReactiveKafkaToS3(S3Client s3Client) {
        this.s3Client = s3Client;

        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-broker:9092");
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "your-client-id");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "your-group-id");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        receiverOptions = ReceiverOptions.create(props);
    }

    public void consumeAndStore() {
        KafkaReceiver.create(receiverOptions.subscription(Collections.singleton("your-kafka-topic")))
                .receive()
                .flatMap(record -> storeInS3(record))
                .subscribe();
    }

    private Flux<Void> storeInS3(ConsumerRecord<String, String> record) {
        PutObjectRequest putObjectRequest = PutObjectRequest.builder()
                .bucket("your-s3-bucket")
                .key("your-object-key")
                .build();

        return Flux.fromFuture(s3Client.putObject(putObjectRequest, software.amazon.awssdk.core.sync.RequestBody.fromString(record.value())).thenRun(() -> {
            // S3에 성공적으로 저장된 후 처리
            System.out.println("Stored in S3: " + record.value());
        }));
    }

    public static void main(String[] args) {
        // S3 클라이언트 초기화 및 서비스 시작
        S3Client s3Client = S3Client.builder().build();
        new ReactiveKafkaToS3(s3Client).consumeAndStore();
    }
}

<dependencies>
    <!-- 기존 의존성 -->

    <!-- Apache Parquet와 Avro -->
    <dependency>
        <groupId>org.apache.parquet</groupId>
        <artifactId>parquet-avro</artifactId>
        <version>YOUR_PARQUET_VERSION</version>
    </dependency>
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>YOUR_AVRO_VERSION</version>
    </dependency>
</dependencies>

private Schema avroSchema; // Avro 스키마

public ReactiveKafkaToS3(S3Client s3Client, String schemaString) {
    this.s3Client = s3Client;
    this.avroSchema = new Schema.Parser().parse(schemaString); // Avro 스키마 초기화

    // 기존 초기화 로직...
}

private Flux<Void> storeInS3(ConsumerRecord<String, String> record) {
    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();

    try (ParquetWriter<GenericRecord> writer = AvroParquetWriter.<GenericRecord>builder(new org.apache.hadoop.fs.Path("data.parquet"))
            .withSchema(avroSchema)
            .withCompressionCodec(CompressionCodecName.SNAPPY)
            .build()) {

        GenericRecord avroRecord = new GenericData.Record(avroSchema);
        avroRecord.put("your_field", record.value()); // Kafka 레코드의 값을 Avro 레코드에 셋팅
        writer.write(avroRecord);
    } catch (IOException e) {
        e.printStackTrace();
    }

    PutObjectRequest putObjectRequest = PutObjectRequest.builder()
            .bucket("your-s3-bucket")
            .key("your-object-key.parquet")
            .build();

    return Flux.fromFuture(s3Client.putObject(putObjectRequest, software.amazon.awssdk.core.sync.RequestBody.fromBytes(outputStream.toByteArray())).thenRun(() -> {
        System.out.println("Stored in S3: " + record.value());
    }));
}