用 InputStream 包装 ByteBuffer

问题描述 投票:0回答:7

我有一个接受 InputStream 并从中读取数据的方法。我也想将此方法与 ByteBuffer 一起使用。有没有办法包装一个 ByteBuffer 以便它可以作为流访问?

java nio inputstream bytebuffer
7个回答
85
投票

Thilo提到的实现似乎有一些bug,也一字不差地复制粘贴到其他网站:

  1. ByteBufferBackedInputStream.read()
    返回它读取的字节的符号扩展 int 表示,这是错误的(值应该在 [-1..255] 范围内)
  2. 根据 API 规范,当缓冲区中没有剩余字节时,
  3. ByteBufferBackedInputStream.read(byte[], int, int)
    不会返回 -1

ByteBufferBackedOutputStream 看起来比较健全。

我在下面展示了一个“固定”版本。如果我发现更多错误(或有人指出),我会在这里更新。

更新:从读/写方法中删除

synchronized
关键字

输入流

public class ByteBufferBackedInputStream extends InputStream {

    ByteBuffer buf;

    public ByteBufferBackedInputStream(ByteBuffer buf) {
        this.buf = buf;
    }

    public int read() throws IOException {
        if (!buf.hasRemaining()) {
            return -1;
        }
        return buf.get() & 0xFF;
    }

    public int read(byte[] bytes, int off, int len)
            throws IOException {
        if (!buf.hasRemaining()) {
            return -1;
        }

        len = Math.min(len, buf.remaining());
        buf.get(bytes, off, len);
        return len;
    }
}

输出流

public class ByteBufferBackedOutputStream extends OutputStream {
    ByteBuffer buf;

    public ByteBufferBackedOutputStream(ByteBuffer buf) {
        this.buf = buf;
    }

    public void write(int b) throws IOException {
        buf.put((byte) b);
    }

    public void write(byte[] bytes, int off, int len)
            throws IOException {
        buf.put(bytes, off, len);
    }

}

16
投票

JDK 中没有,但是有很多实现,谷歌搜索 ByteBufferInputStream。基本上,它们包装一个或多个 ByteBuffer 并跟踪其中的一个索引,该索引记录了已经读取了多少。 像这样的东西经常出现,但显然是有问题的,请参阅@Mike Houston 的改进版本的答案)。


7
投票

如果它由字节数组支持,您可以使用

ByteArrayInputStream
并通过
ByteBuffer.array()
获取字节数组。如果您在本机 ByteBuffer 上尝试,这将引发异常。


3
投票

如果可用,直接使用堆缓冲区(字节数组),否则使用包装字节缓冲区(参见答案 Mike Houston)

public static InputStream asInputStream(ByteBuffer buffer) {
    if (buffer.hasArray()) {
        // use heap buffer; no array is created; only the reference is used
        return new ByteArrayInputStream(buffer.array());
    }
    return new ByteBufferInputStream(buffer);
}

另请注意,包装缓冲区可以有效地支持标记/重置和跳过操作。


3
投票

这是我的

InputStream
OutputStream
实现的版本:

ByteBufferBackedInputStream

public class ByteBufferBackedInputStream extends InputStream
{
  private ByteBuffer backendBuffer;

  public ByteBufferBackedInputStream(ByteBuffer backendBuffer) {
      Objects.requireNonNull(backendBuffer, "Given backend buffer can not be null!");
      this.backendBuffer = backendBuffer;
  }

  public void close() throws IOException {
      this.backendBuffer = null;
  }

  private void ensureStreamAvailable() throws IOException {
      if (this.backendBuffer == null) {
          throw new IOException("read on a closed InputStream!");
      }
  }

  @Override
  public int read() throws IOException {
      this.ensureStreamAvailable();
      return this.backendBuffer.hasRemaining() ? this.backendBuffer.get() & 0xFF : -1;
  }

  @Override
  public int read(@Nonnull byte[] buffer) throws IOException {
      return this.read(buffer, 0, buffer.length);
  }

  @Override
  public int read(@Nonnull byte[] buffer, int offset, int length) throws IOException {
      this.ensureStreamAvailable();
      Objects.requireNonNull(buffer, "Given buffer can not be null!");
      if (offset >= 0 && length >= 0 && length <= buffer.length - offset) {
          if (length == 0) {
              return 0;
          }
          else {
              int remainingSize = Math.min(this.backendBuffer.remaining(), length);
              if (remainingSize == 0) {
                  return -1;
              }
              else {
                  this.backendBuffer.get(buffer, offset, remainingSize);
                  return remainingSize;
              }
          }
      }
      else {
          throw new IndexOutOfBoundsException();
      }
  }

  public long skip(long n) throws IOException {
      this.ensureStreamAvailable();
      if (n <= 0L) {
          return 0L;
      }
      int length = (int) n;
      int remainingSize = Math.min(this.backendBuffer.remaining(), length);
      this.backendBuffer.position(this.backendBuffer.position() + remainingSize);
      return (long) length;
  }

  public int available() throws IOException {
      this.ensureStreamAvailable();
      return this.backendBuffer.remaining();
  }

  public synchronized void mark(int var1) {
  }

  public synchronized void reset() throws IOException {
      throw new IOException("mark/reset not supported");
  }

  public boolean markSupported() {
      return false;
  }
}

ByteBufferBackedOutputStream

public class ByteBufferBackedOutputStream extends OutputStream
{
    private ByteBuffer backendBuffer;

    public ByteBufferBackedOutputStream(ByteBuffer backendBuffer) {
        Objects.requireNonNull(backendBuffer, "Given backend buffer can not be null!");
        this.backendBuffer = backendBuffer;
    }

