gitee
/
roses
mirror of https://gitee.com/stylefeng/roses
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

【7.0.4】【socket】整理代码

pull/22/head
fengshuonan 2021-06-12 09:19:00 +08:00
parent 01aa9f1b2e
commit 98a00adcf8
13 changed files with 52 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
kernel-d-socket/socket-api/src/main/java/cn/stylefeng/roses/kernel/socket/api/SocketOperatorApi.java
View File

@ -1,6 +1,5 @@
package cn.stylefeng.roses.kernel.socket.api;
import cn.stylefeng.roses.kernel.socket.api.enums.ServerMessageTypeEnum;
import cn.stylefeng.roses.kernel.socket.api.message.SocketMsgCallbackInterface;
/**
@ -45,4 +44,5 @@ public interface SocketOperatorApi {
* @date 2021/6/2 上午9:54
**/
void msgTypeCallback(String msgType, SocketMsgCallbackInterface callbackInterface);
}

6
kernel-d-socket/socket-api/src/main/java/cn/stylefeng/roses/kernel/socket/api/expander/SocketConfigExpander.java
View File

@ -42,7 +42,7 @@ public class SocketConfigExpander {
* @date 2021/6/7 上午11:39
**/
public static String getSocketHost() {
return ConfigContext.me().getSysConfigValueWithDefault("socket_host", String.class, "0.0.0.0");
return ConfigContext.me().getSysConfigValueWithDefault("SOCKET_HOST", String.class, "0.0.0.0");
}
/**
@ -53,7 +53,7 @@ public class SocketConfigExpander {
* @date 2021/6/7 上午11:41
**/
public static Integer getSocketPort() {
return ConfigContext.me().getSysConfigValueWithDefault("socket_port", Integer.class, 11130);
return ConfigContext.me().getSysConfigValueWithDefault("SOCKET_PORT", Integer.class, 11130);
}
/**
@ -64,7 +64,7 @@ public class SocketConfigExpander {
* @date 2021/6/7 上午11:41
**/
public static Integer getSocketServerChunkSize() {
return ConfigContext.me().getSysConfigValueWithDefault("socket_server_chunk_size", Integer.class, 512 * 1024 * 1024);
return ConfigContext.me().getSysConfigValueWithDefault("SOCKET_SERVER_CHUNK_SIZE", Integer.class, 512 * 1024 * 1024);
}
}

1
kernel-d-socket/socket-api/src/main/java/cn/stylefeng/roses/kernel/socket/api/session/SocketSessionOperatorApi.java
View File

@ -67,4 +67,5 @@ public interface SocketSessionOperatorApi {
* @date 2021/6/1 上午11:50
**/
boolean isInvalid();
}

2
kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/spring/WebSocketApplicationRunnerImpl.java
View File

