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

前言

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

到此这篇关于Springboot+Netty+Websocket实现消息推送实例的文章就介绍到这了,更多相关Springboot Websocket消息推送内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • Java中websocket消息推送的实现代码

    一.服务层 package com.demo.websocket; import java.io.IOException; import java.util.Iterator; import java.util.concurrent.ConcurrentLinkedQueue; import javax.servlet.ServletContextEvent; import javax.servlet.ServletContextListener; import org.springframew

  • 详解在Spring Boot框架下使用WebSocket实现消息推送

    spring Boot的学习持续进行中.前面两篇博客我们介绍了如何使用Spring Boot容器搭建Web项目以及怎样为我们的Project添加HTTPS的支持,在这两篇文章的基础上,我们今天来看看如何在Spring Boot中使用WebSocket. 什么是WebSocket WebSocket为浏览器和服务器之间提供了双工异步通信功能,也就是说我们可以利用浏览器给服务器发送消息,服务器也可以给浏览器发送消息,目前主流浏览器的主流版本对WebSocket的支持都算是比较好的,但是在实际开发中使

  • Android中使用WebSocket实现群聊和消息推送功能(不使用WebView)

    WebSocket protocol 是HTML5一种新的协议.它实现了浏览器与服务器全双工通信(full-duplex).WebSocket是Web2.0时代的新产物,用于弥补HTTP协议的某些不足,不过他们之间真实的关系是兄弟关系,都是对socket的进一步封装,其目前最直观的表现就是服务器推送和聊天功能.更多知识参考:如何理解 TCP/IP, SPDY, WebSocket 三者之间的关系? 今天的重点是讲如何在Android中脱离WebView使用WebSocket,而不是在Web浏览器

  • SpringBoot+WebSocket+Netty实现消息推送的示例代码

    上一篇文章讲了Netty的理论基础,这一篇讲一下Netty在项目中的应用场景之一:消息推送功能,可以满足给所有用户推送,也可以满足给指定某一个用户推送消息,创建的是SpringBoot项目,后台服务端使用Netty技术,前端页面使用WebSocket技术. 大概实现思路: 前端使用webSocket与服务端创建连接的时候,将用户ID传给服务端 服务端将用户ID与channel关联起来存储,同时将channel放入到channel组中 如果需要给所有用户发送消息,直接执行channel组的writ

  • php实现websocket实时消息推送

    php实现websocket实时消息推送,供大家参考,具体内容如下 SocketService.php <?php /** * Created by xwx * Date: 2017/10/18 * Time: 14:33 */ class SocketService { private $address = '0.0.0.0'; private $port = 8083; private $_sockets; public function __construct($address = '',

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

    前言 WebSocket 使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据.在 WebSocket API 中,浏览器和服务器只需要完成一次握手,两者之间就直接可以创建持久性的连接,并进行双向数据传输. Netty框架的优势 1. API使用简单,开发门槛低:  2. 功能强大,预置了多种编解码功能,支持多种主流协议:  3. 定制能力强,可以通过ChannelHandler对通信框架进行灵活地扩展:  4. 性能高,通过与其他业界主流的NIO框架对比,Netty的综

  • SpringBoot+WebSocket实现消息推送功能

    目录 背景 WebSocket简介 协议原理 WebSocket与HTTP协议的区别 WebSocket特点 应用场景 系统集成Websocket jar包引入 Websocket配置 具体实现 测试示例 页面请求websocket 测试效果 背景 项目中经常会用到消息推送功能,关于推送技术的实现,我们通常会联想到轮询.comet长连接技术,虽然这些技术能够实现,但是需要反复连接,对于服务资源消耗过大,随着技术的发展,HtML5定义了WebSocket协议,能更好的节省服务器资源和带宽,并且能够

  • SpringBoot+Netty+WebSocket实现消息发送的示例代码

    一.导入Netty依赖 <dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>4.1.25.Final</version> </dependency> 二.搭建websocket服务器 @Component public class WebSocketServer { /** * 主线程池 */

  • SpringBoot+netty-socketio实现服务器端消息推送

    首先:因为工作需要,需要对接socket.io框架对接,所以目前只能使用netty-socketio.websocket是不支持对接socket.io框架的. netty-socketio顾名思义他是一个底层基于netty'实现的socket. 在springboot项目中的集成,请看下面的代码 maven依赖 <dependency> <groupId>com.corundumstudio.socketio</groupId> <artifactId>ne

  • java后端+前端使用WebSocket实现消息推送的详细流程

    目录 前言 创建WebSocket的简单实例操作流程 1.引入Websocket依赖 2.创建配置类WebSocketConfig 3.创建WebSocketServer 4.websocket调用 总结 前言 在项目的开发时,遇到实现服务器主动发送数据到前端页面的功能的需求.实现该功能不外乎使用轮询和websocket技术,但在考虑到实时性和资源损耗后,最后决定使用websocket.现在就记录一下用Java实现Websocket技术吧~ Java实现Websocket通常有两种方式:1.创建

  • Java应用层协议WebSocket实现消息推送

    目录 前言 浏览器端 服务器端 前言   大部分的web开发者,开发的业务都是基于Http协议的:前端请求后端接口,携带参数,后端执行业务代码,再返回结果给前端.作者参与开发的项目,有一个报警推送的功能,服务端实时推送报警信息给浏览器端:还有像抖音里面,如果有人关注.回复你的评论时,抖音就会推送相关消息给你了,你就会收到一条消息.   有些同学会说了,基于Http协议也能实现啊:前端定时访问后端(每隔3s或者几秒),后端返回消息数据,前端拿到后弹出消息.这种方式太low了,而且每个浏览器都这样,

  • SpringMVC整合websocket实现消息推送及触发功能

    本文为大家分享了SpringMVC整合websocket实现消息推送,供大家参考,具体内容如下 1.创建websocket握手协议的后台 (1)HandShake的实现类 /** *Project Name: price *File Name: HandShake.java *Package Name: com.yun.websocket *Date: 2016年9月3日 下午4:44:27 *Copyright (c) 2016,578888218@qq.com All Rights Rese

  • SpringBoot整合WxJava开启消息推送的实现

    目录 1.引入 WxJava 依赖 2.申请微信小程序 3.微信小程序配置信息 4.消息推送配置 5.接收消息推送的 API 6.消息推送测试 接入微信小程序消息推送服务,可以3种方式选择其一: 1.开发者服务器接收消息推送2.云函数接收消息推送3.微信云托管服务接收消息推送 开发者服务器接收消息推送,开发者需要按照如下步骤完成: 1.填写服务器配置2.验证服务器地址的有效性3.据接口文档实现业务逻辑,接收消息和事件 1.引入 WxJava 依赖 <!-- web支持 --> <depe

随机推荐