Springboot+Netty+Websocket實現消息推送實例

作者：麥叔 2021-02-05 07:28:11

開發前端

WebSocket 使得客戶端和服務器之間的數據交換變得更加簡單，允許服務端主動向客戶端推送數據。在WebSocket API 中，瀏覽器和服務器只需要完成一次握手，兩者之間就直接可以創建持久性的連接，并進行雙向數據傳輸。

[[380697]]

前言

Netty框架的優勢

1.API使用簡單，開發門檻低;

2.功能強大，預置了多種編解碼功能，支持多種主流協議;

3.定制能力強，可以通過ChannelHandler對通信框架進行靈活地擴展;

4.性能高，通過與其他業界主流的NIO框架對比，Netty的綜合性能最優;

5.成熟、穩定，Netty修復了已經發現的所有JDK NIO BUG，業務開發人員不需要再為NIO的BUG而煩惱

提示：以下是本篇文章正文內容，下面案例可供參考

一、引入netty依賴

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

二、使用步驟

1.引入基礎配置類

package com.test.netty; 
 
public enum Cmd { 
 START("000", "連接成功"), 
 WMESSAGE("001", "消息提醒"), 
 ; 
 private String cmd; 
 private String desc; 
 
 Cmd(String cmd, String desc) { 
  this.cmd = cmd; 
  this.desc = desc; 
 } 
 
 public String getCmd() { 
  return cmd; 
 } 
 
 public String getDesc() { 
  return desc; 
 } 
}

2.netty服務啟動監聽器

package com.test.netty; 
 
import io.netty.bootstrap.ServerBootstrap; 
import io.netty.channel.ChannelFuture; 
import io.netty.channel.ChannelOption; 
import io.netty.channel.EventLoopGroup; 
import io.netty.channel.nio.NioEventLoopGroup; 
import io.netty.channel.socket.nio.NioServerSocketChannel; 
import lombok.extern.slf4j.Slf4j; 
import org.springframework.beans.factory.annotation.Autowired; 
import org.springframework.beans.factory.annotation.Value; 
import org.springframework.boot.ApplicationRunner; 
import org.springframework.context.annotation.Bean; 
import org.springframework.stereotype.Component; 
 
/** 
 * @author test 
 * <p> 
 * 服務啟動監聽器 
 **/ 
@Slf4j 
@Component 
public class NettyServer { 
 
 @Value("${server.netty.port}") 
 private int port; 
 
 @Autowired 
 private ServerChannelInitializer serverChannelInitializer; 
 
 @Bean 
 ApplicationRunner nettyRunner() { 
  return args -> { 
   //new 一個主線程組 
   EventLoopGroup bossGroup = new NioEventLoopGroup(1); 
   //new 一個工作線程組 
   EventLoopGroup workGroup = new NioEventLoopGroup(); 
   ServerBootstrap bootstrap = new ServerBootstrap() 
     .group(bossGroup, workGroup) 
     .channel(NioServerSocketChannel.class) 
     .childHandler(serverChannelInitializer) 
     //設置隊列大小 
     .option(ChannelOption.SO_BACKLOG, 1024) 
     // 兩小時內沒有數據的通信時,TCP會自動發送一個活動探測數據報文 
     .childOption(ChannelOption.SO_KEEPALIVE, true); 
   //綁定端口,開始接收進來的連接 
   try { 
    ChannelFuture future = bootstrap.bind(port).sync(); 
    log.info("服務器啟動開始監聽端口: {}", port); 
    future.channel().closeFuture().sync(); 
   } catch (InterruptedException e) { 
    e.printStackTrace(); 
   } finally { 
    //關閉主線程組 
    bossGroup.shutdownGracefully(); 
    //關閉工作線程組 
    workGroup.shutdownGracefully(); 
   } 
  }; 
 } 
}

3.netty服務端處理器

package com.test.netty; 
 
import com.test.common.util.JsonUtil; 
import io.netty.channel.Channel; 
import io.netty.channel.ChannelHandler; 
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 lombok.Data; 
import lombok.extern.slf4j.Slf4j; 
import org.springframework.beans.factory.annotation.Autowired; 
import org.springframework.stereotype.Component; 
 
import java.net.URLDecoder; 
import java.util.*; 
 
/** 
 * @author test 
 * <p> 
 * netty服務端處理器 
 **/ 
@Slf4j 
@Component 
@ChannelHandler.Sharable 
public class NettyServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> { 
 
 @Autowired 
 private ServerChannelCache cache; 
 private static final String dataKey = "test="; 
 
