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

新增Socket模块，默认实现WebSocket

pull/22/head
rays 2021-06-02 17:14:23 +08:00
parent 2b8ec525b6
commit 7858b13c29
29 changed files with 1298 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
kernel-d-socket/pom.xml Normal file
View File

@ -0,0 +1,37 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>cn.stylefeng.roses</groupId>
<artifactId>roses-kernel</artifactId>
<version>7.0.4</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>kernel-d-socket</artifactId>
<packaging>pom</packaging>
<modules>
<module>socket-api</module>
<module>socket-sdk-websocket</module>
<module>socket-business-websocket</module>
<module>socket-spring-boot-starter</module>
</modules>
<dependencies>
<!-- 开发规则 -->
<dependency>
<groupId>cn.stylefeng.roses</groupId>
<artifactId>kernel-a-rule</artifactId>
<version>7.0.4</version>
</dependency>
</dependencies>
</project>

1
kernel-d-socket/socket-api/README.md Normal file
View File

@ -0,0 +1 @@
socket模块的api

41
kernel-d-socket/socket-api/pom.xml Normal file
View File

@ -0,0 +1,41 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>cn.stylefeng.roses</groupId>
<artifactId>kernel-d-socket</artifactId>
<version>7.0.4</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>socket-api</artifactId>
<packaging>jar</packaging>
<dependencies>
<!--config模块的api-->
<dependency>
<groupId>cn.stylefeng.roses</groupId>
<artifactId>config-api</artifactId>
<version>7.0.4</version>
</dependency>
<!-- Getty核心包 -->
<dependency>
<groupId>com.gettyio</groupId>
<artifactId>getty-core</artifactId>
</dependency>
<!-- 拓展包 -->
<dependency>
<groupId>com.gettyio</groupId>
<artifactId>getty-expansion</artifactId>
</dependency>
</dependencies>
</project>

24
kernel-d-socket/socket-api/src/main/java/cn/stylefeng/roses/kernel/socket/api/SocketMsgCallbackInterface.java Normal file
View File

@ -0,0 +1,24 @@
package cn.stylefeng.roses.kernel.socket.api;
import cn.stylefeng.roses.kernel.socket.api.session.pojo.SocketSession;
/**
* Socket消息接收回调接口
*
* @author majianguo
* @date 2021/6/2 上午9:53
*/
@FunctionalInterface
public interface SocketMsgCallbackInterface {
/**
* 收到消息的回调
*
* @param msgType 消息类型
* @param msg 消息体
* @param socketSession 本次通信的会话
* @author majianguo
* @date 2021/6/2 上午9:51
**/
void callback(String msgType, Object msg, SocketSession socketSession);
}

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

@ -0,0 +1,43 @@
package cn.stylefeng.roses.kernel.socket.api;
/**
* Socket通用操作类
* <p>
* 可通过该类直接发送消息，每一个Socket实现的子模块必须实现该接口，以提供统一的操作API
*
* @author majianguo
* @date 2021/6/2 上午9:25
*/
public interface SocketOperatorApi {
/**
* 发送消息到指定会话
*
* @param sessionId 会话ID(会话ID具体看业务的实现，WebSocket中使用的是用户传入的formId)
* @param msg 消息体
* @author majianguo
* @date 2021/6/2 上午9:35
**/
void sendMsgOfSession(String sessionId, Object msg);
/**
* 发送消息到所有会话
*
* @param msg 消息体
* @author majianguo
* @date 2021/6/2 上午9:35
**/
void sendMsgOfAllSession(Object msg);
/**
* 监听指定类型消息
* <p>
* 1.该方法每调用一次即注册一个监听,同一个消息类型多次调用只有最后一次生效
*
* @param msgType 消息类型
* @param callbackInterface 消息监听器
* @author majianguo
* @date 2021/6/2 上午9:54
**/
void msgTypeCallback(String msgType, SocketMsgCallbackInterface callbackInterface);
}

60
kernel-d-socket/socket-api/src/main/java/cn/stylefeng/roses/kernel/socket/api/constants/SocketConstants.java Normal file
View File

