Springboot+Netty+WebSocket搭建簡單的消息通知

一、快速開始

1、添加依賴

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.36.Final</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

2、添加配置

spring:
  http:
    encoding:
      force: true
      charset: UTF-8
  application:
    name: spring-cloud-study-websocket-reids
  freemarker:
    request-context-attribute: request
    #prefix: /templates/
    suffix: .html
    content-type: text/html
    enabled: true
    cache: false
    charset: UTF-8
    allow-request-override: false
    expose-request-attributes: true
    expose-session-attributes: true
    expose-spring-macro-helpers: true
    #template-loader-path: classpath:/templates/

3、添加啟動類

@SpringBootApplication
public class WebSocketApplication {

    public static void main(String[] args) {
        SpringApplication.run(WebSocketApplication.class);
        try {
            new NettyServer(12345).start();
            System.out.println("https://blog.csdn.net/moshowgame");
            System.out.println("http://127.0.0.1:6688/netty-websocket/index");
        }catch(Exception e) {
            System.out.println("NettyServerError:"+e.getMessage());
        }


    }
}

二、添加WebSocket部分代碼

1、WebSocketServer

@Slf4j
@ServerEndpoint("/imserver/{userId}")
@Component
public class WebSocketServer {

    /**靜態(tài)變量，用來記錄當(dāng)前在線連接數(shù)。應(yīng)該把它設(shè)計成線程安全的。*/
	private static AtomicInteger onlineCount = new AtomicInteger(0);
    /**concurrent包的線程安全Set，用來存放每個客戶端對應(yīng)的MyWebSocket對象。*/
    private static ConcurrentHashMap<String,WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    /**與某個客戶端的連接會話，需要通過它來給客戶端發(fā)送數(shù)據(jù)*/
    private Session session;
    /**接收userId*/
    private String userId="";

    /**
     * 連接建立成功調(diào)用的方法*/
    @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);
        }else{
            webSocketMap.put(userId,this);
            onlineCount.incrementAndGet(); // 在線數(shù)加1
        }
        log.info("用戶連接:"+userId+",當(dāng)前在線人數(shù)為:" +  onlineCount.get());
        try {
            sendMessage("連接成功");
        } catch (IOException e) {
            log.error("用戶:"+userId+",網(wǎng)絡(luò)異常!!!!!!");
        }
    }

    /**
     * 連接關(guān)閉調(diào)用的方法
     */
    @OnClose
    public void onClose() {
        if(webSocketMap.containsKey(userId)){
            webSocketMap.remove(userId);
            onlineCount.decrementAndGet(); // 在線數(shù)減1
        }
        log.info("用戶退出:"+userId+",當(dāng)前在線人數(shù)為:" +  onlineCount.get());
    }

    /**
     * 收到客戶端消息后調(diào)用的方法
     *
     * @param message 客戶端發(fā)送過來的消息*/
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("用戶消息:"+userId+",報文:"+message);
        //可以群發(fā)消息
        //消息保存到數(shù)據(jù)庫、redis
        if(StrUtil.isNotBlank(message)){
            try {
                //解析發(fā)送的報文
                JSONObject jsonObject = JSON.parseObject(message);
                //追加發(fā)送人(防止串改)
                jsonObject.put("fromUserId",this.userId);
                String toUserId=jsonObject.getString("toUserId");
                //傳送給對應(yīng)toUserId用戶的websocket
                if(StrUtil.isNotBlank(toUserId)&&webSocketMap.containsKey(toUserId)){
                    webSocketMap.get(toUserId).sendMessage(jsonObject.toJSONString());
                }else{
                    log.error("請求的userId:"+toUserId+"不在該服務(wù)器上");
                    //否則不在這個服務(wù)器上，發(fā)送到mysql或者redis
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    /**
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("用戶錯誤:"+this.userId+",原因:"+error.getMessage());
        error.printStackTrace();
    }
    /**
     * 實現(xiàn)服務(wù)器主動推送
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }


    /**
     * 發(fā)送自定義消息
     * */
    public static void sendInfo(String message,@PathParam("userId") String userId) throws IOException {
        log.info("發(fā)送消息到:"+userId+"，報文:"+message);
        if(StrUtil.isNotBlank(userId)&&webSocketMap.containsKey(userId)){
            webSocketMap.get(userId).sendMessage(message);
        }else{
            log.error("用戶"+userId+",不在線！");
        }
    }

    public static synchronized AtomicInteger getOnlineCount() {
        return onlineCount;
    }
}

2、WebSocketConfig

@Configuration
public class WebSocketConfig {  
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();  
    }  
} 

3、DemoController

import cn.vipthink.socket.server.WebSocketServer;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.ModelAndView;

import java.io.IOException;

@RestController
public class DemoController {

    @GetMapping("index")
    public ResponseEntity<String> index(){
        return ResponseEntity.ok("請求成功");
    }

    @GetMapping("page")
    public ModelAndView page(){
        return new ModelAndView("index");
    }

    @RequestMapping("/push/{toUserId}")
    public ResponseEntity<String> pushToWeb(String message, @PathVariable String toUserId) throws IOException {
        WebSocketServer.sendInfo(message,toUserId);
        return ResponseEntity.ok("MSG SEND SUCCESS");
    }
}

6、添加templates/index.html

<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1transitional.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<head>
    <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
    <title>Netty-Websocket</title>
    <script type="text/javascript">
        // by zhengkai.blog.csdn.net
        var socket;
        if(!window.WebSocket){
            window.WebSocket = window.MozWebSocket;
        }
        if(window.WebSocket){
            socket = new WebSocket("ws://127.0.0.1:12345/ws");
            socket.onmessage = function(event){
                var ta = document.getElementById('responseText');
                ta.value += event.data+"\r\n";
            };
            socket.onopen = function(event){
                var ta = document.getElementById('responseText');
                ta.value = "Netty-WebSocket服務(wù)器。。。。。。連接  \r\n";
            };
            socket.onclose = function(event){
                var ta = document.getElementById('responseText');
                ta.value = "Netty-WebSocket服務(wù)器。。。。。。關(guān)閉 \r\n";
            };
        }else{
            alert("您的瀏覽器不支持WebSocket協(xié)議！");
        }
        function send(message){
            if(!window.WebSocket){return;}
            if(socket.readyState == WebSocket.OPEN){
                socket.send(message);
            }else{
                alert("WebSocket 連接沒有建立成功！");
            }

        }

    </script>
</head>
<body>
<form onSubmit="return false;">
    <label>ID</label><input type="text" name="uid" value="${uid!!}" /> <br />
    <label>TEXT</label><input type="text" name="message" value="這里輸入消息" /> <br />
    <br /> <input type="button" value="發(fā)送ws消息"
                  onClick="send(this.form.uid.value+':'+this.form.message.value)" />
    <hr color="black" />
    <h3>服務(wù)端返回的應(yīng)答消息</h3>
    <textarea id="responseText" style="width: 1024px;height: 300px;"></textarea>
</form>
</body>
</html>

三、添加Netty部分

1、NettyServer

import cn.vipthink.socket.handler.WSWebSocketHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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.stream.ChunkedWriteHandler;

public class NettyServer {
    private final int port;
 
    public NettyServer(int port) {
        this.port = port;
    }
 
    public void start() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
 
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            ServerBootstrap sb = new ServerBootstrap();
            sb.option(ChannelOption.SO_BACKLOG, 1024);
            sb.group(group, bossGroup) // 綁定線程池
                    .channel(NioServerSocketChannel.class) // 指定使用的channel
                    .localAddress(this.port)// 綁定監(jiān)聽端口
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 綁定客戶端連接時候觸發(fā)操作
 
                        @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 WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10));
                            ch.pipeline().addLast(new WSWebSocketHandler());
                        }
                    });
            ChannelFuture cf = sb.bind().sync(); // 服務(wù)器異步創(chuàng)建綁定
            System.out.println(NettyServer.class + " 啟動正在監(jiān)聽： " + cf.channel().localAddress());
            cf.channel().closeFuture().sync(); // 關(guān)閉服務(wù)器通道
        } finally {
            group.shutdownGracefully().sync(); // 釋放線程池資源
            bossGroup.shutdownGracefully().sync();
        }
    }
}

