《Java核心技术II》可中断套接字

4.2.4 可中断套接字

SocketChannel可以中断套接字

SocketChannel channel.open(new InetSocketAddress(host,port));

通道(channel)并没有与之相关联的流，实际上，所拥有的read和write方法都是通过Buffer对象实现的。

如果不想处理缓冲区，可以使用Scanner从SocketChannel中读取信息。

Scanner有一个带ReadableByteChannel参数的构造器。

var in = new Scanner(channel,StandardCharsets.UTF-8);

通过调用静态方法Channels.newOutputStream，可以将通道转换成输出流。

OutputStream outStream = Channels.newOutputStream(channel);

对比中断套接字和阻塞套接字

package 第4章网络.interruptible;import java.awt.BorderLayout;
import java.awt.Dimension;
import java.awt.EventQueue;
import java.awt.HeadlessException;
import java.awt.Toolkit;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;import javax.swing.JButton;
import javax.swing.JFrame;
import javax.swing.JPanel;
import javax.swing.JScrollPane;
import javax.swing.JTextArea;public class InterruptibleSocketTest {public static void main(String[] args) {var frame = new InterruptibleSocketFrame();frame.setTitle("InterruptibleSocketTest");
//        frame.setSize(400, 400);//初始打开为屏幕中央Toolkit toolkit = Toolkit.getDefaultToolkit();Dimension screenSize = toolkit.getScreenSize();int screenWidth = (int) screenSize.getWidth();int screenHeight = (int) screenSize.getHeight();//计算窗口位置int x = (screenWidth - frame.getWidth())/2;int y = (screenHeight - frame.getHeight())/2;frame.setLocation(x, y);frame.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);frame.setVisible(true);}}
class InterruptibleSocketFrame extends JFrame {private Scanner in;private JButton interruptibleButton;private JButton blockingButton;private JButton cancelButton;private JTextArea messages;private TestServer server;private Thread connectThread;public InterruptibleSocketFrame() throws HeadlessException {var northPanel = new JPanel();add(northPanel,BorderLayout.NORTH);final int TEXT_ROWS = 20;final int TEXT_COLUMNS = 60;messages = new JTextArea(TEXT_ROWS, TEXT_COLUMNS);add(new JScrollPane(messages));interruptibleButton = new JButton("可中断");blockingButton = new JButton("阻塞中");northPanel.add(interruptibleButton);northPanel.add(blockingButton);interruptibleButton.addActionListener(e -> {interruptibleButton.setEnabled(false);blockingButton.setEnabled(false);cancelButton.setEnabled(true);connectThread = new Thread(() -> {try {connectInterruptibly();} catch (IOException e1) {messages.append("\nInterruptibleSocketTest.connectInterruptibly: "+e);}});connectThread.start();});blockingButton.addActionListener(e -> {interruptibleButton.setEnabled(false);blockingButton.setEnabled(false);cancelButton.setEnabled(true);connectThread = new Thread(() -> {try {connectBlocking();} catch (IOException e1) {messages.append("\nInterruptibleSocketTest.connectBlocking: "+e);}});connectThread.start();});cancelButton = new JButton("取消");cancelButton.setEnabled(false);northPanel.add(cancelButton);cancelButton.addActionListener(e -> {connectThread.interrupt();cancelButton.setEnabled(false);});server = new TestServer();new Thread(server).start();pack();}//连接测试服务器，用可中断IOpublic void connectInterruptibly() throws IOException {messages.append("可中断:\n");try (SocketChannel channel = SocketChannel.open(new InetSocketAddress("localhost",8189))){in = new Scanner(channel,StandardCharsets.UTF_8);while(!Thread.currentThread().isInterrupted()) {messages.append("读取 ");if(in.hasNextLine()) {String line = in.nextLine();messages.append(line);messages.append("\n");}}} finally {EventQueue.invokeLater(() -> {messages.append("管道关闭\n");interruptibleButton.setEnabled(true);blockingButton.setEnabled(true);});}}//连接测试服务器，用阻塞IOpublic void connectBlocking() throws UnknownHostException, IOException {messages.append("阻塞中:\n");try (var sock = new Socket("localhost",8189)){in = new Scanner(sock.getInputStream(),StandardCharsets.UTF_8);while(!Thread.currentThread().isInterrupted()) {messages.append("读取 ");if (in.hasNextLine()) {String line = in.nextLine();messages.append(line);messages.append("\n");}}} finally {EventQueue.invokeLater(() -> {messages.append("套接字关闭\n");interruptibleButton.setEnabled(true);blockingButton.setEnabled(true);});}}class TestServer implements Runnable {@Overridepublic void run() {try(var s = new ServerSocket(8189)){while (true) {Socket incoming = s.accept();Runnable r = new TestServerHandler(incoming);new Thread(r).start();}} catch (IOException e) {messages.append("\nTestServer.run: "+e);}}}class TestServerHandler implements Runnable{private Socket incoming;private int counter;public TestServerHandler(Socket incoming) {super();this.incoming = incoming;}@Overridepublic void run() {OutputStream outStream;try {try {outStream = incoming.getOutputStream();var out = new PrintWriter(new OutputStreamWriter(outStream, StandardCharsets.UTF_8),true);while(counter < 100) {counter++;if(counter<=10) out.println(counter);Thread.sleep(100);}}finally {incoming.close();messages.append("关闭服务\n");}} catch (Exception e) {messages.append("\nTestServerHandler.run:"+e);}}}
}