@ -0,0 +1,60 @@
/*
* Copyright [2020-2030] [https://www.stylefeng.cn]
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* Guns采用APACHE LICENSE 2.0开源协议，您在使用过程中，需要注意以下几点：
*
* 1.请不要删除和修改根目录下的LICENSE文件。
* 2.请不要删除和修改Guns源码头部的版权声明。
* 3.请保留源码和相关描述文件的项目出处，作者声明等。
* 4.分发源码时候，请注明软件出处 https://gitee.com/stylefeng/guns
* 5.在修改包名，模块名称，项目代码等时，请注明软件出处 https://gitee.com/stylefeng/guns
* 6.若您的项目无法满足以上几点，可申请商业授权
*/
package cn.stylefeng.roses.kernel.socket.api.constants;
/**
* socket 模块常量
*
* @author majianguo
* @date 2021/6/1 上午11:21
*/
public interface SocketConstants {
/**
* socket模块的名称
*/
String SOCKET_MODULE_NAME = "kernel-d-socket";
/**
* 异常枚举的步进值
*/
String SOCKET_EXCEPTION_STEP_CODE = "30";
/**
* Socket监听地址
*/
String SOCKET_HOST = "socket_host";
/**
* Socket监听端口
*/
String SOCKET_PORT = "socket_port";
/**
* Socket服务器内存池最大可分配空间大小
*/
String SOCKET_SERVER_CHUNK_SIZE = "socket_server_chunk_size";
}

47
kernel-d-socket/socket-api/src/main/java/cn/stylefeng/roses/kernel/socket/api/exception/SocketException.java Normal file
View File

@ -0,0 +1,47 @@
/*
* Copyright [2020-2030] [https://www.stylefeng.cn]
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* Guns采用APACHE LICENSE 2.0开源协议，您在使用过程中，需要注意以下几点：
*
* 1.请不要删除和修改根目录下的LICENSE文件。
* 2.请不要删除和修改Guns源码头部的版权声明。
* 3.请保留源码和相关描述文件的项目出处，作者声明等。
* 4.分发源码时候，请注明软件出处 https://gitee.com/stylefeng/guns
* 5.在修改包名，模块名称，项目代码等时，请注明软件出处 https://gitee.com/stylefeng/guns
* 6.若您的项目无法满足以上几点，可申请商业授权
*/
package cn.stylefeng.roses.kernel.socket.api.exception;
import cn.stylefeng.roses.kernel.rule.exception.AbstractExceptionEnum;
import cn.stylefeng.roses.kernel.rule.exception.base.ServiceException;
import cn.stylefeng.roses.kernel.socket.api.constants.SocketConstants;
/**
* Socket模块异常
*
* @author majianguo
* @date 2021/6/1 上午11:23
*/
public class SocketException extends ServiceException {
public SocketException(AbstractExceptionEnum exception) {
super(SocketConstants.SOCKET_MODULE_NAME, exception);
}
public SocketException(String errorCode, String userTip) {
super(SocketConstants.SOCKET_MODULE_NAME, errorCode, userTip);
}
}

68
kernel-d-socket/socket-api/src/main/java/cn/stylefeng/roses/kernel/socket/api/exception/enums/SocketExceptionEnum.java Normal file
View File

@ -0,0 +1,68 @@
/*
* Copyright [2020-2030] [https://www.stylefeng.cn]
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* Guns采用APACHE LICENSE 2.0开源协议，您在使用过程中，需要注意以下几点：
*
* 1.请不要删除和修改根目录下的LICENSE文件。
* 2.请不要删除和修改Guns源码头部的版权声明。
* 3.请保留源码和相关描述文件的项目出处，作者声明等。
* 4.分发源码时候，请注明软件出处 https://gitee.com/stylefeng/guns
* 5.在修改包名，模块名称，项目代码等时，请注明软件出处 https://gitee.com/stylefeng/guns
* 6.若您的项目无法满足以上几点，可申请商业授权
*/
package cn.stylefeng.roses.kernel.socket.api.exception.enums;
import cn.stylefeng.roses.kernel.rule.constants.RuleConstants;
import cn.stylefeng.roses.kernel.rule.exception.AbstractExceptionEnum;
import cn.stylefeng.roses.kernel.socket.api.constants.SocketConstants;
import lombok.Getter;
/**
* Socket模块相关异常枚举
*
* @author majianguo
* @date 2021/6/1 上午11:25
*/
@Getter
public enum SocketExceptionEnum implements AbstractExceptionEnum {
/**
* Socket操作异常
*/
SOCKET_ERROR(RuleConstants.THIRD_ERROR_TYPE_CODE + SocketConstants.SOCKET_EXCEPTION_STEP_CODE + "01", "操作异常，具体信息为：{}"),
/**
* 会话不存在
*/
SESSION_NOT_EXIST(RuleConstants.THIRD_ERROR_TYPE_CODE + SocketConstants.SOCKET_EXCEPTION_STEP_CODE + "02", "会话不存在"),
;
/**
* 错误编码
*/
private final String errorCode;
/**
* 提示用户信息
*/
private final String userTip;
SocketExceptionEnum(String errorCode, String userTip) {
this.errorCode = errorCode;
this.userTip = userTip;
}
}

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

