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); } }, }, 8.服务器下发消息
@Autowired private NettyServerHandler nettyServerHandler; nettyServerHandler.send(CmdWeb.WMESSAGE, id, message);
看完上述内容，你们掌握Springboot+Netty+Websocket实现消息推送实例是怎样的的方法了吗？如果还想学到更多技能或想了解更多相关内容，欢迎关注亿速云行业资讯频道，感谢各位的阅读！
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
