import java.util.Objects;

public class User {

? ? public String id;
? ? public String nickname;

? ? public User(String id, String nickname) {
? ? ? ? super();
? ? ? ? this.id = id;
? ? ? ? this.nickname = nickname;
? ? }

? ? public String getId() {
? ? ? ? return id;
? ? }

? ? public void setId(String id) {
? ? ? ? this.id = id;
? ? }

? ? public String getNickname() {
? ? ? ? return nickname;
? ? }

? ? public void setNickname(String nickname) {
? ? ? ? this.nickname = nickname;
? ? }

? ? @Override
? ? public boolean equals(Object o) {
? ? ? ? if (this == o)
? ? ? ? ? ? return true;
? ? ? ? if (o == null || getClass() != o.getClass())
? ? ? ? ? ? return false;
? ? ? ? User user = (User) o;
? ? ? ? return id.equals(user.getId());
? ? }

? ? @Override
? ? public int hashCode() {

? ? ? ? return Objects.hash(id);
? ? }

? ? public String getUid() {

? ? ? ? return id;
? ? }
}

2.SocketSession類

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.AttributeKey;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

public class SocketSession {

? ? public static final AttributeKey<SocketSession> SESSION_KEY = AttributeKey.valueOf("SESSION_KEY");

? ? /**
? ? ?* 用戶實(shí)現(xiàn)服務(wù)端會(huì)話管理的核心
? ? ?*/
// 通道
? ? private Channel channel;
? ? // 用戶
? ? private User user;

? ? // session唯一標(biāo)示
? ? private final String sessionId;

? ? private String group;

? ? /**
? ? ?* session中存儲(chǔ)的session 變量屬性值
? ? ?*/
? ? private Map<String, Object> map = new HashMap<String, Object>();

? ? public SocketSession(Channel channel) {//注意傳入?yún)?shù)channel。不同客戶端會(huì)有不同channel
? ? ? ? this.channel = channel;
? ? ? ? this.sessionId = buildNewSessionId();
? ? ? ? channel.attr(SocketSession.SESSION_KEY).set(this);
? ? }

? ? // 反向?qū)Ш?
? ? public static SocketSession getSession(ChannelHandlerContext ctx) {//注意ctx，不同的客戶端會(huì)有不同ctx
? ? ? ? Channel channel = ctx.channel();
? ? ? ? return channel.attr(SocketSession.SESSION_KEY).get();
? ? }

? ? // 反向?qū)Ш?
? ? public static SocketSession getSession(Channel channel) {
? ? ? ? return channel.attr(SocketSession.SESSION_KEY).get();
? ? }

? ? public String getId() {
? ? ? ? return sessionId;
? ? }

? ? private static String buildNewSessionId() {
? ? ? ? String uuid = UUID.randomUUID().toString();
? ? ? ? return uuid.replaceAll("-", "");
? ? }

? ? public synchronized void set(String key, Object value) {
? ? ? ? map.put(key, value);
? ? }

? ? public synchronized <T> T get(String key) {
? ? ? ? return (T) map.get(key);
? ? }

? ? public boolean isValid() {
? ? ? ? return getUser() != null ? true : false;
? ? }

? ? public User getUser() {
? ? ? ? return user;
? ? }

? ? public void setUser(User user) {
? ? ? ? this.user = user;
? ? }

? ? public String getGroup() {
? ? ? ? return group;
? ? }

? ? public void setGroup(String group) {
? ? ? ? this.group = group;
? ? }

? ? public Channel getChannel() {
? ? ? ? return channel;
? ? }
}

3.SessionGroup