@ -0,0 +1,70 @@
/*
* Copyright [2020-2030] [https://www.stylefeng.cn]
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* Guns采用APACHE LICENSE 2.0开源协议，您在使用过程中，需要注意以下几点：
*
* 1.请不要删除和修改根目录下的LICENSE文件。
* 2.请不要删除和修改Guns源码头部的版权声明。
* 3.请保留源码和相关描述文件的项目出处，作者声明等。
* 4.分发源码时候，请注明软件出处 https://gitee.com/stylefeng/guns
* 5.在修改包名，模块名称，项目代码等时，请注明软件出处 https://gitee.com/stylefeng/guns
* 6.若您的项目无法满足以上几点，可申请商业授权
*/
package cn.stylefeng.roses.kernel.socket.api.session;
/**
* socket会话操作接口
* <p>
* 该接口面向会话，须基于会话的通道调用。
* 该接口支持扩展，可参考WebSocket模块中{@link cn.stylefeng.roses.kernel.socket.websocket.channel}包下的类
*
* @author majianguo
* @date 2021/6/1 上午11:46
*/
public interface SocketSessionOperatorApi {
/**
* 写出数据，经过责任链
*
* @author majianguo
* @date 2021/6/1 上午11:48
**/
void writeAndFlush(Object obj);
/**
* 写出数据，不经过责任链
*
* @author majianguo
* @date 2021/6/1 上午11:48
**/
void writeToChannel(Object obj);
/**
* 关闭会话
*
* @author majianguo
* @date 2021/6/1 上午11:48
**/
void close();
/**
* 是否存活
*
* @return {@link boolean}
* @author majianguo
* @date 2021/6/1 上午11:50
**/
boolean isInvalid();
}

47
kernel-d-socket/socket-api/src/main/java/cn/stylefeng/roses/kernel/socket/api/session/pojo/SocketSession.java Normal file
View File

@ -0,0 +1,47 @@
package cn.stylefeng.roses.kernel.socket.api.session.pojo;
import cn.stylefeng.roses.kernel.socket.api.session.SocketSessionOperatorApi;
import lombok.Data;
import java.util.Set;
/**
* Socket会话
*
* @author majianguo
* @date 2021/6/1 上午11:28
*/
@Data
public class SocketSession<T extends SocketSessionOperatorApi> {
/**
* 会话唯一标识
*/
private String sessionId;
/**
* 该会话所有的监听消息类型
*/
private Set<String> messageTypes;
/**
* 连接时间
*/
private Long connectionTime;
/**
* 最后活跃时间
*/
private Long lastActiveTime;
/**
* 操作API
*/
private T socketOperatorApi;
/**
* 自定义数据
*/
private Object data;
}

31
kernel-d-socket/socket-business-websocket/pom.xml Normal file
View File

@ -0,0 +1,31 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>cn.stylefeng.roses</groupId>
<artifactId>kernel-d-socket</artifactId>
<version>7.0.4</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>socket-business-websocket</artifactId>
<dependencies>
<!--Socket模块的Websocket实现-->
<dependency>
<groupId>cn.stylefeng.roses</groupId>
<artifactId>socket-sdk-websocket</artifactId>
<version>7.0.4</version>
</dependency>
<!--web模块-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
</dependencies>
</project>

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

@ -0,0 +1,52 @@
package cn.stylefeng.roses.kernel.socket.business.websocket.spring;
import cn.stylefeng.roses.kernel.config.api.constants.ConfigConstants;
import cn.stylefeng.roses.kernel.config.api.context.ConfigContext;
import cn.stylefeng.roses.kernel.socket.websocket.server.WebSocketServer;
import com.gettyio.core.channel.config.ServerConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import java.net.StandardSocketOptions;
import static cn.stylefeng.roses.kernel.socket.api.constants.SocketConstants.*;
/**
* Spring Boot启动完成拉起WebSocket
*
* @author majianguo
* @date 2021/6/2 上午11:06
*/
@Component
@Slf4j
public class WebSocketApplicationRunnerImpl implements ApplicationRunner {
@Override
public void run(ApplicationArguments args) throws Exception {
// 初始化配置对象
ServerConfig aioServerConfig = new ServerConfig();
// 设置host,不设置默认0.0.0.0
String socketHost = ConfigContext.me().getSysConfigValueWithDefault(SOCKET_HOST, String.class, "0.0.0.0");
aioServerConfig.setHost(socketHost);
// 设置端口号
Integer socketPort = ConfigContext.me().getSysConfigValueWithDefault(SOCKET_PORT, Integer.class, 11130);
aioServerConfig.setPort(socketPort);
// 设置服务器端内存池最大可分配空间大小，默认512mb，内存池空间可以根据吞吐量设置。
// 尽量可以设置大一点，因为这不会真正的占用系统内存，只有真正使用时才会分配
Integer socketServerChunkSize = ConfigContext.me().getSysConfigValueWithDefault(SOCKET_SERVER_CHUNK_SIZE, Integer.class, 512 * 1024 * 1024);
aioServerConfig.setServerChunkSize(socketServerChunkSize);
// 设置SocketOptions
aioServerConfig.setOption(StandardSocketOptions.SO_RCVBUF, 8192);
// 启动
WebSocketServer.run(aioServerConfig);
log.info("WebSocket Server Start Success!");
}
}

