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服务端处理器

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

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

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

相关文章

快网idc优惠网

相关文章

微信

快网idc优惠网

QQ交流群