2、WSChannelHandlerPool

import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * WSChannelHandlerPool
 * 通道組池，管理所有websocket連接
 */
public class WSChannelHandlerPool {

    public WSChannelHandlerPool(){}

    public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

}

3、WSWebSocketHandler

import cn.vipthink.socket.config.WSChannelHandlerPool;
import com.alibaba.fastjson.JSON;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

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

public class WSWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("與客戶端建立連接，通道開啟！");

        //添加到channelGroup通道組
        WSChannelHandlerPool.channelGroup.add(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("與客戶端斷開連接，通道關(guān)閉！");
        //添加到channelGroup 通道組
        WSChannelHandlerPool.channelGroup.remove(ctx.channel());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //首次連接是FullHttpRequest，處理參數(shù) by zhengkai.blog.csdn.net
        if (null != msg && msg instanceof FullHttpRequest) {
            FullHttpRequest request = (FullHttpRequest) msg;
            String uri = request.uri();

            Map paramMap=getUrlParams(uri);
            System.out.println("接收到的參數(shù)是："+ JSON.toJSONString(paramMap));
            //如果url包含參數(shù)，需要處理
            if(uri.contains("?")){
                String newUri=uri.substring(0,uri.indexOf("?"));
                System.out.println(newUri);
                request.setUri(newUri);
            }

        }else if(msg instanceof TextWebSocketFrame){
            //正常的TEXT消息類型
            TextWebSocketFrame frame=(TextWebSocketFrame)msg;
            System.out.println("客戶端收到服務(wù)器數(shù)據(jù)：" +frame.text());
            sendAllMessage(frame.text());
        }
        super.channelRead(ctx, msg);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {

    }

    private void sendAllMessage(String message){
        //收到信息后，群發(fā)給所有channel
        WSChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
    }

    private static Map getUrlParams(String url){
        Map<String,String> map = new HashMap<>();
        url = url.replace("?",";");
        if (!url.contains(";")){
            return map;
        }
        if (url.split(";").length > 0){
            String[] arr = url.split(";")[1].split("&");
            for (String s : arr){
                String key = s.split("=")[0];
                String value = s.split("=")[1];
                map.put(key,value);
            }
            return  map;

        }else{
            return map;
        }
    }
}

四、啟動服務(wù)

http://localhost:9999/demo/page

可以通過前端發(fā)送消息到后端，通過日志查看文章來源地址http://www.zghlxwxcb.cn/news/detail-666934.html

到了這里，關(guān)于Springboot+Netty+WebSocket搭建簡單的消息通知的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！