@Data
public class Chat {
    @TableId(type = IdType.AUTO)
    private Long id;
    private Long userId;
    private Long targetUserId;
    private LocalDateTime createTime;
    private String userName;
    private String targetUserName;
    private String content;
}

@ServerEndpoint("/imserver/{userId}")
@Component
public class WebSocketService {
    /**
     * concurrent包的線程安全Set，用來存放每個客戶端對應(yīng)的MyWebSocket對象。
     */
    private static ConcurrentHashMap<String, WebSocketService> webSocketMap = new ConcurrentHashMap<>();
    /**
     * 與某個客戶端的連接會話，需要通過它來給客戶端發(fā)送數(shù)據(jù)
     */
    private Session session;
    /**
     * 接收userId
     */
    private String userId = "";
    public static ChatMapper chatMapper = null;
    /**
     * 連接建立成功調(diào)用的方法
     * <p>
     * 1.用map存 每個客戶端對應(yīng)的MyWebSocket對象
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        this.userId = userId;
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            webSocketMap.put(userId, this);
            //加入set中
        } else {
            webSocketMap.put(userId, this);
            //加入set中
        }
    }
    /**
     * 報錯
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        error.printStackTrace();
    }
    /**
     * 實現(xiàn)服務(wù)器推送到對應(yīng)的客戶端
     */
    public void sendMessage(String message) {
        try {
            this.session.getBasicRemote().sendText(message);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * 自定義 指定的userId服務(wù)端向客戶端發(fā)送消息
     */
    public static void sendInfo(Chat chat) {
        QueryWrapper<Chat> queryWrapper = new QueryWrapper();
        List<Chat> chats=chatMapper.selectList(queryWrapper.lambda()
                .eq(Chat::getTargetUserId, chat.getTargetUserId()).or().eq(Chat::getUserId, chat.getTargetUserId()).or().eq(Chat::getTargetUserId, chat.getUserId()).or().eq(Chat::getUserId, chat.getUserId()));
        //log.info("發(fā)送消息到:"+userId+"，報文:"+message);
        if (!StringUtils.isEmpty(chat.getTargetUserId().toString()) && webSocketMap.containsKey(chat.getTargetUserId().toString())) {
            webSocketMap.get(chat.getUserId().toString()).sendMessage(JSONObject.toJSONString(chats));
            webSocketMap.get(chat.getTargetUserId().toString()).sendMessage(JSONObject.toJSONString(chats));
        } else {
            webSocketMap.get(chat.getUserId().toString()).sendMessage(JSONObject.toJSONString(chats));
        }
    }
    /**
     * 自定義關(guān)閉
     *
     * @param userId
     */
    public static void close(String userId) {
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
        }
    }
    /**
     * 獲取在線用戶信息
     *
     * @return
     */
    public static Map getOnlineUser() {
        return webSocketMap;
    }

@RestController
public class DemoController {
    @Autowired
    private ChatService chatService;
    @PostMapping("/push")
    public ResponseEntity<String> pushToWeb(@RequestBody Chat chat) throws IOException {
        chat.setCreateTime(LocalDateTime.now());
        chatService.save(chat);
        WebSocketService.sendInfo(chat);
        return ResponseEntity.ok("MSG SEND SUCCESS");
    }
    @GetMapping("/close")
    public String close(String userId) {
        WebSocketService.close(userId);
        return "ok";
    }
    @GetMapping("/getOnlineUser")
    public Map getOnlineUser() {
        return WebSocketService.getOnlineUser();
    }
    @GetMapping("/getMessage")
    public ResponseEntity<List<Chat>> getMessage(String userId) {
        QueryWrapper<Chat> queryWrapper = new QueryWrapper();
        List<Chat> list = chatService.
                list(queryWrapper.lambda().eq(Chat::getTargetUserId, userId).or().eq(Chat::getUserId, userId));
        return ResponseEntity.ok(list);
    }
}

package com.example.demo.config;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
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.stream.ChunkedWriteHandler;
import org.springframework.stereotype.Component;
public class NettyServer {
    public void start(){
        //創(chuàng)建兩個線程組boosGroup和workerGroup,含有的子線程NioEventLoop的個數(shù)默認(rèn)為cpu核數(shù)的兩倍
        //boosGroup只是處理鏈接請求,真正的和客戶端業(yè)務(wù)處理,會交給workerGroup完成
        EventLoopGroup boosGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //創(chuàng)建服務(wù)器的啟動對象
            ServerBootstrap bootstrap = new ServerBootstrap();
            //使用鏈?zhǔn)骄幊虂砼渲脜?shù)
            //設(shè)置兩個線程組
            bootstrap.group(boosGroup,workerGroup)
                    //使用NioSctpServerChannel作為服務(wù)器的通道實現(xiàn)
                    .channel(NioServerSocketChannel.class)
                    //初始化服務(wù)器鏈接隊列大小，服務(wù)端處理客戶端鏈接請求是順序處理的，所以同一時間只能處理一個客戶端鏈接
                    //多個客戶端同時來的時候，服務(wù)端將不能處理的客戶端鏈接請求放在隊列中等待處理
                    .option(ChannelOption.SO_BACKLOG,1024)
                    //創(chuàng)建通道初始化對象，設(shè)置初始化參數(shù)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            System.out.println("收到到新的鏈接");
                            //websocket協(xié)議本身是基于http協(xié)議的，所以這邊也要使用http解編碼器
                            ch.pipeline().addLast(new HttpServerCodec());
                            //以塊的方式來寫的處理器
                            ch.pipeline().addLast(new ChunkedWriteHandler());
                            ch.pipeline().addLast(new HttpObjectAggregator(8192));
                            ch.pipeline().addLast(new MessageHandler());//添加測試的聊天消息處理類
                            ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10));
                        }
                    });
            System.out.println("netty server start..");
            //綁定一個端口并且同步，生成一個ChannelFuture異步對象，通過isDone()等方法判斷異步事件的執(zhí)行情況
            //啟動服務(wù)器(并綁定端口)，bind是異步操作，sync方法是等待異步操作執(zhí)行完畢
            ChannelFuture cf = bootstrap.bind(1245).sync();
            //給cf注冊監(jiān)聽器,監(jiān)聽我們關(guān)心的事件
            cf.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (cf.isSuccess()){
                        System.out.println("監(jiān)聽端口成功");
                    }else {
                        System.out.println("監(jiān)聽端口失敗");
                    }
                }
            });
            //對通道關(guān)閉進(jìn)行監(jiān)聽，closeFuture是異步操作,監(jiān)聽通道關(guān)閉
            //通過sync方法同步等待通道關(guān)閉處理完畢，這里會阻塞等待通道關(guān)閉
            cf.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            boosGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

if (!window.WebSocket) {
							window.WebSocket = window.MozWebSocket;
						}
						if (window.WebSocket) {
							me.websocket = new WebSocket(me.ws + me.info.id);
							me.websocket.onmessage = function(event) {
								var json = JSON.parse(event.data);
								me.msgListMethod();
								console.log(json);
							};
							console.log(me.websocket)
							me.websocket.onopen = function(event) {
								console.log("Netty-WebSocket服務(wù)器。。。。。。連接");
							};
							me.websocket.onerror = function(evt) {
								console.log('發(fā)生錯誤..., evt');
							};
							me.websocket.CONNECTING = function(evt) {
								console.log('正在鏈接中');
							};
						} else {
							alert("您的瀏覽器不支持WebSocket協(xié)議！");
						}
						//監(jiān)聽窗口關(guān)閉事件，當(dāng)窗口關(guān)閉時，主動去關(guān)閉websocket連接，防止連接還沒斷開就關(guān)閉窗口，server端會拋異常。
						window.onbeforeunload = function() {
							if (me.websocket != null) {
								me.websocket.close();
							}
						};