自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

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

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

[[380697]]

 前言

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)單，開(kāi)發(fā)門檻低;

2.功能強(qiáng)大，預(yù)置了多種編解碼功能，支持多種主流協(xié)議;

3.定制能力強(qiáng)，可以通過(guò)ChannelHandler對(duì)通信框架進(jìn)行靈活地?cái)U(kuò)展;

4.性能高，通過(guò)與其他業(yè)界主流的NIO框架對(duì)比，Netty的綜合性能最優(yōu);

5.成熟、穩(wěn)定，Netty修復(fù)了已經(jīng)發(fā)現(xiàn)的所有JDK NIO BUG，業(yè)務(wù)開(kāi)發(fā)人員不需要再為NIO的BUG而煩惱

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

一、引入netty依賴

  1. <dependency> 
  2.    <groupId>io.netty</groupId> 
  3.    <artifactId>netty-all</artifactId> 
  4.    <version>4.1.48.Final</version> 
  5. </dependency> 

二、使用步驟

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

  1. package com.test.netty; 
  2.  
  3. public enum Cmd { 
  4.  START("000", "連接成功"), 
  5.  WMESSAGE("001", "消息提醒"), 
  6.  ; 
  7.  private String cmd; 
  8.  private String desc; 
  9.  
  10.  Cmd(String cmd, String desc) { 
  11.   this.cmd = cmd; 
  12.   this.desc = desc; 
  13.  } 
  14.  
  15.  public String getCmd() { 
  16.   return cmd; 
  17.  } 
  18.  
  19.  public String getDesc() { 
  20.   return desc; 
  21.  } 
  22. } 

