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

Netty框架的優(yōu)勢

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

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

<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ù)啟動監(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ù)啟動監(jiān)聽器 **/@Slf4j@Componentpublic 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) //設(shè)置隊列大小 .option(ChannelOption.SO_BACKLOG, 1024) // 兩小時內(nèi)沒有數(shù)據(jù)的通信時,TCP會自動發(fā)送一個活動探測數(shù)據(jù)報文 .childOption(ChannelOption.SO_KEEPALIVE, true); //綁定端口,開始接收進(jìn)來的連接 try { ChannelFuture future = bootstrap.bind(port).sync(); log.info('服務(wù)器啟動開始監(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.Sharablepublic class NettyServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> { @Autowired private ServerChannelCache cache; private static final String dataKey = 'test='; @Data public static class ChannelCache { } /** * 客戶端連接會觸發(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ā)消息會觸發(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;@Componentpublic 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ù)初始化器 **/@Componentpublic 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() 方法發(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測試

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.服務(wù)器下發(fā)消息

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

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