1
kernel-d-socket/socket-sdk-websocket/README.md Normal file
View File

@ -0,0 +1 @@
socket模块的websocket实现

24
kernel-d-socket/socket-sdk-websocket/pom.xml Normal file
View File

@ -0,0 +1,24 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>cn.stylefeng.roses</groupId>
<artifactId>kernel-d-socket</artifactId>
<version>7.0.4</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>socket-sdk-websocket</artifactId>
<dependencies>
<!--短信模块的api-->
<dependency>
<groupId>cn.stylefeng.roses</groupId>
<artifactId>socket-api</artifactId>
<version>7.0.4</version>
</dependency>
</dependencies>
</project>

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

@ -0,0 +1,46 @@
package cn.stylefeng.roses.kernel.socket.websocket.message;
import cn.stylefeng.roses.kernel.socket.api.SocketMsgCallbackInterface;
import java.util.HashMap;
import java.util.Map;
/**
* 会话消息中心
* <p>
* 维护所有消息类型对应的处理器
*
* @author majianguo
* @date 2021/6/1 下午2:20
*/
public class SocketMessageCenter {
/**
* 所有消息监听器维护
*/
private static Map<String, SocketMsgCallbackInterface> messageListenerMap = new HashMap<>();
/**
* 设置消息类型的监听器
*
* @param msgType 消息类型
* @param listener 监听器
* @author majianguo
* @date 2021/6/1 下午2:25
**/
public static void setMessageListener(String msgType, SocketMsgCallbackInterface listener) {
messageListenerMap.put(msgType, listener);
}
/**
* 获取消息监听器
*
* @param msgType 消息类型
* @return {@link SocketMsgCallbackInterface}
* @author majianguo
* @date 2021/6/1 下午2:26
**/
public static SocketMsgCallbackInterface getSocketMsgCallbackInterface(String msgType) {
return messageListenerMap.get(msgType);
}
}

33
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/message/WebSocketMessagePOJO.java Normal file
View File

@ -0,0 +1,33 @@
package cn.stylefeng.roses.kernel.socket.websocket.message;
import lombok.Data;
/**
* WebSocket交互通用对象
*
* @author majianguo
* @date 2021/6/1 下午2:56
*/
@Data
public class WebSocketMessagePOJO {
/**
* 消息类型
*/
private String type;
/**
* 目标Id
*/
private String toId;
/**
* 发送者ID
*/
private String formId;
/**
* 数据
*/
private String data;
}

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

@ -0,0 +1,54 @@
package cn.stylefeng.roses.kernel.socket.websocket.operator;
import cn.hutool.core.util.ObjectUtil;
import cn.stylefeng.roses.kernel.socket.api.SocketMsgCallbackInterface;
import cn.stylefeng.roses.kernel.socket.api.SocketOperatorApi;
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.operator.channel.GettySocketOperator;
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 java.util.Collection;
/**
* WebSocket操作实现类
* <p>
* 如果是Spring boot项目，通过注入SocketOperatorApi接口操作socket，需将本来交给Spring管理
*
* @author majianguo
* @date 2021/6/2 上午10:41
*/
public class WebSocketOperator implements SocketOperatorApi {
@Override
public void sendMsgOfSession(String sessionId, Object msg) {
// 获取会话
SocketSession<GettySocketOperator> socketSession = SessionCenter.getSessionById(sessionId);
if (ObjectUtil.isEmpty(socketSession)) {
throw new SocketException(SocketExceptionEnum.SESSION_NOT_EXIST);
}
// 发送内容
socketSession.getSocketOperatorApi().writeAndFlush(msg);
}
@Override
public void sendMsgOfAllSession(Object msg) {
// 获取所有会话
Collection<SocketSession<GettySocketOperator>> socketSessions = SessionCenter.getSocketSessionMap().values();
if (ObjectUtil.isNotEmpty(socketSessions)) {
// 给所有会话发送消息
for (SocketSession<?> socketSession : socketSessions) {
// 发送内容
socketSession.getSocketOperatorApi().writeAndFlush(msg);
}
}
}
@Override
public void msgTypeCallback(String msgType, SocketMsgCallbackInterface callbackInterface) {
SocketMessageCenter.setMessageListener(msgType, callbackInterface);
}
}

15
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/channel/GettyChannelExpandInterFace.java Normal file
View File