import com.google.gson.Gson;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.ChannelGroupFuture;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.ImmediateEventExecutor;
import org.springframework.util.StringUtils;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public final class SessionGroup {

? ? private static SessionGroup singleInstance = new SessionGroup();

? ? // 組的映射
? ? private ConcurrentHashMap<String, ChannelGroup> groupMap = new ConcurrentHashMap<>();

? ? public static SessionGroup inst() {
? ? ? ? return singleInstance;
? ? }

? ? public void shutdownGracefully() {

? ? ? ? Iterator<ChannelGroup> groupIterator = groupMap.values().iterator();
? ? ? ? while (groupIterator.hasNext()) {
? ? ? ? ? ? ChannelGroup group = groupIterator.next();
? ? ? ? ? ? group.close();
? ? ? ? }
? ? }

? ? public void sendToOthers(Map<String, String> result, SocketSession s) {
? ? ? ? // 獲取組
? ? ? ? ChannelGroup group = groupMap.get(s.getGroup());
? ? ? ? if (null == group) {
? ? ? ? ? ? return;
? ? ? ? }
? ? ? ? Gson gson=new Gson();
? ? ? ? String json = gson.toJson(result);
? ? ? ? // 自己發(fā)送的消息不返回給自己
// ? ? ?Channel channel = s.getChannel();
? ? ? ? // 從組中移除通道
// ? ? ?group.remove(channel);
? ? ? ? ChannelGroupFuture future = group.writeAndFlush(new TextWebSocketFrame(json));
? ? ? ? future.addListener(f -> {
? ? ? ? ? ? System.out.println("完成發(fā)送："+json);
// ? ? ? ? ?group.add(channel);//發(fā)送消息完畢重新添加。

? ? ? ? });
? ? }

? ? public void addSession(SocketSession session) {

? ? ? ? String groupName = session.getGroup();
? ? ? ? if (StringUtils.isEmpty(groupName)) {
? ? ? ? ? ? // 組為空，直接返回
? ? ? ? ? ? return;
? ? ? ? }
? ? ? ? ChannelGroup group = groupMap.get(groupName);
? ? ? ? if (null == group) {
? ? ? ? ? ? group = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
? ? ? ? ? ? groupMap.put(groupName, group);
? ? ? ? }
? ? ? ? group.add(session.getChannel());
? ? }

? ? /**
? ? ?* 關(guān)閉連接， 關(guān)閉前發(fā)送一條通知消息
? ? ?*/
? ? public void closeSession(SocketSession session, String echo) {
? ? ? ? ChannelFuture sendFuture = session.getChannel().writeAndFlush(new TextWebSocketFrame(echo));
? ? ? ? sendFuture.addListener(new ChannelFutureListener() {
? ? ? ? ? ? public void operationComplete(ChannelFuture future) {
? ? ? ? ? ? ? ? System.out.println("關(guān)閉連接："+echo);
? ? ? ? ? ? ? ? future.channel().close();
? ? ? ? ? ? }
? ? ? ? });
? ? }

? ? /**
? ? ?* 關(guān)閉連接
? ? ?*/
? ? public void closeSession(SocketSession session) {

? ? ? ? ChannelFuture sendFuture = session.getChannel().close();
? ? ? ? sendFuture.addListener(new ChannelFutureListener() {
? ? ? ? ? ? public void operationComplete(ChannelFuture future) {
? ? ? ? ? ? ? ? System.out.println("發(fā)送所有完成："+session.getUser().getNickname());
? ? ? ? ? ? }
? ? ? ? });

? ? }

? ? /**
? ? ?* 發(fā)送消息
? ? ?* @param ctx 上下文
? ? ?* @param msg 待發(fā)送的消息
? ? ?*/
? ? public void sendMsg(ChannelHandlerContext ctx, String msg) {
? ? ? ? ChannelFuture sendFuture = ctx.writeAndFlush(new TextWebSocketFrame(msg));
? ? ? ? sendFuture.addListener(f -> {//發(fā)送監(jiān)聽
? ? ? ? ? ? System.out.println("對(duì)所有發(fā)送完成："+msg);
? ? ? ? });
? ? }
}

4.WebSocketTextHandler類

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

import java.util.HashMap;
import java.util.Map;

public class WebSocketTextHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

? ? @Override
? ? protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
? ? ? ? SocketSession session = SocketSession.getSession(ctx);
? ? ? ? TypeToken<HashMap<String, String>> typeToken = new TypeToken<HashMap<String, String>>() {
? ? ? ? };