@ -7,6 +7,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import java.net.StandardSocketOptions;
/**
@ -35,6 +36,7 @@ public class WebSocketApplicationRunnerImpl implements ApplicationRunner {
aioServerConfig.setServerChunkSize(SocketConfigExpander.getSocketServerChunkSize());
// 设置SocketOptions
// 每个套接口都有一个发送缓冲区和一个接收缓冲区，使用SO_RCVBUF可以改变缺省缓冲区大小。
aioServerConfig.setOption(StandardSocketOptions.SO_RCVBUF, 8192);
// 启动

5
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/message/SocketMessageCenter.java
View File

@ -17,8 +17,10 @@ public class SocketMessageCenter {
/**
* 所有消息监听器维护
* <p>
* key是msgType，value是消息回调监听器
*/
private static Map<String, SocketMsgCallbackInterface> messageListenerMap = new HashMap<>();
private static final Map<String, SocketMsgCallbackInterface> messageListenerMap = new HashMap<>();
/**
* 设置消息类型的监听器
@ -43,4 +45,5 @@ public class SocketMessageCenter {
public static SocketMsgCallbackInterface getSocketMsgCallbackInterface(String msgType) {
return messageListenerMap.get(msgType);
}
}

29
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/WebSocketOperator.java
View File

@ -1,20 +1,15 @@
package cn.stylefeng.roses.kernel.socket.websocket.operator;
import cn.hutool.core.util.ObjectUtil;
import cn.stylefeng.roses.kernel.socket.api.message.SocketMsgCallbackInterface;
import cn.stylefeng.roses.kernel.socket.api.SocketOperatorApi;
import cn.stylefeng.roses.kernel.socket.api.enums.ServerMessageTypeEnum;
import cn.stylefeng.roses.kernel.socket.api.exception.SocketException;
import cn.stylefeng.roses.kernel.socket.api.exception.enums.SocketExceptionEnum;
import cn.stylefeng.roses.kernel.socket.websocket.message.WebSocketMessagePOJO;
import cn.stylefeng.roses.kernel.socket.websocket.operator.channel.GettySocketOperator;
import cn.stylefeng.roses.kernel.socket.websocket.message.SocketMessageCenter;
import cn.stylefeng.roses.kernel.socket.api.message.SocketMsgCallbackInterface;
import cn.stylefeng.roses.kernel.socket.api.session.pojo.SocketSession;
import cn.stylefeng.roses.kernel.socket.websocket.message.SocketMessageCenter;
import cn.stylefeng.roses.kernel.socket.websocket.operator.channel.GettySocketOperator;
import cn.stylefeng.roses.kernel.socket.websocket.pojo.WebSocketMessageDTO;
import cn.stylefeng.roses.kernel.socket.websocket.session.SessionCenter;
import com.alibaba.fastjson.JSON;
import com.gettyio.expansion.handler.codec.websocket.frame.TextWebSocketFrame;
import java.util.List;
/**
* WebSocket操作实现类
@ -33,21 +28,21 @@ public class WebSocketOperator implements SocketOperatorApi {
if (ObjectUtil.isEmpty(socketSession)) {
throw new SocketException(SocketExceptionEnum.SESSION_NOT_EXIST);
}
WebSocketMessagePOJO webSocketMessagePOJO = new WebSocketMessagePOJO();
webSocketMessagePOJO.setData(msg);
webSocketMessagePOJO.setServerMsgType(msgType);
WebSocketMessageDTO webSocketMessageDTO = new WebSocketMessageDTO();
webSocketMessageDTO.setData(msg);
webSocketMessageDTO.setServerMsgType(msgType);
// 发送内容
socketSession.getSocketOperatorApi().writeAndFlush(webSocketMessagePOJO);
socketSession.getSocketOperatorApi().writeAndFlush(webSocketMessageDTO);
}
@Override
public void sendMsgOfAllUserSession(String msgType, Object msg) {
for (SocketSession<GettySocketOperator> socketSession : SessionCenter.getSocketSessionMap().values()) {
WebSocketMessagePOJO webSocketMessagePOJO = new WebSocketMessagePOJO();
webSocketMessagePOJO.setData(msg);
webSocketMessagePOJO.setServerMsgType(msgType);
WebSocketMessageDTO webSocketMessageDTO = new WebSocketMessageDTO();
webSocketMessageDTO.setData(msg);
webSocketMessageDTO.setServerMsgType(msgType);
// 发送内容
socketSession.getSocketOperatorApi().writeAndFlush(webSocketMessagePOJO);
socketSession.getSocketOperatorApi().writeAndFlush(webSocketMessageDTO);
}
}

1
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/channel/GettySocketOperator.java
View File

@ -1,6 +1,5 @@
package cn.stylefeng.roses.kernel.socket.websocket.operator.channel;
import cn.stylefeng.roses.kernel.socket.websocket.message.WebSocketMessagePOJO;
import com.alibaba.fastjson.JSON;
import com.gettyio.core.channel.SocketChannel;
import com.gettyio.expansion.handler.codec.websocket.frame.TextWebSocketFrame;

10
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/message/WebSocketMessagePOJO.java → kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/pojo/WebSocketMessageDTO.java
View File

@ -1,4 +1,4 @@
package cn.stylefeng.roses.kernel.socket.websocket.message;
package cn.stylefeng.roses.kernel.socket.websocket.pojo;
import cn.stylefeng.roses.kernel.socket.api.SocketOperatorApi;
import lombok.Data;
@ -9,15 +9,16 @@ import lombok.Data;
* 特殊说明一下serverMsgType和clientMsgType的区别
* 1.serverMsgType字段是服务端发送给客户端的字段
* 例如：服务端发送一个系统消息(type:100001),客户端接收到该消息以后判断需不需要处理，不需要处理跳过即可
* <p>
* 2.clientMsgType字段是客户端发送给服务器的字段
* 例如：客户端发送给服务器一个心跳消息(type:299999)，服务端如果需要处理该消息就注册一个该消息的监听器，
* 那么收到消息服务端会把消息推送给对应的监听器，接口见{@link SocketOperatorApi#msgTypeCallback}
* 那么收到消息服务端会把消息推送给对应的监听器，接口见{@link SocketOperatorApi#msgTypeCallback}
*
* @author majianguo
* @date 2021/6/1 下午2:56
*/
@Data
public class WebSocketMessagePOJO {
public class WebSocketMessageDTO {
/**
* 服务端发送的消息类型(客户端如果需要监听该消息类型，注册对应的消息处理器即可)
@ -40,7 +41,8 @@ public class WebSocketMessagePOJO {
private String formUserId;
/**
* 数据
* 具体发送的数据
*/
private Object data;
}

2
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/WebSocketInitializer.java
View File

@ -27,6 +27,6 @@ public class WebSocketInitializer extends ChannelInitializer {
// 添加自定义的消息处理器
pipeline.addLast(new WebSocketMessageHandler());
}
}