@ -0,0 +1,15 @@
package cn.stylefeng.roses.kernel.socket.websocket.operator.channel;
import cn.stylefeng.roses.kernel.socket.api.session.SocketSessionOperatorApi;
/**
* 对Api模块的操作类进行扩展
* <p>
* 暂时只写接口，SocketOperatorApi方法不够用时再对此类进行扩展
*
* @author majianguo
* @date 2021/6/1 下午3:44
*/
public interface GettyChannelExpandInterFace extends SocketSessionOperatorApi {
}

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

@ -0,0 +1,56 @@
package cn.stylefeng.roses.kernel.socket.websocket.operator.channel;
import com.gettyio.core.channel.SocketChannel;
import com.gettyio.expansion.handler.codec.websocket.frame.TextWebSocketFrame;
/**
* Socket操作类实现
* <p>
* 这里使用的是Getty,所以对Getty的SocketChannel对象做简单封装
*
* @author majianguo
* @date 2021/6/1 下午3:41
*/
public class GettySocketOperator implements GettyChannelExpandInterFace {
/**
* 实际操作的通道
*/
private SocketChannel socketChannel;
public GettySocketOperator(SocketChannel socketChannel) {
this.socketChannel = socketChannel;
}
@Override
public void writeAndFlush(Object obj) {
if (obj instanceof String) {
// 处理WebSocket的数据
TextWebSocketFrame textWebSocketFrame = new TextWebSocketFrame(obj.toString());
socketChannel.writeAndFlush(textWebSocketFrame);
return;
}
socketChannel.writeAndFlush(obj);
}
@Override
public void writeToChannel(Object obj) {
if (obj instanceof String) {
// 处理WebSocket的数据
TextWebSocketFrame textWebSocketFrame = new TextWebSocketFrame(obj.toString());
socketChannel.writeToChannel(textWebSocketFrame);
return;
}
socketChannel.writeToChannel(obj);
}
@Override
public void close() {
socketChannel.close();
}
@Override
public boolean isInvalid() {
return socketChannel.isInvalid();
}
}

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

@ -0,0 +1,48 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package cn.stylefeng.roses.kernel.socket.websocket.server;
import cn.stylefeng.roses.kernel.socket.websocket.server.handler.WebSocketMessageHandler;
import com.gettyio.core.channel.SocketChannel;
import com.gettyio.core.pipeline.ChannelInitializer;
import com.gettyio.core.pipeline.DefaultChannelPipeline;
import com.gettyio.expansion.handler.codec.websocket.WebSocketDecoder;
import com.gettyio.expansion.handler.codec.websocket.WebSocketEncoder;
/**
* WebSocket通道责任链对象
*
* @author majianguo
* @date 2021/6/1 下午2:36
*/
public class WebSocketInitializer extends ChannelInitializer {
@Override
public void initChannel(SocketChannel channel) throws Exception {
// 获取责任链对象
DefaultChannelPipeline pipeline = channel.getDefaultChannelPipeline();
// 先把ws的编解码器添加到责任链前面。注意，只有先通过ws的编解码器，才能解析ws的消息帧，
// 后续的解码器才能继续解析期望得到的结果
pipeline.addLast(new WebSocketEncoder());
pipeline.addLast(new WebSocketDecoder());
// 添加自定义的消息处理器
pipeline.addLast(new WebSocketMessageHandler());
}
}

70
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/WebSocketServer.java Normal file
View File

@ -0,0 +1,70 @@
package cn.stylefeng.roses.kernel.socket.websocket.server;
import com.gettyio.core.channel.config.ServerConfig;
import com.gettyio.core.channel.starter.AioServerStarter;
import java.net.StandardSocketOptions;
/**
* WebSocket服务端
*
* @author majianguo
* @date 2021/6/1 下午2:40
*/
public class WebSocketServer {
/**
* 无参数启动
*
* @author majianguo
* @date 2021/6/2 上午11:10
**/
public static void start() {
// 初始化配置对象
ServerConfig aioServerConfig = new ServerConfig();
// 设置host,不设置默认localhost
aioServerConfig.setHost("0.0.0.0");
// 设置端口号
aioServerConfig.setPort(11130);
// 设置服务器端内存池最大可分配空间大小，默认256mb，内存池空间可以根据吞吐量设置。
// 尽量可以设置大一点，因为这不会真正的占用系统内存，只有真正使用时才会分配
aioServerConfig.setServerChunkSize(512 * 1024 * 1024);
// 设置数据输出器队列大小，一般不用设置这个参数，默认是10*1024*1024
aioServerConfig.setBufferWriterQueueSize(10 * 1024 * 1024);
// 设置读取缓存块大小，一般不用设置这个参数，默认128字节
aioServerConfig.setReadBufferSize(2048);
// 设置内存池等待分配内存的最大阻塞时间，默认是1秒
aioServerConfig.setChunkPoolBlockTime(2000);
// 设置SocketOptions
aioServerConfig.setOption(StandardSocketOptions.SO_RCVBUF, 8192);
// 启动
run(aioServerConfig);
}
/**
* 启动Socket服务
*
* @param aioServerConfig 服务器配置
* @author majianguo
* @date 2021/6/1 下午2:40
**/
public static void run(ServerConfig aioServerConfig) {
final AioServerStarter starter = new AioServerStarter(aioServerConfig);
starter.channelInitializer(new WebSocketInitializer());
try {
// 启动服务
starter.start();
} catch (Exception e) {
e.printStackTrace();
}
}
}

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