    public void close() throws IOException {
        this.backendBuffer = null;
    }

    private void ensureStreamAvailable() throws IOException {
        if (this.backendBuffer == null) {
            throw new IOException("write on a closed OutputStream");
        }
    }

    @Override
    public void write(int b) throws IOException {
        this.ensureStreamAvailable();
        backendBuffer.put((byte) b);
    }

    @Override
    public void write(@Nonnull byte[] bytes) throws IOException {
        this.write(bytes, 0, bytes.length);
    }

    @Override
    public void write(@Nonnull byte[] bytes, int off, int len) throws IOException {
        this.ensureStreamAvailable();
        Objects.requireNonNull(bytes, "Given buffer can not be null!");
        if ((off < 0) || (off > bytes.length) || (len < 0) ||
            ((off + len) > bytes.length) || ((off + len) < 0))
        {
            throw new IndexOutOfBoundsException();
        }
        else if (len == 0) {
            return;
        }

        backendBuffer.put(bytes, off, len);
    }
}

2
投票

基于 ByteArrayInputStream 代码的衍生... 要求提供的 ByteBuffer 预先根据需要正确设置位置和限制。

    public class ByteBufferInputStream extends InputStream
    {
        /**
         * The input ByteBuffer that was provided.
         * The ByteBuffer should be supplied with position and limit correctly set as appropriate
         */
        protected ByteBuffer buf;

        public ByteBufferInputStream(ByteBuffer buf)
        {
            this.buf = buf;
            buf.mark(); // to prevent java.nio.InvalidMarkException on InputStream.reset() if mark had not been set
        }

        /**
         * Reads the next byte of data from this ByteBuffer. The value byte is returned as an int in the range 0-255.
         * If no byte is available because the end of the buffer has been reached, the value -1 is returned.
         * @return  the next byte of data, or -1 if the limit/end of the buffer has been reached.
         */
        public int read()
        {
            return buf.hasRemaining()
                ? (buf.get() & 0xff)
                : -1;
        }

        /**
         * Reads up to len bytes of data into an array of bytes from this ByteBuffer.
         * If the buffer has no remaining bytes, then -1 is returned to indicate end of file.
         * Otherwise, the number k of bytes read is equal to the smaller of len and buffer remaining.
         * @param   b     the buffer into which the data is read.
         * @param   off   the start offset in the destination array b
         * @param   len   the maximum number of bytes read.
         * @return  the total number of bytes read into the buffer, or -1 if there is no more data because the limit/end of
         *          the ByteBuffer has been reached.
         * @exception  NullPointerException If b is null.
         * @exception  IndexOutOfBoundsException If off is negative, len is negative, or len is greater than b.length - off
         */
        public int read(byte b[], int off, int len)
        {
            if (b == null)
            {
                throw new NullPointerException();
            }
            else if (off < 0 || len < 0 || len > b.length - off)
            {
                throw new IndexOutOfBoundsException();
            }

            if (!buf.hasRemaining())
            {
                return -1;
            }

            int remaining = buf.remaining();
            if (len > remaining)
            {
                len = remaining;
            }

            if (len <= 0)
            {
                return 0;
            }

            buf.get(b, off, len);

            return len;
        }

        /**
         * Skips n bytes of input from this ByteBuffer. Fewer bytes might be skipped if the limit is reached.
         *
         * @param   n   the number of bytes to be skipped.
         * @return  the actual number of bytes skipped.
         */
        public long skip(long n)
        {
            int skipAmount = (n < 0)
                ? 0
                : ((n > Integer.MAX_VALUE)
                ? Integer.MAX_VALUE
                : (int) n);

            if (skipAmount > buf.remaining())
            {
                skipAmount = buf.remaining();
            }

            int newPos = buf.position() + skipAmount;

            buf.position(newPos);

            return skipAmount;
        }

        /**
         * Returns remaining bytes available in this ByteBuffer
         * @return the number of remaining bytes that can be read (or skipped over) from this ByteBuffer.
         */
        public int available()
        {
            return buf.remaining();
        }

        public boolean markSupported()
        {
            return true;
        }

        /**
         * Set the current marked position in the ByteBuffer.
         * <p> Note: The readAheadLimit for this class has no meaning.
         */
        public void mark(int readAheadLimit)
        {
            buf.mark();
        }

        /**
         * Resets the ByteBuffer to the marked position.
         */
        public void reset()
        {
            buf.reset();
        }

        /**
         * Closing a ByteBuffer has no effect.
         * The methods in this class can be called after the stream has been closed without generating an IOException.
         */
        public void close() throws IOException
        {
        }
    }

0
投票

复制粘贴 ByteBufferBackedInputStream 并想使用它后,IDE 提示我 Jackson 已经有一个。所以我把它贴在这里作为参考:

package com.fasterxml.jackson.databind.util;

import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;

/**
 * Simple {@link InputStream} implementation that exposes currently
 * available content of a {@link ByteBuffer}.
 */
public class ByteBufferBackedInputStream extends InputStream {
    protected final ByteBuffer _b;

    public ByteBufferBackedInputStream(ByteBuffer buf) { _b = buf; }

    @Override public int available() { return _b.remaining(); }
    
    @Override
    public int read() throws IOException { return _b.hasRemaining() ? (_b.get() & 0xFF) : -1; }

    @Override
    public int read(byte[] bytes, int off, int len) throws IOException {
        if (!_b.hasRemaining()) return -1;
        len = Math.min(len, _b.remaining());
        _b.get(bytes, off, len);
        return len;
    }
}
© www.soinside.com 2019 - 2024. All rights reserved.