? ? ? ? Gson gson=new Gson();
? ? ? ? java.util.Map<String,String> map = gson.fromJson(msg.text(), typeToken.getType());
? ? ? ? User user = null;
? ? ? ? switch (map.get("type")) {
? ? ? ? ? ? case "msg":
? ? ? ? ? ? ? ? Map<String, String> result = new HashMap<>();
? ? ? ? ? ? ? ? user = session.getUser();
? ? ? ? ? ? ? ? result.put("type", "msg");
? ? ? ? ? ? ? ? result.put("msg", map.get("msg"));
? ? ? ? ? ? ? ? result.put("sendUser", user.getNickname());
? ? ? ? ? ? ? ? SessionGroup.inst().sendToOthers(result, session);
? ? ? ? ? ? ? ? break;
? ? ? ? ? ? case "init":
? ? ? ? ? ? ? ? String room = map.get("room");
? ? ? ? ? ? ? ? session.setGroup(room);
? ? ? ? ? ? ? ? String nick = map.get("nick");
? ? ? ? ? ? ? ? user = new User(session.getId(), nick);
? ? ? ? ? ? ? ? session.setUser(user);
? ? ? ? ? ? ? ? SessionGroup.inst().addSession(session);
? ? ? ? ? ? ? ? break;
? ? ? ? }
? ? }

? ? @Override
? ? public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

? ? ? ? // 是否握手成功，升級(jí)為 Websocket 協(xié)議
? ? ? ? if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
? ? ? ? ? ? // 握手成功，移除 HttpRequestHandler，因此將不會(huì)接收到任何消息
? ? ? ? ? ? // 并把握手成功的 Channel 加入到 ChannelGroup 中
? ? ? ? ? ? new SocketSession(ctx.channel());
? ? ? ? } else if (evt instanceof IdleStateEvent) {
? ? ? ? ? ? IdleStateEvent stateEvent = (IdleStateEvent) evt;
? ? ? ? ? ? if (stateEvent.state() == IdleState.READER_IDLE) {
? ? ? ? ? ? ? ? System.out.println("bb22");
? ? ? ? ? ? }
? ? ? ? } else {
? ? ? ? ? ? super.userEventTriggered(ctx, evt);
? ? ? ? }
? ? }
}

5.WebSocketServer類

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
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.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

