Java 网络编程：TCP/UDP 协议与 Socket 实战

3.2 客户端

package com.network.echo;

import java.io.*;
import java.net.Socket;

/**
 * 客户端套接字
 */
public class ClientSocketDemo {
    public static void main(String[] args) {
        try {
            // 创建客户端套接字，指定服务器地址和端口号
            Socket socket = new Socket("localhost", 6666);
            
            // 向服务器发送数据
            PrintStream ps = new PrintStream(new BufferedOutputStream(socket.getOutputStream()));
            ps.println("hello world, I am java");
            ps.flush();
            
            // 接收服务器返回的数据
            BufferedReader br = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            String info = br.readLine();
            System.out.println("我从服务器接收的数据是-->" + info);
            
            br.close();
            ps.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

四、服务端与多客户端通信

一个服务器处理多个客户端，服务端启动一次，客户端可以启动多次。

4.1 服务器端代码

主程序用来监听客户端请求，来一个客户端请求开一个线程。

package com.network.multi;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 主程序用来监听客户端请求，来一个客户端请求开一个线程
 */
public class MultiServerDemo {
    public static void main(String[] args) {
        // 线程池里放 3 个线程
        ExecutorService es = Executors.newFixedThreadPool(3);
        try {
            // 服务器端口
            ServerSocket server = new ServerSocket(6666);
            
            // 多个客户端来请求服务器，来一个创建一个线程去处理
            while (true) {
                Socket socket = server.accept();
                System.out.println("客户端连接成功：" + socket.getInetAddress().getHostAddress());
                // 连接成功就让线程去处理具体的任务
                es.execute(new ClientThread(socket));
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

// 处理客户端请求的线程，具体实现放到线程中处理
class ClientThread implements Runnable {
    private Socket socket;

    public ClientThread(Socket socket) {
        this.socket = socket;
    }

    /**
     * 功能：服务器接收客户端的内容，先打印出来再返回给客户端
     */
    @Override
    public void run() {
        try {
            // 创建输入输出流
            BufferedReader br = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            PrintStream ps = new PrintStream(new BufferedOutputStream(socket.getOutputStream()));
            
            // 读取客户端发送的数据
            String info = br.readLine();
            System.out.println(info);
            
            // 返回给客户端
            ps.println("echo:" + info);
            ps.flush();
            ps.close();
            br.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

4.2 客户端代码

package com.network.multi;

import java.io.*;
import java.net.Socket;
import java.util.Scanner;

/**
 * 客户端：就给服务器发消息，然后接收服务器返回的消息。
 * 可以同时启动多个客户端
 */
public class MultiClientDemo {
    public static void main(String[] args) {
        Scanner sc = new Scanner(System.in);
        try {
            Socket socket = new Socket("localhost", 6666);
            
            // 创建输入输出流
            PrintStream ps = new PrintStream(new BufferedOutputStream(socket.getOutputStream()));
            BufferedReader br = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            
            // 发送数据
            System.out.print("请输入内容：");
            String s = sc.nextLine();
            ps.println(s);
            ps.flush();
            
            // 接收数据
            String info = br.readLine();
            System.out.println(info);
            
            ps.close();
            br.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

五、多客户端之间的通信

5.1 概念

客户端之间的通信，需要通过服务器中转。 A 客户端发送消息到 B 客户端，需要先向服务器发消息，再由服务器发消息到 B 客户端。

发送的细节，需要一个数据包包含：我是谁、他是谁、消息、消息类型。

5.2 数据包类

package com.network.commutation;

import java.io.Serializable;

/**
 * 未来该对象要在网络上传输，因此需要序列化
 */
public class Message implements Serializable {
    private String from;
    private String to;
    private String info;
    private int type;

    public Message(String from, String to, String info, int type) {
        this.from = from;
        this.to = to;
        this.info = info;
        this.type = type;
    }

    public Message() {}

    public String getFrom() { return from; }
    public void setFrom(String from) { this.from = from; }
    public String getTo() { return to; }
    public void setTo(String to) { this.to = to; }
    public int getType() { return type; }
    public void setType(int type) { this.type = type; }
    public String getInfo() { return info; }
    public void setInfo(String info) { this.info = info; }

    @Override
    public String toString() {
        return "Message{" + "from='" + from + '\'' + ", to='" + to + '\'' + ", info='" + info + '\'' + ", type=" + type + '}';
    }
}

5.3 数据类型类

package com.network.commutation;

public final class MessageType {
    // 登录的消息类型
    public static final int LOGIN_TYPE = 0x1;
    // 发送的消息类型
    public static final int SEND_TYPE = 0x2;
}

5.4 服务器端及其内部的客户端处理线程

package com.network.commutation;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 服务器是一个中转站，里面要维护客户端线程信息
 */
public class Server {
    public static void main(String[] args) {
        // 多个客户端线程
        Vector<ClientThread> vector = new Vector<>();
        // 线程池，用来执行任务
        ExecutorService es = Executors.newFixedThreadPool(5);
        try {
            // 启动服务器
            ServerSocket server = new ServerSocket(8888);
            System.out.println("服务器已启动，等待客户端连接...");
            
            // 不停的接收客户端的连接
            while (true) {
                Socket socket = server.accept();
                // 客户端连接
                es.execute(new ClientThread(socket, vector));
                // 启动一个客户端处理线程
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

/**
 * 服务器中的客户端处理线程
 * 要考虑的是这个用户线程什么时候结束
 */
class ClientThread implements Runnable {
    private String name; // 客户端线程的名称 (唯一)
    private Socket socket; // 客户端的连接 socket
    private Vector<ClientThread> vector; // 服务器中的客户端处理线程集合
    private ObjectInputStream ois; // 对象输入流
    private ObjectOutputStream oos; // 对象输出流
    private boolean flag = true; // 线程运行标志

    public ClientThread(Socket socket, Vector<ClientThread> vector) {
        this.socket = socket;
        this.vector = vector;
        this.vector.add(this); // vector 集合中把当前线程对象添加进去
    }

    // 一个客户端连接进来，服务器就要开启一个客户端处理线程
    @Override
    public void run() {
        try {
            ois = new ObjectInputStream(socket.getInputStream());
            oos = new ObjectOutputStream(socket.getOutputStream());
            System.out.println("客户端 " + socket.getInetAddress().getHostAddress() + " 连接成功...");
            
            // 循环接收客户端发送的消息
            while (flag) {
                /*
                 * 登录消息类型
                 * 1. 客户端给服务器发消息
                 * 2. 服务器给客户端回复：欢迎您，xx
                 * 发送消息类型
                 * 1. 客户端给服务器发消息
                 * 2. 服务器将消息转发给目标客户端
                 */
                Message message = (Message) ois.readObject();
                int type = message.getType();
                
                switch (type) {
                    case MessageType.LOGIN_TYPE:
                        // 给客户端回个消息
                        name = message.getFrom();
                        message.setInfo("欢迎您，");
                        // message 对象本来就是客户端发过来的，这里加个内容再发回去
                        oos.writeObject(message);
                        break;
                    case MessageType.SEND_TYPE:
                        // 发给目标客户端对象
                        String to = message.getTo();
                        for (ClientThread ct : vector) {
                            // 从服务器的客户端处理线程集合中匹配目标线程名字、且不是发给自己
                            // 下面找到的那个 ct 对象，其实已经不是当前线程对象了，而是目标客户端对应的客户端线程对象
                            if (to.equals(ct.name) && ct != this) {
                                ct.oos.writeObject(message);
                                // 服务器中目标客户端处理线程会转发消息给目标客户端
                                break;
                            }
                        }
                        break;
                }
            }
            // 关闭流
            ois.close();
            oos.close();
        } catch (IOException | ClassNotFoundException e) {
            e.printStackTrace();
        }
    }
}

5.5 客户端及其内部的读取消息的线程

package com.network.commutation;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.Scanner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Client {
    public static void main(String[] args) {
        Scanner sc = new Scanner(System.in);
        ExecutorService es = Executors.newSingleThreadExecutor();
        try {
            // 连接服务器
            Socket socket = new Socket("localhost", 8888);
            System.out.println("服务器连接成功...");
            
            // 获取输入输出流
            ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
            ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
            
            // 告诉服务器我在登录
            System.out.print("请输入用户名：");
            String name = sc.nextLine();
            Message msg = new Message(name, null, null, MessageType.LOGIN_TYPE);
            oos.writeObject(msg);
            
            // 客户端给服务器写消息
            msg = (Message) ois.readObject();
            // 从服务器读取消息，这个 msg 就是服务器的消息
            System.out.println(msg.getInfo() + msg.getFrom());
            
            // 启动子读取线程：客户端专门读取其他客户端的消息
            es.execute(new ReadThread(ois));
            
            // 客户端主线程：给其他客户端发送消息
            boolean flag = true;
            while (flag) {
                msg = new Message();
                msg.setFrom(name);
                System.out.print("to: ");
                msg.setTo(sc.nextLine());
                System.out.print("info: ");
                msg.setInfo(sc.nextLine());
                msg.setType(MessageType.SEND_TYPE);
                oos.writeObject(msg);
                // 客户端给服务器写消息
            }
        } catch (IOException | ClassNotFoundException e) {
            e.printStackTrace();
        }
    }
}

// 子线程：客户端专门读取其他客户端的消息
class ReadThread implements Runnable {
    private ObjectInputStream ois; // 从服务器读取消息的输入流（因为需要借助服务器转发过来）
    private Boolean flag = true; // 标记线程是否运行

    public ReadThread(ObjectInputStream ois) {
        this.ois = ois;
    }

    @Override
    public void run() {
        try {
            // 不停的接收消息
            while (flag) {
                Message msg = (Message) ois.readObject();
                // 别人发过来的消息（由服务器转发过来的）
                System.out.println("[" + msg.getFrom() + "] 对我说：" + msg.getInfo());
            }
            if (ois != null) {
                ois.close();
            }
        } catch (IOException | ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }
}

5.6 代码原理分析

实现过程：多个客户端通过一台服务器进行通信。

客户端 A、客户端 B 都要连接服务器。一旦连接上服务器后，服务器中会产生各个客户端对应的客户端处理线程，比如客户端 A 处理线程、客户端 B 处理线程，当然服务器中会用一个集合来存储这些客户端处理线程。

我们做两件事：客户端连接服务器、客户端之间发消息。

5.6.1 客户端连接服务器

场景：客户端连接上服务器后，会在客户端上要求输入一个用户名，这个名字以后就代表客户端的唯一名字，然后客户端会将名字发送给服务器，服务器收到消息后会返回"欢迎你，xxx"给客户端，客户端拿到消息后会打印到控制台上。

5.6.2 客户端之间通信

场景：客户端都成功连上服务器后，客户端 A 给客户端 B 发送一个消息。客户端 A 会将消息先转发到服务器上，然后服务器再转发到客户端 B 上。

我们好奇这个转发的过程是如何实现的。 用到上面的原理，一个客户端连接上服务器后，服务器中会产生客户端对应的客户端处理线程，现在我们的客户端 A 和客户端 B 都连接上服务器了，那么在服务器中会创建客户端处理线程 A、客户端处理线程 B。 这个时候，我们可以理解，客户端处理线程 A 就代替了客户端 A 的工作，客户端处理线程 B 就代替了 B 的工作，虽然客户端 A 和客户端 B 无法直接通信，但是客户端处理线程 A 和客户端处理线程 B 都在服务器中，所以可以通信。

当然，我们的客户端中不能如此简单，人家服务器给我发消息，我也得读取。一般读取的操作，单独写一个读取线程，不停的读取； 当然，客户端也可以不停的给其他人发消息，可以直接在客户端的主线程中操作。

六、网络编程 UDP 协议

要求接收者先等着。 在网络上以任意可能的路径传到目的地，但是到达目的地的时间、内容的正确性无法保证，每个传输的数据报必须在 64kb 之间。

使用 2 个类：

• DatagramPacket: 数据报包类
• DatagramSocket: 接收和发送数据报的套接字

先启动客户端，再启动服务器。

客户端

package com.network.udp;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;

/**
 * 客户端，用来接收服务器发来的内容
 */
public class UDPClient {
    public static void main(String[] args) {
        // 封装数据报，用来接收内容
        byte[] bytes = new byte[1024];
        DatagramPacket dp = new DatagramPacket(bytes, bytes.length);
        try {
            // 客户端接收数据报的套接字
            DatagramSocket socket = new DatagramSocket(8000);
            System.out.println("客户端正在接收...");
            socket.receive(dp);
            // 接收服务器发来的数据，存储到数据报中
            // 读取数据报内容
            String info = new String(dp.getData(), 0, dp.getLength());
            System.out.println(info);
            socket.close();
        } catch (SocketException e) {
            e.printStackTrace();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

服务器

package com.network.udp;

import java.io.IOException;
import java.net.*;

/**
 * 服务器发给客户端
 */
public class UDPServer {
    public static void main(String[] args) {
        String s = "good good study, 天天 up";
        byte[] bytes = s.getBytes();
        try {
            // 封装数据报，要发给客户端的
            DatagramPacket dp = new DatagramPacket(bytes, 0, bytes.length, InetAddress.getByName("localhost"), 8000);
            // 创建服务器的数据报套接字，用于接收和发送 UDP 数据包
            DatagramSocket ds = new DatagramSocket(9000);
            ds.send(dp);
        } catch (UnknownHostException | SocketException e) {
            e.printStackTrace();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

UDP 记得先启动客户端，等待服务器发送消息。

七、URL

网络上，一个 URL 就是一个资源。 抽象类 URLConnection 是所有类的超类，表示程序、URL 之间的通信连接。

package url;

import java.io.*;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;

public class URLDemo {
    public static void main(String[] args) {
        // URL 统一资源定位符
        try {
            URL url = new URL("https://p1.meituan.net/biztone/1730f0baefe95984e632a52590f0b446393917.jpg");
            // 抽象父类 UrlConnection
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            // 输入输出流
            BufferedInputStream bis = new BufferedInputStream(conn.getInputStream());
            BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream("/Users/mac/D/hello.png"));
            
            // 读取到数组中
            byte[] bytes = new byte[1024];
            int len = -1;
            while ((len = bis.read(bytes)) != -1) {
                bos.write(bytes, 0, len);
            }
            System.out.println("下载成功！");
            bis.close();
            bos.close();
        } catch (MalformedURLException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

八、MINA 框架

基于 TCP/IP 的 Java 框架。

8.1 最小核心依赖包

<!-- MINA 核心 -->
<dependency>
    <groupId>org.apache.mina</groupId>
    <artifactId>mina-core</artifactId>
    <version>2.1.3</version>
</dependency>
<!-- SLF4J 简单实现 -->
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-simple</artifactId>
    <version>1.7.35</version>
</dependency>

8.2 服务器端

一个启动类 + 一个消息处理器类。

package com.mina;

import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import java.io.IOException;
import java.net.InetSocketAddress;

public class Server {
    public static void main(String[] args) {
        // 非阻塞的 server 端 socket。创建数据接收过滤器
        NioSocketAcceptor acceptor = new NioSocketAcceptor();
        DefaultIoFilterChainBuilder chain = acceptor.getFilterChain();
        // 添加过滤器链
        chain.addLast("myChain", new ProtocolCodecFilter(new TextLineCodecFactory()));
        
        // ★★★ 服务端消息处理器：需要实现 IoHandler 接口的处理器类
        acceptor.setHandler(new MinaServerHandler());
        try {
            // 绑定端口 + 启动服务器
            int port = 9999;
            acceptor.bind(new InetSocketAddress(port));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        System.out.println("Mina Server is started...");
    }
}

package com.mina;

import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;

/**
 * 服务器端消息处理器
 */
public class MinaServerHandler extends IoHandlerAdapter {
    // 开启会话
    @Override
    public void sessionOpened(IoSession session) throws Exception {
        super.sessionOpened(session);
        System.out.println("欢迎你，" + session.getRemoteAddress());
    }

    // 关闭会话
    @Override
    public void sessionClosed(IoSession session) throws Exception {
        super.sessionClosed(session);
        System.out.println("客户端退出了....");
    }

    // 接收消息
    @Override
    public void messageReceived(IoSession session, Object message) throws Exception {
        super.messageReceived(session, message);
        String msg = (String) message;
        // 服务器要给客户端回复消息
        session.write("echo:" + msg);
    }
}

8.3 客户端

一个启动类 + 一个消息处理器。

package com.mina;

import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import java.net.InetSocketAddress;
import java.util.Scanner;

public class Client {
    public static void main(String[] args) {
        // 创建客户端连接
        NioSocketConnector connector = new NioSocketConnector();
        // 过滤器链
        DefaultIoFilterChainBuilder chain = connector.getFilterChain();
        // 添加过滤器链
        chain.addLast("myChain", new ProtocolCodecFilter(new TextLineCodecFactory()));
        
        // ★★★ 客户端消息处理器：需要实现 IoHandler 接口的处理器类
        connector.setHandler(new MinaClientHandler());
        connector.setConnectTimeoutCheckInterval(10000);
        
        // 连接服务器
        ConnectFuture cf = connector.connect(new InetSocketAddress("localhost", 9999));
        cf.awaitUninterruptibly(); // 等待连接成功
        
        Scanner sc = new Scanner(System.in);
        while (true) {
            System.out.print("请输入：");
            String info = sc.nextLine();
            cf.getSession().write(info); // 写给服务器端
        }
    }
}

package com.mina;

import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;

public class MinaClientHandler extends IoHandlerAdapter {
    @Override
    public void sessionOpened(IoSession session) throws Exception {
        super.sessionOpened(session);
        System.out.println("sessionOpened....");
    }

    @Override
    public void sessionClosed(IoSession session) throws Exception {
        super.sessionClosed(session);
        System.out.println("sessionClosed....");
    }

    // 客户端收到消息直接打印
    @Override
    public void messageReceived(IoSession session, Object message) throws Exception {
        super.messageReceived(session, message);
        String info = (String) message;
        System.out.println(info);
    }
}

8.4 结果

（此处省略运行截图，实际运行时请参照上述代码配置）

8.5 传输对象类型

new TextLineCodecFactory() 改成 new 某对象类，就可以实现自定义对象传输。