2.netty服務(wù)啟動(dòng)監(jiān)聽(tīng)器

  1. package com.test.netty; 
  2.  
  3. import io.netty.bootstrap.ServerBootstrap; 
  4. import io.netty.channel.ChannelFuture; 
  5. import io.netty.channel.ChannelOption; 
  6. import io.netty.channel.EventLoopGroup; 
  7. import io.netty.channel.nio.NioEventLoopGroup; 
  8. import io.netty.channel.socket.nio.NioServerSocketChannel; 
  9. import lombok.extern.slf4j.Slf4j; 
  10. import org.springframework.beans.factory.annotation.Autowired; 
  11. import org.springframework.beans.factory.annotation.Value; 
  12. import org.springframework.boot.ApplicationRunner; 
  13. import org.springframework.context.annotation.Bean; 
  14. import org.springframework.stereotype.Component; 
  15.  
  16. /** 
  17.  * @author test 
  18.  * <p> 
  19.  * 服務(wù)啟動(dòng)監(jiān)聽(tīng)器 
  20.  **/ 
  21. @Slf4j 
  22. @Component 
  23. public class NettyServer { 
  24.  
  25.  @Value("${server.netty.port}") 
  26.  private int port; 
  27.  
  28.  @Autowired 
  29.  private ServerChannelInitializer serverChannelInitializer; 
  30.  
  31.  @Bean 
  32.  ApplicationRunner nettyRunner() { 
  33.   return args -> { 
  34.    //new 一個(gè)主線程組 
  35.    EventLoopGroup bossGroup = new NioEventLoopGroup(1); 
  36.    //new 一個(gè)工作線程組 
  37.    EventLoopGroup workGroup = new NioEventLoopGroup(); 
  38.    ServerBootstrap bootstrap = new ServerBootstrap() 
  39.      .group(bossGroup, workGroup) 
  40.      .channel(NioServerSocketChannel.class) 
  41.      .childHandler(serverChannelInitializer) 
  42.      //設(shè)置隊(duì)列大小 
  43.      .option(ChannelOption.SO_BACKLOG, 1024) 
  44.      // 兩小時(shí)內(nèi)沒(méi)有數(shù)據(jù)的通信時(shí),TCP會(huì)自動(dòng)發(fā)送一個(gè)活動(dòng)探測(cè)數(shù)據(jù)報(bào)文 
  45.      .childOption(ChannelOption.SO_KEEPALIVE, true); 
  46.    //綁定端口,開(kāi)始接收進(jìn)來(lái)的連接 
  47.    try { 
  48.     ChannelFuture future = bootstrap.bind(port).sync(); 
  49.     log.info("服務(wù)器啟動(dòng)開(kāi)始監(jiān)聽(tīng)端口: {}", port); 
  50.     future.channel().closeFuture().sync(); 
  51.    } catch (InterruptedException e) { 
  52.     e.printStackTrace(); 
  53.    } finally { 
  54.     //關(guān)閉主線程組 
  55.     bossGroup.shutdownGracefully(); 
  56.     //關(guān)閉工作線程組 
  57.     workGroup.shutdownGracefully(); 
  58.    } 
  59.   }; 
  60.  } 
  61. } 

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

  1. package com.test.netty; 
  2.  
  3. import com.test.common.util.JsonUtil; 
  4. import io.netty.channel.Channel; 
  5. import io.netty.channel.ChannelHandler; 
  6. import io.netty.channel.ChannelHandlerContext; 
  7. import io.netty.channel.SimpleChannelInboundHandler; 
  8. import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; 
  9. import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; 
  10. import lombok.Data; 
  11. import lombok.extern.slf4j.Slf4j; 
  12. import org.springframework.beans.factory.annotation.Autowired; 
  13. import org.springframework.stereotype.Component; 
  14.  
  15. import java.net.URLDecoder; 
  16. import java.util.*; 
  17.  
  18. /** 
  19.  * @author test 
  20.  * <p> 
  21.  * netty服務(wù)端處理器 
  22.  **/ 
  23. @Slf4j 
  24. @Component 
  25. @ChannelHandler.Sharable 
  26. public class NettyServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> { 
  27.  
  28.  @Autowired 
  29.  private ServerChannelCache cache; 
  30.  private static final String dataKey = "test="; 
  31.  
  32.  @Data 
  33.  public static class ChannelCache { 
  34.  } 
  35.  
  36.  
  37.  /** 
  38.   * 客戶端連接會(huì)觸發(fā) 
  39.   */ 
  40.  @Override 
  41.  public void channelActive(ChannelHandlerContext ctx) throws Exception { 
  42.   Channel channel = ctx.channel(); 
  43.   log.info("通道連接已打開(kāi)，ID->{}......", channel.id().asLongText()); 
  44.  } 
  45.  
  46.  @Override 
  47.  public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { 
  48.   if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) { 
  49.    Channel channel = ctx.channel(); 
  50.    WebSocketServerProtocolHandler.HandshakeComplete handshakeComplete = (WebSocketServerProtocolHandler.HandshakeComplete) evt; 
  51.    String requestUri = handshakeComplete.requestUri(); 
  52.    requestUri = URLDecoder.decode(requestUri, "UTF-8"); 
  53.    log.info("HANDSHAKE_COMPLETE，ID->{}，URI->{}", channel.id().asLongText(), requestUri); 
  54.    String socketKey = requestUri.substring(requestUri.lastIndexOf(dataKey) + dataKey.length()); 
  55.    if (socketKey.length() > 0) { 
  56.     cache.add(socketKey, channel); 
  57.     this.send(channel, Cmd.DOWN_START, null); 
  58.    } else { 
  59.     channel.disconnect(); 
  60.     ctx.close(); 
  61.    } 
  62.   } 
  63.   super.userEventTriggered(ctx, evt); 
  64.  } 
  65.  
  66.  @Override 
  67.  public void channelInactive(ChannelHandlerContext ctx) throws Exception { 
  68.   Channel channel = ctx.channel(); 
  69.   log.info("通道連接已斷開(kāi)，ID->{}，用戶ID->{}......", channel.id().asLongText(), cache.getCacheId(channel)); 
  70.   cache.remove(channel); 
  71.  } 
  72.  
  73.  /** 
  74.   * 發(fā)生異常觸發(fā) 
  75.   */ 
  76.  @Override 
  77.  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
  78.   Channel channel = ctx.channel(); 
  79.   log.error("連接出現(xiàn)異常，ID->{}，用戶ID->{}，異常->{}......", channel.id().asLongText(), cache.getCacheId(channel), cause.getMessage(), cause); 
  80.   cache.remove(channel); 
  81.   ctx.close(); 
  82.  } 
  83.  
  84.  /** 
  85.   * 客戶端發(fā)消息會(huì)觸發(fā) 
  86.   */ 
  87.  @Override 
  88.  protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { 
  89.   try { 
  90.    // log.info("接收到客戶端發(fā)送的消息：{}", msg.text()); 
  91.    ctx.channel().writeAndFlush(new TextWebSocketFrame(JsonUtil.toString(Collections.singletonMap("cmd", "100")))); 
  92.   } catch (Exception e) { 
  93.    log.error("消息處理異常：{}", e.getMessage(), e); 
  94.   } 
  95.  } 
  96.  
  97.  public void send(Cmd cmd, String id, Object obj) { 
  98.   HashMap<String, Channel> channels = cache.get(id); 
  99.   if (channels == null) { 
  100.    return; 
  101.   } 
  102.   Map<String, Object> data = new LinkedHashMap<>(); 
  103.   data.put("cmd", cmd.getCmd()); 
  104.   data.put("data", obj); 
  105.   String msg = JsonUtil.toString(data); 
  106.   log.info("服務(wù)器下發(fā)消息: {}", msg); 
  107.   channels.values().forEach(channel -> { 
  108.    channel.writeAndFlush(new TextWebSocketFrame(msg)); 
  109.   }); 
  110.  } 
  111.  
  112.  public void send(Channel channel, Cmd cmd, Object obj) { 
  113.   Map<String, Object> data = new LinkedHashMap<>(); 
  114.   data.put("cmd", cmd.getCmd()); 
  115.   data.put("data", obj); 
  116.   String msg = JsonUtil.toString(data); 
  117.   log.info("服務(wù)器下發(fā)消息: {}", msg); 
  118.   channel.writeAndFlush(new TextWebSocketFrame(msg)); 
  119.  } 
  120.  
  121. } 

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

  1. package com.test.netty; 
  2.  
  3. import io.netty.channel.Channel; 
  4. import io.netty.util.AttributeKey; 
  5. import org.springframework.stereotype.Component; 
  6.  
  7. import java.util.HashMap; 
  8. import java.util.concurrent.ConcurrentHashMap; 
  9.  
  10. @Component 
  11. public class ServerChannelCache { 
  12.  private static final ConcurrentHashMap<String, HashMap<String, Channel>> CACHE_MAP = new ConcurrentHashMap<>(); 
  13.  private static final AttributeKey<String> CHANNEL_ATTR_KEY = AttributeKey.valueOf("test"); 
  14.  
  15.  public String getCacheId(Channel channel) { 
  16.   return channel.attr(CHANNEL_ATTR_KEY).get(); 
  17.  } 
  18.  
  19.  public void add(String cacheId, Channel channel) { 
  20.   channel.attr(CHANNEL_ATTR_KEY).set(cacheId); 
  21.   HashMap<String, Channel> hashMap = CACHE_MAP.get(cacheId); 
  22.   if (hashMap == null) { 
  23.    hashMap = new HashMap<>(); 
  24.   } 
  25.   hashMap.put(channel.id().asShortText(), channel); 
  26.   CACHE_MAP.put(cacheId, hashMap); 
  27.  } 
  28.  
  29.  public HashMap<String, Channel> get(String cacheId) { 
  30.   if (cacheId == null) { 
  31.    return null; 
  32.   } 
  33.   return CACHE_MAP.get(cacheId); 
  34.  } 
  35.  
  36.  public void remove(Channel channel) { 
  37.   String cacheId = getCacheId(channel); 
  38.   if (cacheId == null) { 
  39.    return; 
  40.   } 
  41.   HashMap<String, Channel> hashMap = CACHE_MAP.get(cacheId); 
  42.   if (hashMap == null) { 
  43.    hashMap = new HashMap<>(); 
  44.   } 
  45.   hashMap.remove(channel.id().asShortText()); 
  46.   CACHE_MAP.put(cacheId, hashMap); 
  47.  } 
  48. } 

