IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Java CSP案例

wen java案例 2

本文目录导读：

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风格的并发编程,提供了清晰的进程通信模式和管道处理能力。

上一篇Java HTTPS配置案例

下一篇Java Cookie安全案例

相关文章

抱歉，评论功能暂时关闭!