Java CSP案例

本文目录导读：

1. Java CSP（通信顺序进程）案例
2. 案例1：使用阻塞队列实现CSP风格
3. 案例2：使用CompletableFuture实现异步CSP
4. 案例3：选择器模式（Select）
5. 案例4：有序通道（Pipeline模式）
6. 关键特点
7. 最佳实践

Java CSP（通信顺序进程）案例

CSP（Communicating Sequential Processes）是一种并发编程模型，通过通道（Channel）在进程（Process）之间进行通信，Java虽然没有原生CSP支持,但可以通过以下方式实现类似模式。

案例1：使用阻塞队列实现CSP风格

生产者-消费者模型

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
// 进程类
class Worker implements Runnable {
    private final BlockingQueue<String> input;
    private final BlockingQueue<String> output;
    private final String name;
    public Worker(BlockingQueue<String> input, BlockingQueue<String> output, String name) {
        this.input = input;
        this.output = output;
        this.name = name;
    }
    @Override
    public void run() {
        try {
            while (true) {
                // 从输入通道接收消息
                String message = input.take();
                if ("DONE".equals(message)) {
                    System.out.println(name + " 收到终止信号");
                    output.put("DONE");
                    break;
                }
                // 处理消息
                String processed = name + "处理: " + message.toUpperCase();
                System.out.println(name + " 发送: " + processed);
                // 发送到输出通道
                output.put(processed);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
public class CSPExample {
    public static void main(String[] args) throws InterruptedException {
        // 创建通道（阻塞队列）
        BlockingQueue<String> channel1 = new ArrayBlockingQueue<>(10);
        BlockingQueue<String> channel2 = new ArrayBlockingQueue<>(10);
        // 创建进程
        Worker worker1 = new Worker(channel1, channel2, "Worker-1");
        Worker worker2 = new Worker(channel2, channel1, "Worker-2");
        // 启动进程
        Thread t1 = new Thread(worker1);
        Thread t2 = new Thread(worker2);
        t1.start();
        t2.start();
        // 发送初始消息
        channel1.put("Hello");
        channel1.put("World");
        channel1.put("CSP");
        // 发送终止信号
        channel1.put("DONE");
        // 等待结束
        t1.join();
        t2.join();
    }
}

案例2：使用CompletableFuture实现异步CSP

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
class CSPProcess {
    private final String name;
    private final ExecutorService executor;
    public CSPProcess(String name, ExecutorService executor) {
        this.name = name;
        this.executor = executor;
    }
    public CompletableFuture<String> process(String input) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println(name + " 接收: " + input);
            // 模拟处理延迟
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            String result = name + " -> " + input.toUpperCase();
            System.out.println(name + " 发送: " + result);
            return result;
        }, executor);
    }
}
public class AsyncCSPExample {
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(4);
        CSPProcess processA = new CSPProcess("Process-A", executor);
        CSPProcess processB = new CSPProcess("Process-B", executor);
        CSPProcess processC = new CSPProcess("Process-C", executor);
        // 构建CSP管道: A -> B -> C
        CompletableFuture<String> future1 = processA.process("hello");
        CompletableFuture<String> future2 = future1.thenCompose(processB::process);
        CompletableFuture<String> future3 = future2.thenCompose(processC::process);
        // 等待最终结果
        String result = future3.get();
        System.out.println("最终结果: " + result);
        executor.shutdown();
    }
}

