java - 相当于套接字连接的 SerialPortEvent

Question

我正在编写一个与测量设备连接的应用程序。该设备可以通过串行和网络连接进行连接。我已经完成了串行端的工作，并捕获使用 SerialPortEvent 发送的数据。

现在我正在尝试使用 socket 完成同样的事情。我的连接正常，并且可以从设备发送/接收数据。问题是我强制 Thread.sleep 等待所有数据准备好。但现在我想像使用 SerialPortEvent 一样自动捕获接收数据。

这是一个由两部分组成的问题:

1. socket 是否有类似的事件？或者在这种情况下定制解决方案更可取？如果是，请添加解释。
2. 如何实现套接字的DATA_AVAILABLE

下面是 SerialEventPort 的代码片段(仅必要的捕获部分)，作为我也想通过套接字完成的任务的引用:

@Override
public void serialEvent(SerialPortEvent event)
{

if (event.getEventType() == SerialPortEvent.DATA_AVAILABLE)
{
  try
  {
    int available = inputStream.available();
    byte[] readBuffer = new byte[available];

    if (available > 0)
    {
      inputStream.read(readBuffer);
    }

  }

}

Answer 1

在此SO answer悲伤陈述如下:

I recommend using java.nio.channels.SocketChannel connected with Selector and SelectionKey. This solution is somewhat event-based, but is more complicated than just plain sockets.

如果您决定采用该解决方案，您将在链接的答案中找到代码示例。

但是，如果您正在谈论 java.net.Socket 那么，不，没有任何事件。我喜欢JTeagle's answer关于类似的问题:

This is often done by spawning a separate thread for the client that continuously makes blocking calls to read() from the stream - that way, as soon as data becomes available the read() call unblocks and can act on what it received ('the event fires'), then it goes back to blocking waiting for the next event.

根据我的经验，这主要是 Java 中处理套接字的方式。我写了一个基于事件的套接字的实现。由于读取是可阻止的，因此很可能需要一个线程来阻止您的主程序:

public class ObservableSocket extends Thread {
    private final Socket socket;
    private final ArrayList<ObservableSocketListener> listeners;
    private volatile boolean isReading;
    private int BUFFER_SIZE = 1024;

    public ObservableSocket(String host, int port) throws UnknownHostException, IOException {
        this.socket = new Socket(host, port);
        this.listeners = new ArrayList<ObservableSocketListener>(1);
        isReading = true;
        this.start();
    }

    public void addListener(ObservableSocketListener l) {
        if (!listeners.contains(l)) {
            listeners.add(l);
        }
    }

    public void removeListener(ObservableSocketListener l) {
        if (!listeners.contains(l)) {
            listeners.remove(l);
        }
    }

    public void die() {
        isReading = false;
        try {
            this.join();
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }
    }

    public void write(byte[] data) throws IOException {
        socket.getOutputStream().write(data);
        socket.getOutputStream().flush();
    }

    private byte[] getData(byte[] buffer, int red) {
        byte[] redData = new byte[red];
        System.arraycopy(buffer, 0, redData, 0, red);
        return redData;
    }

    @Override
    public void run() {
        byte[] buffer = new byte[BUFFER_SIZE];
        int red;
        ObservableSocketEvent event;
        try {
            while (isReading && (red = socket.getInputStream().read(buffer)) > -1) {
                event = new ObservableSocketEvent(this, getData(buffer, red));
                for (ObservableSocketListener l : listeners) {
                    l.dataAvailable(event);
                }
            }
        } 
        catch (Exception exception) {
            event = new ObservableSocketEvent(this, exception);
            for (ObservableSocketListener l : listeners) {
                l.errorOccured(event);
            }
        } 
        finally {
            if (socket != null) {
                try {
                    socket.close();
                    for (ObservableSocketListener l : listeners) {
                        l.closed(new ObservableSocketEvent(this));
                    }
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }   
}

这是您需要实现的监听器类:

public interface ObservableSocketListener extends EventListener {
    public void dataAvailable(ObservableSocketEvent event);
    public void errorOccured(ObservableSocketEvent event);
    public void closed(ObservableSocketEvent event);
}

以及事件类:

public class ObservableSocketEvent extends EventObject {
    private final byte[] data;
    private final Exception exception;

    public ObservableSocketEvent(Object source) {
        super(source);
        this.data = null;
        this.exception = null;
    }

    public ObservableSocketEvent(Object source, byte[] data) {
        super(source);
        this.data = data;
        this.exception = null;
    }

    public ObservableSocketEvent(Object source, Exception exception) {
        super(source);
        this.data = null;
        this.exception = exception;
    }

    public byte[] getData() {
        return data;
    }

    public Exception getException() {
        return exception;
    }      
}

我创建了一个服务器，生成一些随机数据来测试此代码，这就是我在客户端的类主方法中使用它的方式:

    ObservableSocket observableSocket = new ObservableSocket("localhost", 3339);
    observableSocket.addListener(new ObservableSocketListener() {

        @Override
        public void dataAvailable(ObservableSocketEvent event) {
            System.out.println("data received: "+new String(event.getData()));
        }

        @Override
        public void closed(ObservableSocketEvent event) {
            System.out.println("closing socket");
        }

        @Override
        public void errorOccured(ObservableSocketEvent event) {
            System.out.println("error occured");
            event.getException().printStackTrace();
        }


    });
    Thread.currentThread().sleep(10000);
    observableSocket.die();

它输出:

data received: data 0
data received: data 1
data received: data 2
data received: data 3
data received: data 4                        
closing socket                              // thread is alive here
BUILD SUCCESSFUL (total time: 10 seconds)  // thread dies here

根据我的测试，sleep在客户端中是需要的，因为 die方法:

• 退出读取循环(通过将标志设置为 false)
• 并等待线程终止 ( Thread.join )

如果没有 sleep ，测试客户端会立即完成(die 方法有效)。如果没有 die 方法，ObservableSocket 线程在测试结束后仍然存在。

使用此代码时，您应该注意两件事:

• 实例化 ObservableSocket 时<a href="http://docs.oracle.com/javase/1.4.2/docs/api/java/net/Socket.html#Socket(java.net.InetAddress,%20int)" rel="noreferrer noopener nofollow">Socket</a>立即连接并且 <a href="http://docs.oracle.com/javase/7/docs/api/java/lang/Thread.html#start()" rel="noreferrer noopener nofollow">Thread</a>已启动。
• 您必须调用 die来自不是 ObservableSocket 的线程的方法线程(例如，不要从该类中调用该方法)