Springboot+Netty+Websocket实现消息推送实例

作者:青椒1013 时间:2022-03-24 09:29:29 

前言

WebSocket 使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据。在 WebSocket API 中,浏览器和服务器只需要完成一次握手,两者之间就直接可以创建持久性的连接,并进行双向数据传输。

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);

来源:https://blog.csdn.net/wu_qing_song/article/details/112311860

标签:Springboot,Websocket,消息推送
0
投稿

猜你喜欢

  • Java的“Goto”与标签及使用详解

    2023-11-11 03:56:09
  • Android中通过样式来去除app的头及界面全屏(备忘)的实现方法

    2023-07-30 00:03:23
  • Spring MVC入门_动力节点Java学院整理

    2023-11-03 20:35:31
  • Java中对于双属性枚举的使用案例

    2023-07-20 15:00:09
  • Flutter倒计时/计时器的实现代码

    2023-07-01 03:50:50
  • Java Scala实现数据库增删查改操作详解

    2022-02-03 05:09:22
  • java arrayList遍历的四种方法及Java中ArrayList类的用法

    2023-11-17 17:49:55
  • java 字段值为null,不返回该字段的问题

    2023-07-13 10:32:34
  • Spring Boot 读取静态资源文件的方法

    2023-08-25 02:53:07
  • 详解SpringBoot+SpringSecurity+jwt整合及初体验

    2023-11-28 23:54:48
  • 浅谈java中守护线程与用户线程

    2023-11-26 20:46:41
  • 基于springboot搭建的web系统架构的方法步骤

    2023-11-21 13:33:51
  • Java编程实现对十六进制字符串异或运算代码示例

    2023-11-06 15:58:15
  • 教你怎么用Java数组和链表实现栈

    2023-10-29 08:13:57
  • Java多线程-线程的同步与锁的问题

    2023-11-29 01:40:12
  • 什么是递归?用Java写一个简单的递归程序

    2022-02-11 19:39:45
  • maven环境变量配置讲解

    2023-11-29 12:35:51
  • Java操作Redis2种方法代码详解

    2022-09-10 22:58:56
  • java字节码框架ASM的深入学习

    2023-11-29 05:51:19
  • java实战之飞机大战小游戏(源码加注释)

    2021-08-01 17:46:08
  • asp之家 软件编程 m.aspxhome.com