7
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/bind/ChannelIdAndUserBindCenter.java
View File

@ -19,13 +19,15 @@ public class ChannelIdAndUserBindCenter {
/**
* 通道和用户绑定关系映射
* <p>
* key是channelId通道id，value是userId
*/
private static ConcurrentMap<String, String> channelIdAndUserBind = new ConcurrentHashMap<>();
private static final ConcurrentMap<String, String> channelIdAndUserBind = new ConcurrentHashMap<>();
/**
* 等待绑定的通道
*/
private static List<SocketChannel> waitingBindList = Collections.synchronizedList(new ArrayList<>());
private static final List<SocketChannel> waitingBindList = Collections.synchronizedList(new ArrayList<>());
/**
* 获取通道ID
@ -95,4 +97,5 @@ public class ChannelIdAndUserBindCenter {
waitingBindList.removeIf(item -> item.getChannelId().equals(channelId));
channelIdAndUserBind.remove(channelId);
}
}

29
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/handler/WebSocketMessageHandler.java
View File

@ -3,12 +3,12 @@ package cn.stylefeng.roses.kernel.socket.websocket.server.handler;
import cn.hutool.core.util.ObjectUtil;
import cn.stylefeng.roses.kernel.socket.api.enums.ClientMessageTypeEnum;
import cn.stylefeng.roses.kernel.socket.api.message.SocketMsgCallbackInterface;
import cn.stylefeng.roses.kernel.socket.websocket.message.SocketMessageCenter;
import cn.stylefeng.roses.kernel.socket.api.session.pojo.SocketSession;
import cn.stylefeng.roses.kernel.socket.websocket.session.SessionCenter;
import cn.stylefeng.roses.kernel.socket.websocket.server.bind.ChannelIdAndUserBindCenter;
import cn.stylefeng.roses.kernel.socket.websocket.message.SocketMessageCenter;
import cn.stylefeng.roses.kernel.socket.websocket.operator.channel.GettySocketOperator;
import cn.stylefeng.roses.kernel.socket.websocket.message.WebSocketMessagePOJO;
import cn.stylefeng.roses.kernel.socket.websocket.pojo.WebSocketMessageDTO;
import cn.stylefeng.roses.kernel.socket.websocket.server.bind.ChannelIdAndUserBindCenter;
import cn.stylefeng.roses.kernel.socket.websocket.session.SessionCenter;
import com.alibaba.fastjson.JSON;
import com.gettyio.core.channel.SocketChannel;
import com.gettyio.core.pipeline.in.SimpleChannelInboundHandler;
@ -52,10 +52,10 @@ public class WebSocketMessageHandler extends SimpleChannelInboundHandler<WebSock
String data = new String(webSocketFrame.getPayloadData(), StandardCharsets.UTF_8);
// 转换为Java对象
WebSocketMessagePOJO webSocketMessagePOJO = JSON.toJavaObject(JSON.parseObject(data), WebSocketMessagePOJO.class);
WebSocketMessageDTO webSocketMessageDTO = JSON.toJavaObject(JSON.parseObject(data), WebSocketMessageDTO.class);
// 心跳包
if (ClientMessageTypeEnum.USER_HEART.getCode().equals(webSocketMessagePOJO.getClientMsgType())) {
if (ClientMessageTypeEnum.USER_HEART.getCode().equals(webSocketMessageDTO.getClientMsgType())) {
// 更新用户最后活跃时间
String userId = ChannelIdAndUserBindCenter.getUserId(socketChannel.getChannelId());
if (ObjectUtil.isNotEmpty(userId)) {
@ -65,17 +65,17 @@ public class WebSocketMessageHandler extends SimpleChannelInboundHandler<WebSock
}
// 用户ID为空不处理直接跳过
if (ObjectUtil.isEmpty(webSocketMessagePOJO.getFormUserId())) {
if (ObjectUtil.isEmpty(webSocketMessageDTO.getFormUserId())) {
return;
}
// 维护通道和用户ID的绑定关系
if (!ChannelIdAndUserBindCenter.isBind(webSocketMessagePOJO.getFormUserId())) {
ChannelIdAndUserBindCenter.bind(socketChannel.getChannelId(), webSocketMessagePOJO.getFormUserId());
if (!ChannelIdAndUserBindCenter.isBind(webSocketMessageDTO.getFormUserId())) {
ChannelIdAndUserBindCenter.bind(socketChannel.getChannelId(), webSocketMessageDTO.getFormUserId());
// 创建api的会话对象
SocketSession<GettySocketOperator> socketSession = new SocketSession<>();
socketSession.setUserId(webSocketMessagePOJO.getFormUserId());
socketSession.setUserId(webSocketMessageDTO.getFormUserId());
socketSession.setSocketOperatorApi(new GettySocketOperator(socketChannel));
socketSession.setConnectionTime(System.currentTimeMillis());
@ -84,20 +84,21 @@ public class WebSocketMessageHandler extends SimpleChannelInboundHandler<WebSock
}
// 更新最后会话时间
SocketSession<GettySocketOperator> userSession = SessionCenter.getSessionByUserId(webSocketMessagePOJO.getFormUserId());
SocketSession<GettySocketOperator> userSession = SessionCenter.getSessionByUserId(webSocketMessageDTO.getFormUserId());
userSession.setLastActiveTime(System.currentTimeMillis());
// 找到该消息的处理器
SocketMsgCallbackInterface socketMsgCallbackInterface = SocketMessageCenter.getSocketMsgCallbackInterface(webSocketMessagePOJO.getClientMsgType());
SocketMsgCallbackInterface socketMsgCallbackInterface = SocketMessageCenter.getSocketMsgCallbackInterface(webSocketMessageDTO.getClientMsgType());
if (ObjectUtil.isNotEmpty(socketMsgCallbackInterface)) {
// 获取会话
SocketSession<GettySocketOperator> session = SessionCenter.getSessionByUserId(webSocketMessagePOJO.getFormUserId());
SocketSession<GettySocketOperator> session = SessionCenter.getSessionByUserId(webSocketMessageDTO.getFormUserId());
// 触发回调
socketMsgCallbackInterface.callback(webSocketMessagePOJO.getClientMsgType(), webSocketMessagePOJO, session);
socketMsgCallbackInterface.callback(webSocketMessageDTO.getClientMsgType(), webSocketMessageDTO, session);
} else {
socketChannel.writeAndFlush(new TextWebSocketFrame("{\"code\":\"404\"}"));
}
}
}
}

5
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/session/SessionCenter.java
View File

@ -19,12 +19,11 @@ public class SessionCenter {
/**
* 所有会话维护
*/
private static ConcurrentMap<String, SocketSession<GettySocketOperator>> socketSessionMap = new ConcurrentHashMap<>();
private static final ConcurrentMap<String, SocketSession<GettySocketOperator>> socketSessionMap = new ConcurrentHashMap<>();
/**
* 获取维护的所有会话
*
* @return {@link ConcurrentMap< String, SocketSession<GettySocketOperator>>}
* @author majianguo
* @date 2021/6/1 下午2:13
**/
@ -36,7 +35,6 @@ public class SessionCenter {
* 根据用户ID获取会话详情
*
* @param userId 用户ID
* @return {@link SocketSession <GettySocketOperator>}
* @author majianguo
* @date 2021/6/1 下午1:48
**/
@ -65,4 +63,5 @@ public class SessionCenter {
public static void closed(String userId) {
socketSessionMap.remove(userId);
}
}

1
kernel-d-socket/socket-spring-boot-starter/src/main/java/cn/stylefeng/roses/kernel/socket/starter/GunsSocketAutoConfiguration.java
View File

@ -42,7 +42,6 @@ public class GunsSocketAutoConfiguration {
/**
* Socket操作实现类
*
* @return {@link cn.stylefeng.roses.kernel.socket.api.SocketOperatorApi}
* @author majianguo
* @date 2021/6/2 下午5:48
**/