Springboot+Netty+Websocket實(shí)現(xiàn)消息推送實(shí)例

更新時(shí)間：2021年02月03日 11:39:58 作者：青椒1013

這篇文章主要介紹了Springboot+Netty+Websocket實(shí)現(xiàn)消息推送實(shí)例,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

前言

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

Netty框架的優(yōu)勢(shì)

1. API使用簡(jiǎn)單，開發(fā)門檻低；
2. 功能強(qiáng)大，預(yù)置了多種編解碼功能，支持多種主流協(xié)議；
3. 定制能力強(qiáng)，可以通過ChannelHandler對(duì)通信框架進(jìn)行靈活地?cái)U(kuò)展；
4. 性能高，通過與其他業(yè)界主流的NIO框架對(duì)比，Netty的綜合性能最優(yōu)；
5. 成熟、穩(wěn)定，Netty修復(fù)了已經(jīng)發(fā)現(xiàn)的所有JDK NIO BUG，業(yè)務(wù)開發(fā)人員不需要再為NIO的BUG而煩惱

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

一、引入netty依賴

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

二、使用步驟

1.引入基礎(chǔ)配置類

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服務(wù)啟動(dòng)監(jiān)聽器

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>
 * 服務(wù)啟動(dòng)監(jiān)聽器
 **/
@Slf4j
@Component
public class NettyServer {

 @Value("${server.netty.port}")
 private int port;

 @Autowired
 private ServerChannelInitializer serverChannelInitializer;

 @Bean
 ApplicationRunner nettyRunner() {
  return args -> {
   //new 一個(gè)主線程組
   EventLoopGroup bossGroup = new NioEventLoopGroup(1);
   //new 一個(gè)工作線程組
   EventLoopGroup workGroup = new NioEventLoopGroup();
   ServerBootstrap bootstrap = new ServerBootstrap()
     .group(bossGroup, workGroup)
     .channel(NioServerSocketChannel.class)
     .childHandler(serverChannelInitializer)
     //設(shè)置隊(duì)列大小
     .option(ChannelOption.SO_BACKLOG, 1024)
     // 兩小時(shí)內(nèi)沒有數(shù)據(jù)的通信時(shí),TCP會(huì)自動(dòng)發(fā)送一個(gè)活動(dòng)探測(cè)數(shù)據(jù)報(bào)文
     .childOption(ChannelOption.SO_KEEPALIVE, true);
   //綁定端口,開始接收進(jìn)來的連接
   try {
    ChannelFuture future = bootstrap.bind(port).sync();
    log.info("服務(wù)器啟動(dòng)開始監(jiān)聽端口: {}", port);
    future.channel().closeFuture().sync();
   } catch (InterruptedException e) {
    e.printStackTrace();
   } finally {
    //關(guān)閉主線程組
    bossGroup.shutdownGracefully();
    //關(guān)閉工作線程組
    workGroup.shutdownGracefully();
   }
  };
 }
}

3.netty服務(wù)端處理器

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服務(wù)端處理器
 **/
@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 {
 }


 /**
  * 客戶端連接會(huì)觸發(fā)
  */
 @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);
 }

 /**
  * 發(fā)生異常觸發(fā)
  */
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  Channel channel = ctx.channel();
  log.error("連接出現(xiàn)異常，ID->{}，用戶ID->{}，異常->{}......", channel.id().asLongText(), cache.getCacheId(channel), cause.getMessage(), cause);
  cache.remove(channel);
  ctx.close();
 }

 /**
  * 客戶端發(fā)消息會(huì)觸發(fā)
  */
 @Override
 protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
  try {
   // log.info("接收到客戶端發(fā)送的消息：{}", 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("服務(wù)器下發(fā)消息: {}", 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("服務(wù)器下發(fā)消息: {}", msg);
  channel.writeAndFlush(new TextWebSocketFrame(msg));
 }

}

4.netty服務(wù)端緩存類

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服務(wù)初始化器

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服務(wù)初始化器
 **/
@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測(cè)試

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

7.vue測(cè)試

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: "建立實(shí)時(shí)連接成功！",
        duration: 1000
       })
       setInterval(()=>{websocket.send("heartbeat")},60*1000);
       break;
      case "001":
       this.$message.warning("收到一條新的信息，請(qǐng)及時(shí)查看！")
       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.服務(wù)器下發(fā)消息

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

到此這篇關(guān)于Springboot+Netty+Websocket實(shí)現(xiàn)消息推送實(shí)例的文章就介紹到這了,更多相關(guān)Springboot Websocket消息推送內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: