<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    
    <!-- pom模型版本 -->
    <modelVersion>4.0.0</modelVersion>
    
    <!-- 項目信息 -->
    <groupId>demo</groupId><!-- 項目唯一標(biāo)識 -->
    <artifactId>springboot</artifactId><!-- 項目名 -->
    <version>0.0.1-SNAPSHOT</version><!-- 版本 -->
    <packaging>jar</packaging><!-- 打包方式 （pom,war,jar） -->
 
    <name>springboot</name><!-- 項目的名稱， Maven 產(chǎn)生的文檔用 -->
    <description>Demo project for Spring Boot</description><!-- 項目的描述, Maven 產(chǎn)生的文檔用 -->
 
    <!-- 父級項目 -->
	<parent> 
        <artifactId>spring-boot-starter-parent</artifactId>  <!-- 被繼承的父項目的構(gòu)件標(biāo)識符 --> 
        <groupId>org.springframework.boot</groupId>  <!-- 被繼承的父項目的全球唯一標(biāo)識符 -->
        <version>1.5.7.RELEASE</version>  <!-- 被繼承的父項目的版本 --> 
        <!-- 父項目的pom.xml文件的相對路徑。相對路徑允許你選擇一個不同的路徑。默認(rèn)值是../pom.xml。
             Maven首先在構(gòu)建當(dāng)前項目的地方尋找父項目的pom，其次在文件系統(tǒng)的這個位置（relativePath位置），
             然后在本地倉庫，最后在遠(yuǎn)程倉庫尋找父項目的pom。 --> 
        <relativePath/> <!-- lookup parent from repository -->
    </parent> 
	
	<!-- 模塊（有時稱作子項目） 被構(gòu)建成項目的一部分。列出的每個模塊元素是指向該模塊的目錄的相對路徑 --> 
    <modules> 
        <!-- 子項目相對路徑 --> 
        <module></module> 
    </modules> 
	
    <!-- 屬性設(shè)置 -->
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><!-- 編譯字符編碼為utf-8 -->
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><!-- 輸出字符編碼為UTF-8  -->
        <java.version>1.8</java.version><!-- jdK版本 -->
    </properties>
    
    <!-- 依賴關(guān)系 -->
    <dependencies>
        <!-- 測試 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- mysql(數(shù)據(jù)庫) -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <scope>runtime</scope>
        </dependency>
    </dependencies>
    <!-- 編譯 -->
    <build>
        <!-- 插件 -->
        <plugins>
            <!-- maven插件 -->
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

<dependencies>
    <dependency>
        <groupid>org.springframework.boot</groupid>
        <artifactid>spring-boot-starter-web</artifactid>
        <exclusions>
            <!-- 去除Tomcat容器 -->
            <exclusion>
                <groupid>org.springframework.boot</groupid>
                <artifactid>spring-boot-starter-tomcat</artifactid>
            </exclusion>
        </exclusions>
    </dependency>
    <!-- 增加Jetty容器 -->
    <dependency>
        <groupid>org.springframework.boot</groupid>
        <artifactid>spring-boot-starter-jetty</artifactid>
    </dependency>
</dependencies>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
    <exclusions>
      <!-- 去除 Tomcat 容器 -->
        <exclusion>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-tomcat</artifactId>
        </exclusion>
    </exclusions>
</dependency>
  <!-- 添加 Undertow 容器 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-undertow</artifactId>
</dependency>

# Undertow 日志存放目錄
server.undertow.accesslog.dir=
# 是否啟動日志
server.undertow.accesslog.enabled=false
# 日志格式
server.undertow.accesslog.pattern=common
# 日志文件名前綴
server.undertow.accesslog.prefix=access_log
# 日志文件名后綴
server.undertow.accesslog.suffix=log
# HTTP POST請求最大的大小
server.undertow.max-http-post-size=0
# 設(shè)置IO線程數(shù), 它主要執(zhí)行非阻塞的任務(wù),它們會負(fù)責(zé)多個連接, 默認(rèn)設(shè)置每個CPU核心一個線程
# 不要設(shè)置過大，如果過大，啟動項目會報錯：打開文件數(shù)過多
server.undertow.io-threads=12
# 阻塞任務(wù)線程池, 當(dāng)執(zhí)行類似servlet請求阻塞IO操作, undertow會從這個線程池中取得線程
# 它的值設(shè)置取決于系統(tǒng)線程執(zhí)行任務(wù)的阻塞系數(shù)，默認(rèn)值是IO線程數(shù)*8
server.undertow.worker-threads=20
# 以下的配置會影響buffer,這些buffer會用于服務(wù)器連接的IO操作,有點類似netty的池化內(nèi)存管理
# 每塊buffer的空間大小,越小的空間被利用越充分，不要設(shè)置太大，以免影響其他應(yīng)用，合適即可
server.undertow.buffer-size=1024
# 每個區(qū)分配的buffer數(shù)量 , 所以pool的大小是buffer-size * buffers-per-region
server.undertow.buffers-per-region=1024
# 是否分配的直接內(nèi)存
server.undertow.direct-buffers=true

<!-- netty -->
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.36.Final</version>
</dependency>