public class WebSocketServer {

? ? private static WebSocketServer wbss;

? ? private static final int READ_IDLE_TIME_OUT = 60; // 讀超時(shí)
? ? private static final int WRITE_IDLE_TIME_OUT = 0;// 寫超時(shí)
? ? private static final int ALL_IDLE_TIME_OUT = 0; // 所有超時(shí)

? ? public static WebSocketServer inst() {
? ? ? ? return wbss = new WebSocketServer();
? ? }

? ? public void run(int port) {
? ? ? ? EventLoopGroup bossGroup = new NioEventLoopGroup();
? ? ? ? EventLoopGroup workerGroup = new NioEventLoopGroup();
? ? ? ? ServerBootstrap b = new ServerBootstrap();
? ? ? ? b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
? ? ? ? ? ? ? ? .childHandler(new ChannelInitializer <SocketChannel>() {
? ? ? ? ? ? ? ? ? ? @Override
? ? ? ? ? ? ? ? ? ? protected void initChannel(SocketChannel ch) throws Exception {
? ? ? ? ? ? ? ? ? ? ? ? ChannelPipeline pipeline = ch.pipeline();
? ? ? ? ? ? ? ? ? ? ? ? // Netty自己的http解碼器和編碼器，報(bào)文級(jí)別 HTTP請(qǐng)求的解碼和編碼
? ? ? ? ? ? ? ? ? ? ? ? pipeline.addLast(new HttpServerCodec());
? ? ? ? ? ? ? ? ? ? ? ? // ChunkedWriteHandler 是用于大數(shù)據(jù)的分區(qū)傳輸
? ? ? ? ? ? ? ? ? ? ? ? // 主要用于處理大數(shù)據(jù)流，比如一個(gè)1G大小的文件如果你直接傳輸肯定會(huì)撐暴jvm內(nèi)存的;
? ? ? ? ? ? ? ? ? ? ? ? // 增加之后就不用考慮這個(gè)問題了
? ? ? ? ? ? ? ? ? ? ? ? pipeline.addLast(new ChunkedWriteHandler());
? ? ? ? ? ? ? ? ? ? ? ? // HttpObjectAggregator 是完全的解析Http消息體請(qǐng)求用的
? ? ? ? ? ? ? ? ? ? ? ? // 把多個(gè)消息轉(zhuǎn)換為一個(gè)單一的完全FullHttpRequest或是FullHttpResponse，
? ? ? ? ? ? ? ? ? ? ? ? // 原因是HTTP解碼器會(huì)在每個(gè)HTTP消息中生成多個(gè)消息對(duì)象HttpRequest/HttpResponse,HttpContent,LastHttpContent
? ? ? ? ? ? ? ? ? ? ? ? pipeline.addLast(new HttpObjectAggregator(64 * 1024));
? ? ? ? ? ? ? ? ? ? ? ? // WebSocket數(shù)據(jù)壓縮
? ? ? ? ? ? ? ? ? ? ? ? pipeline.addLast(new WebSocketServerCompressionHandler());
? ? ? ? ? ? ? ? ? ? ? ? // WebSocketServerProtocolHandler是配置websocket的監(jiān)聽地址/協(xié)議包長度限制
? ? ? ? ? ? ? ? ? ? ? ? pipeline.addLast(new WebSocketServerProtocolHandler("/ws", null, true, 10 * 1024));

? ? ? ? ? ? ? ? ? ? ? ? // 當(dāng)連接在60秒內(nèi)沒有接收到消息時(shí)，就會(huì)觸發(fā)一個(gè) IdleStateEvent 事件，
? ? ? ? ? ? ? ? ? ? ? ? // 此事件被 HeartbeatHandler 的 userEventTriggered 方法處理到
? ? ? ? ? ? ? ? ? ? ? ? pipeline.addLast(
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? new IdleStateHandler(READ_IDLE_TIME_OUT, WRITE_IDLE_TIME_OUT, ALL_IDLE_TIME_OUT, TimeUnit.SECONDS));

? ? ? ? ? ? ? ? ? ? ? ? // WebSocketServerHandler、TextWebSocketFrameHandler 是自定義邏輯處理器，
? ? ? ? ? ? ? ? ? ? ? ? pipeline.addLast(new WebSocketTextHandler());
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? });
? ? ? ? Channel ch = b.bind(port).syncUninterruptibly().channel();
? ? ? ? ch.closeFuture().syncUninterruptibly();

? ? ? ? // 返回與當(dāng)前Java應(yīng)用程序關(guān)聯(lián)的運(yùn)行時(shí)對(duì)象
? ? ? ? Runtime.getRuntime().addShutdownHook(new Thread() {
? ? ? ? ? ? @Override
? ? ? ? ? ? public void run() {
? ? ? ? ? ? ? ? SessionGroup.inst().shutdownGracefully();
? ? ? ? ? ? ? ? bossGroup.shutdownGracefully();
? ? ? ? ? ? ? ? workerGroup.shutdownGracefully();
? ? ? ? ? ? }
? ? ? ? });
? ? }
}

6.index.html

<!DOCTYPE HTML>
<html>
<head>
? ? <meta http-equiv="Content-Type" content="text/html; charset=utf-8"/>
? ? <title>群聊天室</title>
? ? <style type="text/css">
? ? ? ? body {
? ? ? ? ? ? margin-right:50px;
? ? ? ? ? ? margin-left:50px;
? ? ? ? }
? ? ? ? .ddois {
? ? ? ? ? ? position: fixed;
? ? ? ? ? ? left: 120px;
? ? ? ? ? ? bottom: 30px;
? ? ? ? }
? ? </style>
</head>
<body>
群名：<input type="text" id="room" name="group" placeholder="請(qǐng)輸入群">
<br /><br />
昵稱：<input type="text" id="nick" name="name" placeholder="請(qǐng)輸入昵稱">
<br /><br />
<button type="button" onclick="enter()">進(jìn)入聊天群</button>
<br /><br />
<div id="message"></div>
<br /><br />
<div class="ddois">
? ? <textarea name="send" id="text" rows="10" cols="30" placeholder="輸入發(fā)送消息"></textarea>
? ? <br /><br />
? ? <button type="button" onclick="send()">發(fā)送</button>
</div>
<script type="text/javascript">
? ? var webSocket;

? ? if (window.WebSocket) {
? ? ? ? webSocket = new WebSocket("ws://localhost:8088/ws");
? ? } else {
? ? ? ? alert("抱歉，您的瀏覽器不支持WebSocket協(xié)議!");
? ? }

? ? //連通之后的回調(diào)事件
? ? webSocket.onopen = function() {
? ? ? ? console.log("已經(jīng)連通了websocket");
// ? ? ? ? ? ? ? ?setMessageInnerHTML("已經(jīng)連通了websocket");
? ? };
? ? //連接發(fā)生錯(cuò)誤的回調(diào)方法
? ? webSocket.onerror = function(event){
? ? ? ? console.log("出錯(cuò)了");
// ? ? ? ? ? ? ?setMessageInnerHTML("連接失敗");
? ? };

? ? //連接關(guān)閉的回調(diào)方法
? ? webSocket.onclose = function(){
? ? ? ? console.log("連接已關(guān)閉...");

? ? }

? ? //接收到消息的回調(diào)方法
? ? webSocket.onmessage = function(event){
? ? ? ? console.log("bbdds");
? ? ? ? var data = JSON.parse(event.data)
? ? ? ? var msg = data.msg;
? ? ? ? var nick = data.sendUser;
? ? ? ? switch(data.type){
? ? ? ? ? ? case 'init':
? ? ? ? ? ? ? ? console.log("mmll");
? ? ? ? ? ? ? ? break;
? ? ? ? ? ? case 'msg':
? ? ? ? ? ? ? ? console.log("bblld");
? ? ? ? ? ? ? ? setMessageInnerHTML(nick+": ?"+msg);
? ? ? ? ? ? ? ? break;
? ? ? ? ? ? default:
? ? ? ? ? ? ? ? break;
? ? ? ? }
? ? }
? ? function enter(){
? ? ? ? var map = new Map();
? ? ? ? var nick=document.getElementById('nick').value;
? ? ? ? var room=document.getElementById('room').value;
? ? ? ? map.set("type","init");
? ? ? ? map.set("nick",nick);
? ? ? ? console.log(room);
? ? ? ? map.set("room",room);
? ? ? ? var message = Map2Json(map);
? ? ? ? webSocket.send(message);
? ? }

? ? function send() {
? ? ? ? var msg = document.getElementById('text').value;
? ? ? ? var nick = document.getElementById('nick').value;
? ? ? ? console.log("1:"+msg);
? ? ? ? if (msg != null && msg != ""){
? ? ? ? ? ? var map = new Map();
? ? ? ? ? ? map.set("type","msg");
? ? ? ? ? ? map.set("msg",msg);
? ? ? ? ? ? var map2json=Map2Json(map);
? ? ? ? ? ? if (map2json.length < 8000){
? ? ? ? ? ? ? ? console.log("4:"+map2json);
? ? ? ? ? ? ? ? webSocket.send(map2json);
? ? ? ? ? ? }else {
? ? ? ? ? ? ? ? console.log("文本太長了，少寫一點(diǎn)吧??");
? ? ? ? ? ? }
? ? ? ? }
? ? }

? ? //將消息顯示在網(wǎng)頁上
? ? function setMessageInnerHTML(innerHTML) {
? ? ? ? document.getElementById("message").innerHTML += innerHTML + "
";
? ? }

? ? function Map2Json(map) {
? ? ? ? var str = "{";
? ? ? ? map.forEach(function (value, key) {
? ? ? ? ? ? str += '"'+key+'"'+':'+ '"'+value+'",';
? ? ? ? })
? ? ? ? str = str.substring(0,str.length-1)
? ? ? ? str +="}";
? ? ? ? return str;
? ? }

</script>

</body>
</html>