@ -0,0 +1,98 @@
package cn.stylefeng.roses.kernel.socket.websocket.server.bind;
import com.gettyio.core.channel.SocketChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* 通道和用户的绑定中心
*
* @author majianguo
* @date 2021/6/1 下午3:09
*/
public class ChannelIdAndUserBindCenter {
/**
* 通道和用户绑定关系映射
*/
private static ConcurrentMap<String, String> channelIdAndUserBind = new ConcurrentHashMap<>();
/**
* 等待绑定的通道
*/
private static List<SocketChannel> waitingBindList = Collections.synchronizedList(new ArrayList<>());
/**
* 获取通道ID
*
* @param channelId 通道ID
* @return {@link java.lang.String}
* @author majianguo
* @date 2021/6/1 下午3:33
**/
public static String getUserId(String channelId) {
return channelIdAndUserBind.get(channelId);
}
/**
* 添加一个未绑定的通道
*
* @param socketChannel 通道对象
* @author majianguo
* @date 2021/6/1 下午3:17
**/
public static void addSocketChannel(SocketChannel socketChannel) {
waitingBindList.add(socketChannel);
}
/**
* 绑定关系
*
* @param channelId 通道ID
* @param userId 用户ID
* @return {@link boolean}
* @author majianguo
* @date 2021/6/1 下午3:21
**/
public static boolean bind(String channelId, String userId) {
Iterator<SocketChannel> iterator = waitingBindList.iterator();
while (iterator.hasNext()) {
SocketChannel item = iterator.next();
if (item.getChannelId().equals(channelId)) {
channelIdAndUserBind.put(channelId, userId);
iterator.remove();
return true;
}
}
return false;
}
/**
* 用户是否已绑定通道
*
* @param userId 用户ID
* @return {@link boolean}
* @author majianguo
* @date 2021/6/1 下午3:29
**/
public static boolean isBind(String userId) {
return channelIdAndUserBind.containsValue(userId);
}
/**
* 关闭通道
*
* @param channelId 通道ID
* @author majianguo
* @date 2021/6/1 下午3:31
**/
public static void closed(String channelId) {
waitingBindList.removeIf(item -> item.getChannelId().equals(channelId));
channelIdAndUserBind.remove(channelId);
}
}

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

