From f3d3ab1267f35a6d4b22b7a34f5f94bb733cd513 Mon Sep 17 00:00:00 2001
From: zengh <123456>
Date: Wed, 07 Apr 2021 17:39:43 +0800
Subject: [PATCH] 点对点websocket实现
---
blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/controller/PushMsgController.java | 30 ++++++++++
blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/WebSocketHandler.java | 24 +++++++
blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/service/IPushMsgService.java | 21 +++++++
blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyServer/NettyConfig.java | 33 +++++++++++
blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/service/impl/PushMsgServiceImpl.java | 29 +++++++++
5 files changed, 136 insertions(+), 1 deletions(-)
diff --git a/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyServer/NettyConfig.java b/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyServer/NettyConfig.java
index ad82647..cbe126b 100644
--- a/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyServer/NettyConfig.java
+++ b/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyServer/NettyConfig.java
@@ -1,7 +1,10 @@
package org.springblade.jfpt.nettyServer;
+import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
+
+import java.util.concurrent.ConcurrentHashMap;
public class NettyConfig {
@@ -9,4 +12,34 @@
* 存储每一个客户端接入进来时的channel对象
*/
public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
+
+ /**
+ * 定义一个channel组，管理所有channel
+ * GlobalEventExecutor.INSTANCE 是全局的事件执行器，是一个单例
+ */
+ private static ChannelGroup channelGroup = new DefaultChannelGroup("用户管理组",GlobalEventExecutor.INSTANCE);
+
+ /**
+ * 存放用户与chanel 的对应的信息，用于给指定用户发送信息
+ */
+ private static ConcurrentHashMap<String, Channel> userChannelMap = new ConcurrentHashMap<>();
+
+ public NettyConfig() {
+ }
+
+ /**
+ * 获取用户channel 组
+ * @return
+ */
+ public static ChannelGroup getChannelGroup() {
+ return channelGroup;
+ }
+
+ /**
+ * 获取用户channel map
+ * @return
+ */
+ public static ConcurrentHashMap<String, Channel> getUserChannelMap() {
+ return userChannelMap;
+ }
}
diff --git a/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/WebSocketHandler.java b/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/WebSocketHandler.java
index 3d6b0cd..07ac3c2 100644
--- a/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/WebSocketHandler.java
+++ b/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/WebSocketHandler.java
@@ -14,9 +14,11 @@
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
+import io.netty.util.AttributeKey;
import io.netty.util.CharsetUtil;
import java.util.Date;
+import org.springblade.jfpt.nettyServer.NettyConfig;
import org.springblade.jfpt.suser.service.ISuserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -69,7 +71,10 @@
//工作状态(0闲置，1工作中)
String workSt = "0";
webSocketHandler.suserService.updateUser(num,name,workSt);
- ChannelSupervise.removeChannel(ctx.channel());
+ //ChannelSupervise.removeChannel(ctx.channel());
+
+ NettyConfig.getChannelGroup().remove(ctx.channel());
+ removeUserId(ctx);
}
}
@@ -100,6 +105,13 @@
String request = ((TextWebSocketFrame) frame).text();
if (!request.equals("ping")){
+
+ NettyConfig.getUserChannelMap().put(request,ctx.channel());
+
+ //将用户id作为自定义属性加入到channel 中，方便随时channel中获取用户id
+ AttributeKey<String> key = AttributeKey.valueOf("userId");
+ ctx.channel().attr(key).setIfAbsent(request);
+
//把用户信息添加到通道里
ChannelSupervise.addChannel(ctx.channel(),request);
//用户在线状态
@@ -169,4 +181,14 @@
ctx.close();
}
+ /**
+ * 删除用户与channel 对应关系
+ * @param ctx
+ */
+ private void removeUserId(ChannelHandlerContext ctx){
+ AttributeKey<String> key = AttributeKey.valueOf("userId");
+ String userId = ctx.channel().attr(key).get();
+ NettyConfig.getUserChannelMap().remove(userId);
+ }
+
}
diff --git a/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/controller/PushMsgController.java b/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/controller/PushMsgController.java
new file mode 100644
index 0000000..99612ff
--- /dev/null
+++ b/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/controller/PushMsgController.java
@@ -0,0 +1,30 @@
+package org.springblade.jfpt.webscoket.controller;
+
+import org.springblade.jfpt.webscoket.service.IPushMsgService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author lq
+ * @date 2020/4/1 11:22
+ */
+@RestController
+public class PushMsgController {
+
+ @Autowired
+ private IPushMsgService pushMsgService;
+
+ @PostMapping("/pushUser")
+ public String pushUser(String userId,String msg){
+ pushMsgService.pushMsg(userId, msg);
+ return "消息发送成功："+msg;
+ }
+
+ @PostMapping("/pushAll")
+ public String pushAll(String msg){
+ pushMsgService.pushMsg(msg);
+ return "消息发送成功："+msg;
+ }
+
+}
diff --git a/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/service/IPushMsgService.java b/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/service/IPushMsgService.java
new file mode 100644
index 0000000..7b1a94f
--- /dev/null
+++ b/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/service/IPushMsgService.java
@@ -0,0 +1,21 @@
+package org.springblade.jfpt.webscoket.service;
+
+/**
+ * @author 123456
+ */
+public interface IPushMsgService {
+
+ /**
+ * 给指定用户发送消息
+ * @param userId
+ * @param msg
+ */
+ void pushMsg(String userId,String msg);
+
+ /**
+ * 给所有用户发送消息
+ * @param msg
+ */
+ void pushMsg(String msg);
+
+}
diff --git a/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/service/impl/PushMsgServiceImpl.java b/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/service/impl/PushMsgServiceImpl.java
new file mode 100644
index 0000000..d81b37d
--- /dev/null
+++ b/blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/service/impl/PushMsgServiceImpl.java
@@ -0,0 +1,29 @@
+package org.springblade.jfpt.webscoket.service.impl;
+
+import io.netty.channel.Channel;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import org.springblade.jfpt.nettyServer.NettyConfig;
+import org.springblade.jfpt.webscoket.service.IPushMsgService;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author lq
+ * @date 2020/4/1 11:20
+ */
+@Service
+public class PushMsgServiceImpl implements IPushMsgService {
+ @Override
+ public void pushMsg(String userId, String msg) {
+ Channel channel = NettyConfig.getUserChannelMap().get(userId);
+ channel.writeAndFlush(new TextWebSocketFrame(msg));
+ }
+
+ @Override
+ public void pushMsg(String msg) {
+ ChannelGroup group = NettyConfig.getChannelGroup();
+ String name = group.name();
+ System.out.println("空间大小："+group.size()+",名字："+name);
+ group.writeAndFlush(new TextWebSocketFrame(msg));
+ }
+}
--
Gitblit v1.9.3