5.netty服務(wù)初始化器

  1. package com.test.netty; 
  2.  
  3. import io.netty.channel.ChannelInitializer; 
  4. import io.netty.channel.ChannelPipeline; 
  5. import io.netty.channel.socket.SocketChannel; 
  6. import io.netty.handler.codec.http.HttpObjectAggregator; 
  7. import io.netty.handler.codec.http.HttpServerCodec; 
  8. import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; 
  9. import io.netty.handler.stream.ChunkedWriteHandler; 
  10. import org.springframework.beans.factory.annotation.Autowired; 
  11. import org.springframework.stereotype.Component; 
  12.  
  13. /** 
  14.  * @author test 
  15.  * <p> 
  16.  * netty服務(wù)初始化器 
  17.  **/ 
  18. @Component 
  19. public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> { 
  20.  
  21.  @Autowired 
  22.  private NettyServerHandler nettyServerHandler; 
  23.  
  24.  @Override 
  25.  protected void initChannel(SocketChannel socketChannel) throws Exception { 
  26.   ChannelPipeline pipeline = socketChannel.pipeline(); 
  27.   pipeline.addLast(new HttpServerCodec()); 
  28.   pipeline.addLast(new ChunkedWriteHandler()); 
  29.   pipeline.addLast(new HttpObjectAggregator(8192)); 
  30.   pipeline.addLast(new WebSocketServerProtocolHandler("/test.io", true, 5000)); 
  31.   pipeline.addLast(nettyServerHandler); 
  32.  } 
  33. } 

