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

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

前言

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

Netty框架的优势

1. API使用简单，开发门槛低；

 2. 功能强大，预置了多种编解码功能，支持多种主流协议；

 3. 定制能力强，可以通过ChannelHandler对通信框架进行灵活地扩展；

 4. 性能高，通过与其他业界主流的NIO框架对比，Netty的综合性能最优；

 5. 成熟、稳定，Netty修复了已经发现的所有JDK NIO BUG，业务开发人员不需要再为NIO的BUG而烦恼

提示：以下是本篇文章正文内容，下面案例可供参考

一、引入netty依赖

io-ty

netty-all

4.1.48.Final

二、使用步骤

1.引入基础配置类

package com.test-ty;

public enum Cmd {

START("000", "连接成功"),

WMESSAGE("001", "消息提醒"),

;

private Stringhttp:// 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-ty服务启动-

package com.test-ty;

import io-ty.bootstrap.ServerBootstrap;

import io-ty.channel.ChannelFuture;

import io-ty.channel.ChannelOption;

import io-ty.channel.EventLoopGroup;

import io-ty.channel.nio.NioEventLoopGroup;

import io-ty.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

*

* 服务启动-

**/

@Slf4j

@Component

public class NettyServer {

@Value("${server-ty.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-ty服务端处理器

package com.test-ty;

import com.test.common.util.jsonUtil;

import io-ty.channel.Channel;

import io-ty.channel.ChannelHandler;

import io-ty.channel.ChannelHandlerContext;

import io-ty.channel.SimpleChannelInboundHandler;

import io-ty.handler.codec.http.websocketx.TextWebSocketFrame;

import io-ty.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-.URLDecoder;

import java.util.*;

/**

* @author test

*

* netty服务端处理器

**/

@Slf4j

@Component

@ChannelHandler.Sharable

public class NettyServerHandler extends SimpleChannelInboundHandler {

@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 channels = cache.get(id);

if (channels == null) {

return;

}

Map 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 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-ty服务端缓存类

package com.test-ty;

import io-ty.channel.Channel;

import io-ty.util.AttributeKey;

import org.springframework.stereotype.Component;

import java.util.HashMap;

import java.util.concurrent.ConcurrentHashMap;

@Component

public class ServerChannelCache {

private static final ConcurrentHashMap> CACHE_MAP = new ConcurrentHashMap<>();

private static final AttributeKey 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 hashMap = CACHE_MAP.get(cacheId);

if (hashMap == null) {

hashMap = new HashMap<>();

}

hashMap.put(channel.id().asShortText(), channel);

CACHE_MAP.put(cacheId, hashMap);

}

public HashMap 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 hashMap = CACHE_MAP.get(cacheId);

if (hashMap == null) {

hashMap = new HashMap<>();

}

hashMap.remove(channel.id().asShortText());

CACHE_MAP.put(cacheId, hashMap);

}

}

5-ty服务初始化器

package com.test-ty;

import io-ty.channel.ChannelInitializer;

import io-ty.channel.ChannelPipeline;

import io-ty.channel.socket.SocketChannel;

import io-ty.handler.codec.http.HttpObjectAggregator;

import io-ty.handler.codec.http.HttpServerCodec;

import io-ty.handler.codec.http.websocketx.WebSocketServerProtocolHandlehttp://r;

import io-ty.handler.stream.ChunkedWriteHandler;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

/**

* @author test

*

* netty服务初始化器

**/

@Component

public class ServerChannelInitializer extends ChannelInitializer {

@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测试

运行 WebSocket

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

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。