将单个输入流的内容同时写入多个输出流

Question

我正在尝试将单个输入流的内容推送到从 tcp 套接字获得的多个输出流。我没有找到任何现有的解决方案，所以我从头开始构建了一些东西，但我真的觉得我在重新发明轮子。我的要求是：

• Java。如果它也能在 Android 上运行就好了，但这是可选的。
• 最多约 10 个客户
• 可以频繁添加/删除客户端
• 每个客户端应接收大致相同的字节/秒
• 添加新客户端时，吞吐量应尽可能减少
• 解决方案不应特定于某些数据（即，当我使用原始 h264 进行测试时，该解决方案应该可以很好地处理文本流）

那么，首先，是否有满足这些要求的库？

如果没有，我该如何提高我自己的解决方案的性能（见下文）。我是这样使用的：

• 在应用程序启动时，我在新线程中运行此类的一个实例。
• 如果客户端连接，则相应套接字的输出流将附加到该实例

虽然它有效，但每个连接有 N 个客户端的客户端的吞吐量极不稳定。似乎没有关于 N 的吞吐量函数。（免责声明：我从单个客户端使用多个线程进行测试）我认为该解决方案的性能主要受两件事的影响：

• 消费者集合的线程同步，添加/删除消费者将阻止写入所有流。
• 该集合的迭代时间，因为 iterator() 可能每次都会创建一个新实例，但我需要那个 remove() 函数。

如有任何建议，我将不胜感激，谢谢。

public class InfiniteStreamingResource implements Runnable {

    private LinkedHashSet<OutputStream> consumers;
    private byte[] buf;
    private boolean running;
    InputStream stream;

    Logger logger = Logger.getLogger(InfiniteStreamingResource.class);

    public InfiniteStreamingResource(InputStream stream, int bufSize) {
        this.stream = stream
        consumers = new LinkedHashSet<>(100);
        buf = new byte[bufSize];
    }

    public synchronized void attachConsumer(OutputStream consumer) {
        consumers.add(consumer);
    }

    public synchronized void stop() {
        running = false;
    }

    @Override
    public void run() {
        running = true;
        int bytesRead;
        Iterator<OutputStream> it;
        OutputStream current;
        try {
            while ((bytesRead = stream.read(buf)) > 0 && running) {
                synchronized(this) {
                    it = consumers.iterator();
                    while (it.hasNext()) {
                        current = it.next();
                        try {
                            current.write(buf, 0, bytesRead);
                        } catch (IOException e) {
                            if (e instanceof SocketException) {
                                it.remove();
                                try {
                                    current.close();
                                } catch (IOException inner) {
                                    //ignore
                                }
                            } else {
                            e.printStackTrace();
                            }
                        }
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        for (OutputStream consumer : consumers) { //if stopped, close any remaining streams
            try {
                consumer.close();
            } catch (IOException e) {
                //ignore
            }
        }

    }


}

Answer 1

我会根据您要发送的数据来决定。

如果是原始字节数据（例如视频流），我不知道库。

如果是文本或 Java 对象，请使用 RabbitMQ 因此，您需要安装代理。这可以接收来自您的发布者的消息并将其转发给所有订阅的客户端。

它易于实现（在 Android 中也是如此）。