案例3：选择器模式（Select）

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
class Selector {
    private final ConcurrentLinkedQueue<BlockingQueue<String>> channels = new ConcurrentLinkedQueue<>();
    public void addChannel(BlockingQueue<String> channel) {
        channels.add(channel);
    }
    public String select() throws InterruptedException {
        while (true) {
            for (BlockingQueue<String> channel : channels) {
                String message = channel.poll();
                if (message != null) {
                    return message;
                }
            }
            // 如果没有消息，等待一段时间再试
            Thread.sleep(10);
        }
    }
}
public class SelectCSPExample {
    public static void main(String[] args) {
        BlockingQueue<String> channelA = new ArrayBlockingQueue<>(10);
        BlockingQueue<String> channelB = new ArrayBlockingQueue<>(10);
        BlockingQueue<String> channelC = new ArrayBlockingQueue<>(10);
        Selector selector = new Selector();
        selector.addChannel(channelA);
        selector.addChannel(channelB);
        selector.addChannel(channelC);
        // 启动多个生产者
        new Thread(() -> {
            try {
                for (int i = 0; i < 3; i++) {
                    channelA.put("A-" + i);
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
        new Thread(() -> {
            try {
                for (int i = 0; i < 3; i++) {
                    channelB.put("B-" + i);
                    Thread.sleep(150);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
        new Thread(() -> {
            try {
                for (int i = 0; i < 3; i++) {
                    channelC.put("C-" + i);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
        // 消费者使用选择器接收消息
        new Thread(() -> {
            try {
                for (int i = 0; i < 9; i++) {
                    String message = selector.select();
                    System.out.println("消费: " + message + " 来自通道: " + message.charAt(0));
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    }
}

案例4：有序通道（Pipeline模式）

import java.util.concurrent.*;
import java.util.function.Function;
class Pipeline<T> {
    private final BlockingQueue<T> input;
    private final BlockingQueue<T> output;
    public Pipeline(int capacity) {
        this.input = new ArrayBlockingQueue<>(capacity);
        this.output = new ArrayBlockingQueue<>(capacity);
    }
    public BlockingQueue<T> getInput() {
        return input;
    }
    public BlockingQueue<T> getOutput() {
        return output;
    }
    public static <T> Pipeline<T> connect(Pipeline<T> source, Pipeline<T> target) {
        // 将source的输出连接到target的输入
        new Thread(() -> {
            try {
                while (true) {
                    T item = source.output.take();
                    target.input.put(item);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
        return target;
    }
}
class FilterProcess {
    public static <T> Pipeline<T> create(String name, Function<T, T> filter, int capacity) {
        Pipeline<T> pipeline = new Pipeline<>(capacity);
        new Thread(() -> {
            try {
                while (true) {
                    T item = pipeline.getInput().take();
                    System.out.println(name + " 处理: " + item);
                    T result = filter.apply(item);
                    pipeline.getOutput().put(result);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
        return pipeline;
    }
}
public class PipelineCSPExample {
    public static void main(String[] args) throws InterruptedException {
        // 创建处理阶段
        Pipeline<String> stage1 = FilterProcess.create("Stage-1", 
            s -> s.toLowerCase(), 10);
        Pipeline<String> stage2 = FilterProcess.create("Stage-2", 
            s -> s.replace(" ", "-"), 10);
        Pipeline<String> stage3 = FilterProcess.create("Stage-3", 
            s -> s + "!", 10);
        // 连接管道
        Pipeline.connect(stage1, stage2);
        Pipeline.connect(stage2, stage3);
        // 发送数据
        String[] messages = {"Hello World", "CSP Pattern", "Java is Fun"};
        for (String msg : messages) {
            stage1.getInput().put(msg);
        }
        // 接收最终结果
        for (int i = 0; i < messages.length; i++) {
            String result = stage3.getOutput().take();
            System.out.println("最终结果: " + result);
        }
    }
}

关键特点

1. 通道通信：进程间通过通道交换数据，而不是共享内存
2. 进程隔离：每个进程拥有独立的状态和逻辑
3. 同步通信：默认是阻塞的，发送和接收操作会等待对方
4. 选择机制：可以从多个通道中选择性接收
5. 管道模式：可以构建复杂的处理流程

最佳实践

1. 使用 BlockingQueue 实现通道
2. 使用 CompletableFuture 实现异步操作
3. 避免共享可变状态
4. 使用明确的通道命名和文档
5. 考虑背压和流量控制

这些案例展示了如何在Java中实现CSP风格的并发编程,提供了清晰的进程通信模式和管道处理能力。