 @Data 
 public static class ChannelCache { 
 } 
 
 
 /** 
  * 客戶端連接會觸發 
  */ 
 @Override 
 public void channelActive(ChannelHandlerContext ctx) throws Exception { 
  Channel channel = ctx.channel(); 
  log.info("通道連接已打開，ID->{}......", channel.id().asLongText()); 
 } 
 
 @Override 
 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { 
  if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) { 
   Channel channel = ctx.channel(); 
   WebSocketServerProtocolHandler.HandshakeComplete handshakeComplete = (WebSocketServerProtocolHandler.HandshakeComplete) evt; 
   String requestUri = handshakeComplete.requestUri(); 
   requestUri = URLDecoder.decode(requestUri, "UTF-8"); 
   log.info("HANDSHAKE_COMPLETE，ID->{}，URI->{}", channel.id().asLongText(), requestUri); 
   String socketKey = requestUri.substring(requestUri.lastIndexOf(dataKey) + dataKey.length()); 
   if (socketKey.length() > 0) { 
    cache.add(socketKey, channel); 
    this.send(channel, Cmd.DOWN_START, null); 
   } else { 
    channel.disconnect(); 
    ctx.close(); 
   } 
  } 
  super.userEventTriggered(ctx, evt); 
 } 
 
 @Override 
 public void channelInactive(ChannelHandlerContext ctx) throws Exception { 
  Channel channel = ctx.channel(); 
  log.info("通道連接已斷開，ID->{}，用戶ID->{}......", channel.id().asLongText(), cache.getCacheId(channel)); 
  cache.remove(channel); 
 } 
 
 /** 
  * 發生異常觸發 
  */ 
 @Override 
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
  Channel channel = ctx.channel(); 
  log.error("連接出現異常，ID->{}，用戶ID->{}，異常->{}......", channel.id().asLongText(), cache.getCacheId(channel), cause.getMessage(), cause); 
  cache.remove(channel); 
  ctx.close(); 
 } 
 
 /** 
  * 客戶端發消息會觸發 
  */ 
 @Override 
 protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { 
  try { 
   // log.info("接收到客戶端發送的消息：{}", msg.text()); 
   ctx.channel().writeAndFlush(new TextWebSocketFrame(JsonUtil.toString(Collections.singletonMap("cmd", "100")))); 
  } catch (Exception e) { 
   log.error("消息處理異常：{}", e.getMessage(), e); 
  } 
 } 
 
 public void send(Cmd cmd, String id, Object obj) { 
  HashMap<String, Channel> channels = cache.get(id); 
  if (channels == null) { 
   return; 
  } 
  Map<String, Object> data = new LinkedHashMap<>(); 
  data.put("cmd", cmd.getCmd()); 
  data.put("data", obj); 
  String msg = JsonUtil.toString(data); 
  log.info("服務器下發消息: {}", msg); 
  channels.values().forEach(channel -> { 
   channel.writeAndFlush(new TextWebSocketFrame(msg)); 
  }); 
 } 
 
 public void send(Channel channel, Cmd cmd, Object obj) { 
  Map<String, Object> data = new LinkedHashMap<>(); 
  data.put("cmd", cmd.getCmd()); 
  data.put("data", obj); 
  String msg = JsonUtil.toString(data); 
  log.info("服務器下發消息: {}", msg); 
  channel.writeAndFlush(new TextWebSocketFrame(msg)); 
 } 
 
}

4.netty服務端緩存類

package com.test.netty; 
 
import io.netty.channel.Channel; 
import io.netty.util.AttributeKey; 
import org.springframework.stereotype.Component; 
 
import java.util.HashMap; 
import java.util.concurrent.ConcurrentHashMap; 
 
@Component 
public class ServerChannelCache { 
 private static final ConcurrentHashMap<String, HashMap<String, Channel>> CACHE_MAP = new ConcurrentHashMap<>(); 
 private static final AttributeKey<String> CHANNEL_ATTR_KEY = AttributeKey.valueOf("test"); 
 
 public String getCacheId(Channel channel) { 
  return channel.attr(CHANNEL_ATTR_KEY).get(); 
 } 
 
 public void add(String cacheId, Channel channel) { 
  channel.attr(CHANNEL_ATTR_KEY).set(cacheId); 
  HashMap<String, Channel> hashMap = CACHE_MAP.get(cacheId); 
  if (hashMap == null) { 
   hashMap = new HashMap<>(); 
  } 
  hashMap.put(channel.id().asShortText(), channel); 
  CACHE_MAP.put(cacheId, hashMap); 
 } 
 