# netty 配置
netty:
  # boss線程數(shù)量
  boss: 4
  # worker線程數(shù)量
  worker: 2
  # 連接超時時間
  timeout: 6000
  # 服務(wù)器主端口
  port: 17000
  # 服務(wù)器備用端口
  portSalve: 18026
  # 服務(wù)器地址
  host: 127.0.0.1

/**
 * Socket攔截器，用于處理客戶端的行為
 **/
@Slf4j
public class SocketHandler extends ChannelInboundHandlerAdapter {
    public static final ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * 讀取到客戶端發(fā)來的消息
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 由于我們配置的是 字節(jié)數(shù)組 編解碼器，所以這里取到的用戶發(fā)來的數(shù)據(jù)是 byte數(shù)組
        byte[] data = (byte[]) msg;
        log.info("收到消息: " + new String(data));
        // 給其他人轉(zhuǎn)發(fā)消息
        for (Channel client : clients) {
            if (!client.equals(ctx.channel())) {
                client.writeAndFlush(data);
            }
        }
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        log.info("新的客戶端鏈接：" + ctx.channel().id().asShortText());
        clients.add(ctx.channel());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        clients.remove(ctx.channel());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.channel().close();
        clients.remove(ctx.channel());
    }
}

/**
 * Socket 初始化器，每一個Channel進(jìn)來都會調(diào)用這里的 InitChannel 方法
 **/
@Component
public class SocketInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        // 添加對byte數(shù)組的編解碼，netty提供了很多編解碼器，你們可以根據(jù)需要選擇
        pipeline.addLast(new ByteArrayDecoder());
        pipeline.addLast(new ByteArrayEncoder());
        // 添加上自己的處理器
        pipeline.addLast(new SocketHandler());
    }
}

@Slf4j
@Component
public class SocketServer {
    @Resource
    private SocketInitializer socketInitializer;

    @Getter
    private ServerBootstrap serverBootstrap;

    /**
     * netty服務(wù)監(jiān)聽端口
     */
    @Value("${netty.port:17000}")
    private int port;
    /**
     * 主線程組數(shù)量
     */
    @Value("${netty.boss:4}")
    private int bossThread;

    /**
     * 啟動netty服務(wù)器
     */
    public void start() {
        this.init();
        this.serverBootstrap.bind(this.port);
        log.info("Netty started on port: {} (TCP) with boss thread {}", this.port, this.bossThread);
    }

    /**
     * 初始化netty配置
     */
    private void init() {
        // 創(chuàng)建兩個線程組，bossGroup為接收請求的線程組，一般1-2個就行
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(this.bossThread);
        // 實際工作的線程組
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        this.serverBootstrap = new ServerBootstrap();
        this.serverBootstrap.group(bossGroup, workerGroup) // 兩個線程組加入進(jìn)來
                .channel(NioServerSocketChannel.class)  // 配置為nio類型
                .childHandler(this.socketInitializer); // 加入自己的初始化器
    }
}

/**
 * 監(jiān)聽Spring容器啟動完成，完成后啟動Netty服務(wù)器
 **/
@Component
public class NettyStartListener implements ApplicationRunner {
    @Resource
    private SocketServer socketServer;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        this.socketServer.start();
    }
}

public class ClientThread implements Runnable{

    private final Selector selector;

    public ClientThread(Selector selector) {
        this.selector = selector;
    }

    @Override
    public void run() {
        try {
            for (; ; ) {
                int channels = selector.select();
                if (channels == 0) {
                    continue;
                }
                Set<SelectionKey> selectionKeySet = selector.selectedKeys();
                Iterator<SelectionKey> keyIterator = selectionKeySet.iterator();
                while (keyIterator.hasNext()) {
                    SelectionKey selectionKey = keyIterator.next();

                    // 移除集合當(dāng)前得selectionKey，避免重復(fù)處理
                    keyIterator.remove();
                    if (selectionKey.isReadable()) {
                        this.handleRead(selector, selectionKey);
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 處理可讀狀態(tài)
    private void handleRead(Selector selector, SelectionKey selectionKey) throws IOException {
        SocketChannel channel = (SocketChannel) selectionKey.channel();
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        StringBuilder message = new StringBuilder();
        if (channel.read(byteBuffer) > 0) {
            byteBuffer.flip();
            message.append(StandardCharsets.UTF_8.decode(byteBuffer));
        }
        // 再次注冊到選擇器上，繼續(xù)監(jiān)聽可讀狀態(tài)
        channel.register(selector, SelectionKey.OP_READ);
        System.out.println(message);
    }
}

public class ChatClient {

    public void start(String name) throws IOException {
        SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 8088));
        socketChannel.configureBlocking(false);
        Selector selector = Selector.open();
        socketChannel.register(selector, SelectionKey.OP_READ);

        // 監(jiān)聽服務(wù)端發(fā)來得消息
        new Thread(new ClientThread(selector)).start();
        // 監(jiān)聽用戶輸入
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNextLine()) {
            String message = scanner.nextLine();
            if (StringUtils.hasText(message)) {
                socketChannel.write(StandardCharsets.UTF_8.encode(name + ": " + message));
            }
        }
    }
}

public class Client1 {
    public static void main(String[] args) throws IOException {
        new ChatClient().start("李四");
    }
}
public class Client2 {
    public static void main(String[] args) throws IOException {
        new ChatClient().start("張三");
    }
}