@ -0,0 +1,86 @@
package cn.stylefeng.roses.kernel.socket.websocket.server.handler;
import cn.hutool.core.util.ObjectUtil;
import cn.stylefeng.roses.kernel.socket.api.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.operator.channel.GettySocketOperator;
import cn.stylefeng.roses.kernel.socket.websocket.message.WebSocketMessagePOJO;
import com.alibaba.fastjson.JSON;
import com.gettyio.core.channel.SocketChannel;
import com.gettyio.core.pipeline.in.SimpleChannelInboundHandler;
import com.gettyio.expansion.handler.codec.websocket.frame.TextWebSocketFrame;
import com.gettyio.expansion.handler.codec.websocket.frame.WebSocketFrame;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.StandardCharsets;
/**
* 消息监听处理器
*
* @author majianguo
* @date 2021/6/1 下午2:35
*/
public class WebSocketMessageHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
private final Logger log = LoggerFactory.getLogger(WebSocketMessageHandler.class);
@Override
public void channelAdded(SocketChannel aioChannel) throws Exception {
log.info(aioChannel.getChannelId() + " connection successful.");
ChannelIdAndUserBindCenter.addSocketChannel(aioChannel);
}
@Override
public void channelClosed(SocketChannel aioChannel) throws Exception {
log.info(aioChannel.getChannelId() + " disconnected");
// 获取用户ID
String userId = ChannelIdAndUserBindCenter.getUserId(aioChannel.getChannelId());
if (ObjectUtil.isNotEmpty(userId)) {
// 根据用户ID关闭会话
SessionCenter.closed(userId);
}
ChannelIdAndUserBindCenter.closed(aioChannel.getChannelId());
}
@Override
public void channelRead0(SocketChannel socketChannel, WebSocketFrame webSocketFrame) throws Exception {
if (webSocketFrame instanceof TextWebSocketFrame) {
String data = new String(webSocketFrame.getPayloadData(), StandardCharsets.UTF_8);
// 转换为Java对象
WebSocketMessagePOJO webSocketMessagePOJO = JSON.toJavaObject(JSON.parseObject(data), WebSocketMessagePOJO.class);
// 维护通道和用户ID的绑定关系
if (!ChannelIdAndUserBindCenter.isBind(webSocketMessagePOJO.getFormId())) {
ChannelIdAndUserBindCenter.bind(socketChannel.getChannelId(), webSocketMessagePOJO.getFormId());
// 创建api的会话对象
SocketSession<GettySocketOperator> socketSession = new SocketSession<>();
socketSession.setSessionId(webSocketMessagePOJO.getFormId());
socketSession.setSocketOperatorApi(new GettySocketOperator(socketChannel));
socketSession.setConnectionTime(System.currentTimeMillis());
socketSession.setLastActiveTime(System.currentTimeMillis());
// 维护会话
SessionCenter.addSocketSession(socketSession);
}
// 找到该消息的处理器
SocketMsgCallbackInterface socketMsgCallbackInterface = SocketMessageCenter.getSocketMsgCallbackInterface(webSocketMessagePOJO.getType());
if (ObjectUtil.isNotEmpty(socketMsgCallbackInterface)) {
// 获取会话
SocketSession<GettySocketOperator> session = SessionCenter.getSessionById(webSocketMessagePOJO.getFormId());
// 触发回调
socketMsgCallbackInterface.callback(webSocketMessagePOJO.getType(), webSocketMessagePOJO, session);
} else {
socketChannel.writeAndFlush(new TextWebSocketFrame("{\"code\":\"404\"}"));
}
}
}
}

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

@ -0,0 +1,141 @@
package cn.stylefeng.roses.kernel.socket.websocket.session;
import cn.hutool.core.util.ObjectUtil;
import cn.stylefeng.roses.kernel.socket.api.session.pojo.SocketSession;
import cn.stylefeng.roses.kernel.socket.websocket.operator.channel.GettySocketOperator;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* 会话中心
* <p>
* 维护所有的会话
*
* @author majianguo
* @date 2021/6/1 下午1:43
*/
public class SessionCenter {
/**
* 所有会话维护
*/
private static ConcurrentMap<String, SocketSession<GettySocketOperator>> socketSessionMap = new ConcurrentHashMap<>();
/**
* 消息类型和会话ID关系维护
*/
private static ConcurrentMap<String, List<String>> messageTypeSessionMap = new ConcurrentHashMap<>();
/**
* 获取维护的所有会话
*
* @return {@link ConcurrentMap< String, SocketSession<GettySocketOperator>>}
* @author majianguo
* @date 2021/6/1 下午2:13
**/
public static ConcurrentMap<String, SocketSession<GettySocketOperator>> getSocketSessionMap() {
return socketSessionMap;
}
/**
* 获取消息和会话ID的完整映射关系
*
* @return {@link ConcurrentMap< String, List< String>>}
* @author majianguo
* @date 2021/6/1 下午2:14
**/
public static ConcurrentMap<String, List<String>> getMessageTypeSessionMap() {
return messageTypeSessionMap;
}
/**
* 根据会话ID获取会话详情
*
* @param sessionId 会话ID
* @return {@link SocketSession <GettySocketOperator>}
* @author majianguo
* @date 2021/6/1 下午1:48
**/
public static SocketSession<GettySocketOperator> getSessionById(String sessionId) {
return socketSessionMap.get(sessionId);
}
/**
* 设置会话
*
* @param socketSession 会话详情
* @author majianguo
* @date 2021/6/1 下午1:49
**/
public static void addSocketSession(SocketSession<GettySocketOperator> socketSession) {
// 维护会话
socketSessionMap.put(socketSession.getSessionId(), socketSession);
// 维护会话所有的消息类型和会话的关系
if (ObjectUtil.isNotEmpty(socketSession.getMessageTypes())) {
for (String messageType : socketSession.getMessageTypes()) {
List<String> sessionIds = messageTypeSessionMap.get(messageType);
if (ObjectUtil.isEmpty(sessionIds)) {
sessionIds = new ArrayList<>();
messageTypeSessionMap.put(messageType, sessionIds);
}
sessionIds.add(socketSession.getSessionId());
}
}
}
/**
* 根据消息类型获取所有的会话
*
* @return {@link List< SocketSession<GettySocketOperator>>}
* @author majianguo
* @date 2021/6/1 下午2:06
**/
public static List<SocketSession<GettySocketOperator>> getSocketSessionByMsgType(String msgType) {
List<SocketSession<GettySocketOperator>> res = new ArrayList<>();
// 获取监听该消息所有的会话
List<String> stringList = messageTypeSessionMap.get(msgType);
if (ObjectUtil.isNotEmpty(stringList)) {
for (String sessionId : stringList) {
SocketSession<GettySocketOperator> socketSession = socketSessionMap.get(sessionId);
res.add(socketSession);
}
}
return res;
}
/**
* 给会话添加监听的消息类型
*
* @param msgType 消息类型
* @param sessionId 会话ID
* @author majianguo
* @date 2021/6/1 下午2:11
**/
public static void addSocketSessionMsgType(String msgType, String sessionId) {
SocketSession<GettySocketOperator> socketSession = socketSessionMap.get(sessionId);
if (ObjectUtil.isNotEmpty(socketSession)) {
socketSession.getMessageTypes().add(msgType);
}
}
/**
* 连接关闭
*
* @param sessionId 会话唯一标识
* @author majianguo
* @date 2021/6/1 下午3:25
**/
public static void closed(String sessionId) {
socketSessionMap.remove(sessionId);
for (Map.Entry<String, List<String>> stringListEntry : messageTypeSessionMap.entrySet()) {
stringListEntry.getValue().removeIf(item -> item.equals(sessionId));
}
}
}