6.html測(cè)試

  1. <!DOCTYPE HTML> 
  2. <html> 
  3.  <head> 
  4.  <meta charset="utf-8"> 
  5.  <title>test</title> 
  6.   
  7.   <script type="text/javascript"> 
  8.    function WebSocketTest() 
  9.    { 
  10.    if ("WebSocket" in window) 
  11.    { 
  12.     alert("您的瀏覽器支持 WebSocket!"); 
  13.      
  14.     // 打開(kāi)一個(gè) web socket 
  15.     var ws = new WebSocket("ws://localhost:port/test.io"); 
  16.      
  17.     ws.onopen = function() 
  18.     { 
  19.      // Web Socket 已連接上，使用 send() 方法發(fā)送數(shù)據(jù) 
  20.      ws.send("發(fā)送數(shù)據(jù)"); 
  21.      alert("數(shù)據(jù)發(fā)送中..."); 
  22.     }; 
  23.      
  24.     ws.onmessage = function (evt)  
  25.     {  
  26.      var received_msg = evt.data; 
  27.      alert("數(shù)據(jù)已接收..."); 
  28.     }; 
  29.      
  30.     ws.onclose = function() 
  31.     {  
  32.      // 關(guān)閉 websocket 
  33.      alert("連接已關(guān)閉...");  
  34.     }; 
  35.    } 
  36.     
  37.    else 
  38.    { 
  39.     // 瀏覽器不支持 WebSocket 
  40.     alert("您的瀏覽器不支持 WebSocket!"); 
  41.    } 
  42.    } 
  43.   </script> 
  44.    
  45.  </head> 
  46.  <body> 
  47.   
  48.   <div id="sse"> 
  49.    <a href="javascript:WebSocketTest()" rel="external nofollow" >運(yùn)行 WebSocket</a> 
  50.   </div> 
  51.    
  52.  </body> 
  53. </html> 

 7.vue測(cè)試

  1. mounted() { 
  2.    this.initWebsocket(); 
  3.   }, 
  4.   methods: { 
  5.    initWebsocket() { 
  6.     let websocket = new WebSocket('ws://localhost:port/test.io?test=123456'); 
  7.     websocket.onmessage = (event) => { 
  8.      let msg = JSON.parse(event.data); 
  9.      switch (msg.cmd) { 
  10.       case "000": 
  11.        this.$message({ 
  12.         type: 'success', 
  13.         message: "建立實(shí)時(shí)連接成功！", 
  14.         duration: 1000 
  15.        }) 
  16.        setInterval(()=>{websocket.send("heartbeat")},60*1000); 
  17.        break; 
  18.       case "001": 
  19.        this.$message.warning("收到一條新的信息，請(qǐng)及時(shí)查看！") 
  20.        break; 
  21.      } 
  22.     } 
  23.     websocket.onclose = () => { 
  24.      setTimeout(()=>{ 
  25.       this.initWebsocket(); 
  26.      },30*1000); 
  27.     } 
  28.     websocket.onerror = () => { 
  29.      setTimeout(()=>{ 
  30.       this.initWebsocket(); 
  31.      },30*1000); 
  32.     } 
  33.    }, 
  34.   }, 
  35. ![在這裡插入圖片描述](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ā)消息

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

到此這篇關(guān)于Springboot+Netty+Websocket實(shí)現(xiàn)消息推送實(shí)例的文章就介紹到這了希望大家以后多多支持麥?zhǔn)?

 

責(zé)任編輯：姜華 來(lái)源： 今日J(rèn)ava
SpringbootNettyWebsocke
相關(guān)推薦

2024-08-02 09:00:17

NettyWebSocketNIO

2021-03-26 08:16:32

SpringbootWebsocket前端

2021-03-25 08:29:33

SpringBootWebSocket即時(shí)消息

2023-08-14 08:01:12

websocket8g用戶

2024-09-02 09:31:19

2024-09-12 14:50:08

2022-06-28 08:37:07

分布式服務(wù)器WebSocket

2024-11-14 11:56:45

2013-05-17 15:34:45

2024-09-11 08:35:54

2023-07-26 07:28:55

WebSocket服務(wù)器方案

2017-09-05 15:30:00

JavascriptSocket.ioNode.js

2023-08-09 08:01:00

WebSockett服務(wù)器web

2023-01-13 00:02:41

2023-01-05 09:17:58

2024-04-07 09:41:18

SignalR實(shí)時(shí)通信開(kāi)發(fā)

2022-12-25 10:47:52

2017-07-11 13:58:10

WebSocket

2018-04-20 09:36:23

NettyWebSocket京東

2023-09-19 15:33:50

Web實(shí)時(shí)消息推送
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號(hào)