Làm cách nào để hợp nhất hai luồng đầu vào trong Java?

Question

Có hai InputStream trong Java, có cách nào hợp nhất chúng để bạn kết thúc bằng một InputStream cung cấp cho bạn đầu ra của cả hai luồng không? Làm sao?

Answer 1

Như đã nhận xét, không rõ ý bạn là gì khi hợp nhất.

Sử dụng đầu vào có sẵn "ngẫu nhiên" hoặc là phức tạp bởi InputStream.available không nhất thiết phải cung cấp cho bạn một câu trả lời hữu ích và hoạt động chặn luồng. Bạn sẽ cần hai luồng để đọc từ các luồng và sau đó truyền lại dữ liệu qua, ví dụ: java.io.Piped(In|Out)putStream (mặc dù các lớp đó có vấn đề). Ngoài ra đối với một số loại luồng, có thể sử dụng một giao diện khác, ví dụ: java.nio các kênh không chặn.

Nếu bạn muốn toàn bộ nội dung của luồng đầu vào đầu tiên theo sau là dòng thứ hai: new java.io.SequenceInputStream(s1, s2).

Answer 2

Không phải là tôi có thể nghĩ đến. Bạn có lẽ sẽ phải đọc nội dung của hai dòng vào một byte [] và sau đó tạo ra một ByteArrayInputStream từ đó.

Answer 3

Bạn có thể viết triển khai InputStream tùy chỉnh thực hiện việc này. Ví dụ:

import java.io.IOException; 
import java.io.InputStream; 
import java.util.Collections; 
import java.util.Deque; 
import java.util.LinkedList; 

public class CatInputStream extends InputStream { 
    private final Deque<InputStream> streams; 

    public CatInputStream(InputStream... streams) { 
     this.streams = new LinkedList<InputStream>(); 
     Collections.addAll(this.streams, streams); 
    } 

    private void nextStream() throws IOException { 
     streams.removeFirst().close(); 
    } 

    @Override 
    public int read() throws IOException { 
     int result = -1; 
     while (!streams.isEmpty() 
       && (result = streams.getFirst().read()) == -1) { 
      nextStream(); 
     } 
     return result; 
    } 

    @Override 
    public int read(byte b[], int off, int len) throws IOException { 
     int result = -1; 
     while (!streams.isEmpty() 
       && (result = streams.getFirst().read(b, off, len)) == -1) { 
      nextStream(); 
     } 
     return result; 
    } 

    @Override 
    public long skip(long n) throws IOException { 
     long skipped = 0L; 
     while (skipped < n && !streams.isEmpty()) { 
      int thisSkip = streams.getFirst().skip(n - skipped); 
      if (thisSkip > 0) 
       skipped += thisSkip; 
      else 
       nextStream(); 
     } 
     return skipped; 
    } 

    @Override 
    public int available() throws IOException { 
     return streams.isEmpty() ? 0 : streams.getFirst().available(); 
    } 

    @Override 
    public void close() throws IOException { 
     while (!streams.isEmpty()) 
      nextStream(); 
    } 
} 

Mã này không được kiểm tra, vì vậy số dặm của bạn có thể thay đổi.

Answer 4

java.io.SequenceInputStream có thể là những gì bạn cần. Nó chấp nhận việc đếm các luồng và sẽ xuất nội dung của luồng đầu tiên, sau đó là luồng thứ hai, v.v. cho đến khi tất cả các luồng đều trống.

Answer 5

Dưới đây là triển khai MVar dành riêng cho mảng byte (đảm bảo thêm định nghĩa gói của riêng bạn). Từ đây, nó là tầm thường để viết một luồng đầu vào trên các dòng đã hợp nhất. Tôi cũng có thể đăng bài đó nếu được yêu cầu.

import java.nio.ByteBuffer; 

public final class MVar { 

    private static enum State { 
    EMPTY, ONE, MANY 
    } 

    private final Object lock; 

    private State state; 

    private byte b; 

    private ByteBuffer bytes; 
    private int length; 

    public MVar() { 
    lock = new Object(); 
    state = State.EMPTY; 
    } 

    public final void put(byte b) { 
    synchronized (lock) { 
     while (state != State.EMPTY) { 
     try { 
      lock.wait(); 
     } catch (InterruptedException e) {} 
     } 
     this.b = b; 
     state = State.ONE; 
     lock.notifyAll(); 
    } 
    } 

    public final void put(byte[] bytes, int offset, int length) { 
    if (length == 0) { 
     return; 
    } 
    synchronized (lock) { 
     while (state != State.EMPTY) { 
     try { 
      lock.wait(); 
     } catch (InterruptedException e) {} 
     } 
     this.bytes = ByteBuffer.allocateDirect(length); 
     this.bytes.put(bytes, offset, length); 
     this.bytes.position(0); 
     this.length = length; 
     state = State.MANY; 
     lock.notifyAll(); 
    } 
    } 

    public final byte take() { 
    synchronized (lock) { 
     while (state == State.EMPTY) { 
     try { 
      lock.wait(); 
     } catch (InterruptedException e) {} 
     } 
     switch (state) { 
     case ONE: { 
     state = State.EMPTY; 
     byte b = this.b; 
     lock.notifyAll(); 
     return b; 
     } 
     case MANY: { 
     byte b = bytes.get(); 
     state = --length <= 0 ? State.EMPTY : State.MANY; 
     lock.notifyAll(); 
     return b; 
     } 
     default: 
     throw new AssertionError(); 
     } 
    } 
    } 

    public final int take(byte[] bytes, int offset, int length) { 
    if (length == 0) { 
     return 0; 
    } 
    synchronized (lock) { 
     while (state == State.EMPTY) { 
     try { 
      lock.wait(); 
     } catch (InterruptedException e) {} 
     } 
     switch (state) { 
     case ONE: 
     bytes[offset] = b; 
     state = State.EMPTY; 
     lock.notifyAll(); 
     return 1; 
     case MANY: 
     if (this.length > length) { 
      this.bytes.get(bytes, offset, length); 
      this.length = this.length - length; 
      synchronized (lock) { 
      lock.notifyAll(); 
      } 
      return length; 
     } 
     this.bytes.get(bytes, offset, this.length); 
     this.bytes = null; 
     state = State.EMPTY; 
     length = this.length; 
     lock.notifyAll(); 
     return length; 
     default: 
     throw new AssertionError(); 
     } 
    } 
    } 
}