1
kernel-d-socket/socket-spring-boot-starter/README.md Normal file
View File

@ -0,0 +1 @@
Socket的spring boot自动加载模块

29
kernel-d-socket/socket-spring-boot-starter/pom.xml Normal file
View File

@ -0,0 +1,29 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>cn.stylefeng.roses</groupId>
<artifactId>kernel-d-socket</artifactId>
<version>7.0.4</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>socket-spring-boot-starter</artifactId>
<packaging>jar</packaging>
<dependencies>
<!-- WebSocket模块 -->
<dependency>
<groupId>cn.stylefeng.roses</groupId>
<artifactId>socket-business-websocket</artifactId>
<version>7.0.4</version>
</dependency>
</dependencies>
</project>

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

@ -0,0 +1,55 @@
/*
* Copyright [2020-2030] [https://www.stylefeng.cn]
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* Guns采用APACHE LICENSE 2.0开源协议，您在使用过程中，需要注意以下几点：
*
* 1.请不要删除和修改根目录下的LICENSE文件。
* 2.请不要删除和修改Guns源码头部的版权声明。
* 3.请保留源码和相关描述文件的项目出处，作者声明等。
* 4.分发源码时候，请注明软件出处 https://gitee.com/stylefeng/guns
* 5.在修改包名，模块名称，项目代码等时，请注明软件出处 https://gitee.com/stylefeng/guns
* 6.若您的项目无法满足以上几点，可申请商业授权
*/
package cn.stylefeng.roses.kernel.socket.starter;
import cn.stylefeng.roses.kernel.socket.api.SocketOperatorApi;
import cn.stylefeng.roses.kernel.socket.websocket.operator.WebSocketOperator;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* Socket的自动配置类
*
* @author fengshuonan
* @date 2020/12/1 21:18
*/
@Configuration
public class GunsSocketAutoConfiguration {
/**
* Socket操作实现类
*
* @return {@link SocketOperatorApi}
* @author majianguo
* @date 2021/6/2 上午11:02
**/
@Bean
@ConditionalOnMissingBean(SocketOperatorApi.class)
public SocketOperatorApi socketOperatorApi() {
return new WebSocketOperator();
}
}

2
kernel-d-socket/socket-spring-boot-starter/src/main/resources/META-INF/spring.factories Normal file
View File

@ -0,0 +1,2 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
cn.stylefeng.roses.kernel.socket.starter.GunsSocketAutoConfiguration

18
pom.xml
View File

@ -103,6 +103,9 @@
<!-- 分布式事务seata -->
<module>kernel-d-seata</module>
<!-- Socket模块 -->
<module>kernel-d-socket</module>
</modules>
<properties>
@ -133,6 +136,7 @@
<groovy.version>3.0.7</groovy.version>
<oshi.version>5.7.1</oshi.version>
<beetl.version>3.3.2.RELEASE</beetl.version>
<getty.version>1.4.9</getty.version>
</properties>
<dependencyManagement>
@ -292,6 +296,20 @@
<version>${oshi.version}</version>
</dependency>
<!--getty核心包-->
<dependency>
<groupId>com.gettyio</groupId>
<artifactId>getty-core</artifactId>
<version>${getty.version}</version>
</dependency>
<!-- 拓展包 -->
<dependency>
<groupId>com.gettyio</groupId>
<artifactId>getty-expansion</artifactId>
<version>${getty.version}</version>
</dependency>
</dependencies>
</dependencyManagement>