 public HashMap<String, Channel> get(String cacheId) { 
  if (cacheId == null) { 
   return null; 
  } 
  return CACHE_MAP.get(cacheId); 
 } 
 
 public void remove(Channel channel) { 
  String cacheId = getCacheId(channel); 
  if (cacheId == null) { 
   return; 
  } 
  HashMap<String, Channel> hashMap = CACHE_MAP.get(cacheId); 
  if (hashMap == null) { 
   hashMap = new HashMap<>(); 
  } 
  hashMap.remove(channel.id().asShortText()); 
  CACHE_MAP.put(cacheId, hashMap); 
 } 
}

5.netty服務初始化器

package com.test.netty; 
 
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; 
import org.springframework.beans.factory.annotation.Autowired; 
import org.springframework.stereotype.Component; 
 
/** 
 * @author test 
 * <p> 
 * netty服務初始化器 
 **/ 
@Component 
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> { 
 
 @Autowired 
 private NettyServerHandler nettyServerHandler; 
 
 @Override 
 protected void initChannel(SocketChannel socketChannel) throws Exception { 
  ChannelPipeline pipeline = socketChannel.pipeline(); 
  pipeline.addLast(new HttpServerCodec()); 
  pipeline.addLast(new ChunkedWriteHandler()); 
  pipeline.addLast(new HttpObjectAggregator(8192)); 
  pipeline.addLast(new WebSocketServerProtocolHandler("/test.io", true, 5000)); 
  pipeline.addLast(nettyServerHandler); 
 } 
}

6.html測試

<!DOCTYPE HTML> 
<html> 
 <head> 
 <meta charset="utf-8"> 
 <title>test</title> 
  
  <script type="text/javascript"> 
   function WebSocketTest() 
   { 
   if ("WebSocket" in window) 
   { 
    alert("您的瀏覽器支持 WebSocket!"); 
     
    // 打開一個 web socket 
    var ws = new WebSocket("ws://localhost:port/test.io"); 
     
    ws.onopen = function() 
    { 
     // Web Socket 已連接上，使用 send() 方法發送數據 
     ws.send("發送數據"); 
     alert("數據發送中..."); 
    }; 
     
    ws.onmessage = function (evt)  
    {  
     var received_msg = evt.data; 
     alert("數據已接收..."); 
    }; 
     
    ws.onclose = function() 
    {  
     // 關閉 websocket 
     alert("連接已關閉...");  
    }; 
   } 
    
   else 
   { 
    // 瀏覽器不支持 WebSocket 
    alert("您的瀏覽器不支持 WebSocket!"); 
   } 
   } 
  </script> 
   
 </head> 
 <body> 
  
  <div id="sse"> 
   <a href="javascript:WebSocketTest()" rel="external nofollow" >運行 WebSocket</a> 
  </div> 
   
 </body> 
</html>

7.vue測試

mounted() { 
   this.initWebsocket(); 
  }, 
  methods: { 
   initWebsocket() { 
    let websocket = new WebSocket('ws://localhost:port/test.io?test=123456'); 
    websocket.onmessage = (event) => { 
     let msg = JSON.parse(event.data); 
     switch (msg.cmd) { 
      case "000": 
       this.$message({ 
        type: 'success', 
        message: "建立實時連接成功！", 
        duration: 1000 
       }) 
       setInterval(()=>{websocket.send("heartbeat")},60*1000); 
       break; 
      case "001": 
       this.$message.warning("收到一條新的信息，請及時查看！") 
       break; 
     } 
    } 
    websocket.onclose = () => { 
     setTimeout(()=>{ 
      this.initWebsocket(); 
     },30*1000); 
    } 
    websocket.onerror = () => { 
     setTimeout(()=>{ 
      this.initWebsocket(); 
     },30*1000); 
    } 
   }, 
  }, 
![在這裡插入圖片描述](https://img-blog.csdnimg.cn/20210107160420568.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1X3Fpbmdfc29uZw==,size_16,color_FFFFFF,t_70#pic_center)

8.服務器下發消息

@Autowired 
 private NettyServerHandler nettyServerHandler; 
nettyServerHandler.send(CmdWeb.WMESSAGE, id, message);

到此這篇關于Springboot+Netty+Websocket實現消息推送實例的文章就介紹到這了希望大家以后多多支持麥叔!

責任編輯：姜華來源：今日Java