SpringBoot+Netty开发IM即时通讯系列（二）

<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">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.pubing</groupId>
  <artifactId>helloNetty</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <dependencies>
    <dependency>
      <groupId>io.netty</groupId>
      <artifactId>netty-all</artifactId>
      <version>4.1.25.Final</version>
    </dependency>
  </dependencies>
</project>

package com.phubing.websokect;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
/**
 * 用于和客户端进行连接
 * 
 * @author phubing
 *
 */
public class WebSocketServer {
  public static void main(String[] args) throws InterruptedException {
    //定义线程组
    EventLoopGroup mainGroup =  new NioEventLoopGroup();
    EventLoopGroup subGroup =  new NioEventLoopGroup();
    try {
    ServerBootstrap server = new ServerBootstrap();
    server.group(mainGroup, subGroup)
    //channel类型
    .channel(NioServerSocketChannel.class)
    //针对subGroup做的子处理器，childHandler针对WebSokect的初始化器
    .childHandler(new WebSocketinitializer());
    //绑定端口并以同步方式进行使用
    ChannelFuture channelFuture = server.bind(10086).sync();
    //针对channelFuture，进行相应的监听
    channelFuture.channel().closeFuture().sync();
    }finally {
      //针对两个group进行优雅地关闭
      mainGroup.shutdownGracefully();
      subGroup.shutdownGracefully();
    }
  }
}

package com.phubing.websokect;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
public class WebSocketinitializer extends ChannelInitializer<SocketChannel>{
  @Override
  protected void initChannel(SocketChannel socketChannel) throws Exception {
    //从Channel中获取对应的pipeline
    ChannelPipeline channelPipeline = socketChannel.pipeline();
    //添加相应的助手类与处理器
    /**
     * WebSokect基于Http，所以要有相应的Http编解码器，HttpServerCodec()
     */
    channelPipeline.addLast(new HttpServerCodec());
    //在Http中有一些数据流的传输，那么数据流有大有小，如果说有一些相应的大数据流处理的话，需要在此添加
    //ChunkedWriteHandler：为一些大数据流添加支持
    channelPipeline.addLast(new ChunkedWriteHandler());
    //UdineHttpMessage进行处理，也就是会用到request以及response
    //HttpObjectAggregator：聚合器，聚合了FullHTTPRequest、FullHTTPResponse。。。，当你不想去管一些HttpMessage的时候，直接把这个handler丢到管道中，让Netty自行处理即可
    channelPipeline.addLast(new HttpObjectAggregator(2048*64));
    //================华丽的分割线：以上是用于支持Http协议================
    //================华丽的分割线：以下是用于支持WebSoket==================
    // /ws：一开始建立连接的时候会使用到，可自定义
    //WebSocketServerProtocolHandler：给客户端指定访问的路由（/ws），是服务器端处理的协议，当前的处理器处理一些繁重的复杂的东西，运行在一个WebSocket服务端
    //另外也会管理一些握手的动作：handshaking(close，ping，pong) ping + pong = 心跳，对于WebSocket来讲，是以frames进行传输的，不同的数据类型对应的frames也不同
    channelPipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
    //添加自动handler，读取客户端消息并进行处理，处理完毕之后将相应信息传输给对应客户端
    channelPipeline.addLast(new ChatHandler());
  }
}

package com.phubing.websokect;
import java.time.LocalDate;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;
//TextWebSocketFrame：处理消息的handler，在Netty中用于处理文本的对象，frames是消息的载体
public class ChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame>{
  //用于记录和管理所有客户端的channel，可以把相应的channel保存到一整个组中
  //DefaultChannelGroup：用于对应ChannelGroup，进行初始化
  private static ChannelGroup channelClient =  new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  @Override
  protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
    //text()获取从客户端发送过来的字符串
    String content = msg.text();
    System.out.println("客户端传输的数据："+content);
    //针对channel进行发送，客户端对应的是channel
    /**
     * 方式一
     */
    for (Channel channel : channelClient) {
      //循环对每一个channel对应输出即可（往缓冲区中写，写完之后再刷到客户端）
      //注：writeAndFlush不可以使用String，因为传输的载体是一个TextWebSocketFrame，需要把消息通过载体再刷到客户端
      channel.writeAndFlush(new TextWebSocketFrame("【服务器于 " + LocalDate.now() + "接收到消息：】 ，消息内容为：" +content));
    }   
    /**
     * 方式二
    channelClient.writeAndFlush(new TextWebSocketFrame("【服务器于 " + LocalDate.now() + "接收到消息：】 ，消息内容为：" +content))
     */
  }
  //当客户端连接服务端（或者是打开连接之后）
  @Override
  public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    //获取客户端所对应的channel，添加到一个管理的容器中即可
    channelClient.add(ctx.channel());
  }
  //客户端断开
  @Override
  public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
    //实际上是多余的，只要handler被移除，client会自动的把对应的channel移除掉
    channelClient.remove(ctx.channel());
    //每一而channel都会有一个长ID与短ID
    //一开始channel就有了，系统会自动分配一串很长的字符串作为唯一的ID，如果使用asLongText()获取的ID是唯一的，asShortText()会把当前ID进行精简，精简过后可能会有重复
    System.out.println("channel的长ID："+ctx.channel().id().asLongText());
    System.out.println("channel的短ID："+ctx.channel().id().asShortText());
  }
}

<!DOCTYPE html>
<html>
  <head>
    <meta charset="utf-8" />
    <title>Netty+WebSocket案例</title>
  </head>
  <body>
    <div id="">发送消息：</div><br>
    <input type="text" name="messageContent" id="messageContent"/>
    <input type="button" name="" id="" value="发送" onclick="CHAT.chat()"/>
    <hr>
    <div id="">接收消息：</div><br>
    <div id="receiveNsg" style="background-color: gainsboro;"></div>
    <script type="text/javascript">
      window.CHAT = {
        socket: null,
        //初始化
        init: function(){
          //首先判断浏览器是否支持WebSocket
          if (window.WebSocket){
            CHAT.socket = new WebSocket("ws://localhost:10086/ws");
            CHAT.socket.onopen = function(){
              console.log("客户端与服务端建立连接成功");
            },
            CHAT.socket.onmessage = function(e){
              console.log("接收到消息："+e.data);
              var receiveNsg = window.document.getElementById("receiveNsg");
              var html = receiveNsg.innerHTML;
              receiveNsg.innerHTML = html + "<br>" + e.data; 
            },
            CHAT.socket.onerror = function(){
              console.log("发生错误");
            },
            CHAT.socket.onclose = function(){
              console.log("客户端与服务端关闭连接成功");
            }           
          }else{
            alert("8102年都过了，升级下浏览器吧");
          }
        },
        chat: function(){
          var msg = window.document.getElementById("messageContent");
          CHAT.socket.send(msg.value);
        }
      }
      CHAT.init();
    </script>
  </body>
</html>