• 基于Netty实现的简单聊天服务组件

    目录

    基于Netty实现的简单聊天服务组件

    本文摘自Quan后台管理服务框架中的quan-chat工具，该工具仅实现了非常简单服务模型。后期本人会视情况扩展更多复杂的业务场景。

    如果本文对您解决问题有帮助，欢迎到Gitee或Github点个star 🤝

    quan-chat 是一个基于 Netty 实现的服务端即时消息通讯组件，组件本身不具备业务处理能力，主要的作用是提供服务端消息中转； 通过实现组件中的接口可以完成与项目相关的业务功能， 例如：点对点消息收发、权限校验、聊天记录保存等。

    web展示层ui基于layim。layim展示的功能较为丰富。为演示服务组件，仅实现点对点聊天功能。其它功能视情况扩展。

    本组件仅用于学习交流使用，本文应用到的 layim 来自互联网，如果您想将 layim 框架用于其它用途，必须取得原作者授权: layui ，否则产生的一切法律责任与本作者无关。

    效果展示

    在这里插入图片描述

    技术选型：

    spring-boot-2.7.16
    netty-4.1.97
    layim-3.9.8

    功能分析

    1. 聊天服务基础设施配置（基于Netty）
    2. 用户上线、下线处理
    3. 用户消息发送、接收处理
    4. 用户登录凭证校验

    完整的组件代码开源地址：https://gitee.com/quan100/quan/tree/main/quan-tools/quan-chat
    下面仅展示部分代码

    聊天服务基础设施配置（基于Netty）

    Netty 是一个异步事件驱动的网络应用框架，用于快速开发可维护的高性能服务器和客户端。

    定义组件基础的配置（ChatProperties）

    ChatProperties 主要用于定义组件内部使用到的配置参数。

    package cn.javaquan.tools.chat.autoconfigure;

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.util.Assert;

/**
 * Configuration properties for im support.
 *
 * @author javaquan
 * @since 1.0.0
 */
@ConfigurationProperties(prefix = "quan.im")
public class ChatProperties {

    /**
     * 默认数据包最大长度
     * 64kb
     */
    private final static int MAX_FRAME_SIZE = 65536;

    /**
     * 默认的消息体最大长度
     * 64kb
     */
    private final static int MAX_CONTENT_LENGTH = 65536;

    /**
     * 空闲检查时间，单位：秒
     */
    private final static long READER_IDLE_TIME = 600L;

    /**
     * 开启IM服务的端口
     */
    private Integer port;

    /**
     * SSL配置
     */
    private Ssl ssl;

    /**
     * websocket 路径
     */
    private String websocketPath;

    /**
     * 数据包最大长度
     * 单位：字节
     */
    private Integer maxFrameSize;

    /**
     * 消息体最大长度
     * 单位：字节
     */
    private Integer maxContentLength;

    /**
     * 允许连接空闲的最大时间
     * 
    
     * 当空闲超过最大时间后，强制下线
     */
    private Long readerIdleTime;

    public Integer getPort() {
        return port;
    }

    public void setPort(Integer port) {
        this.port = port;
    }

    public int determineDefaultPort() {
        Assert.notNull(this.port, "[Assertion failed chat server port] - this numeric argument must have value; it must not be null");
        return this.port;
    }

    public Ssl getSsl() {
        return ssl;
    }

    public void setSsl(Ssl ssl) {
        this.ssl = ssl;
    }

    public String getWebsocketPath() {
        return websocketPath;
    }

    public void setWebsocketPath(String websocketPath) {
        this.websocketPath = websocketPath;
    }

    public String determineDefaultWebsocketPath() {
        Assert.hasText(this.websocketPath, "[Assertion failed chat server websocketPath] - it must not be null or empty");
        return this.websocketPath;
    }

    public Integer getMaxFrameSize() {
        return maxFrameSize;
    }

    public void setMaxFrameSize(Integer maxFrameSize) {
        this.maxFrameSize = maxFrameSize;
    }

    public Integer determineDefaultMaxFrameSize() {
        if (null == maxFrameSize) {
            this.setMaxFrameSize(MAX_FRAME_SIZE);
        }
        return this.maxFrameSize;
    }

    public Integer getMaxContentLength() {
        return maxContentLength;
    }

    public void setMaxContentLength(Integer maxContentLength) {
        this.maxContentLength = maxContentLength;
    }

    public Integer determineDefaultMaxContentLength() {
        if (null == maxContentLength) {
            this.setMaxContentLength(MAX_CONTENT_LENGTH);
        }
        return this.maxContentLength;
    }

    public Long getReaderIdleTime() {
        return readerIdleTime;
    }

    public void setReaderIdleTime(Long readerIdleTime) {
        this.readerIdleTime = readerIdleTime;
    }

    public Long determineDefaultReaderIdleTime() {
        if (null == readerIdleTime) {
            this.setReaderIdleTime(READER_IDLE_TIME);
        }
        return this.readerIdleTime;
    }

    /**
     * ssl properties.
     */
    public static class Ssl {

        private boolean enabled = false;
        private String protocol = "TLS";

        /**
         * an X.509 certificate chain file in PEM format
         */
        private String keyCertChainFilePath;

        /**
         * a PKCS#8 private key file in PEM format
         */
        private String keyFilePath;

        public boolean isEnabled() {
            return enabled;
        }

        public void setEnabled(boolean enabled) {
            this.enabled = enabled;
        }

        public String getProtocol() {
            return protocol;
        }

        public void setProtocol(String protocol) {
            this.protocol = protocol;
        }

        public String getKeyCertChainFilePath() {
            return keyCertChainFilePath;
        }

        public void setKeyCertChainFilePath(String keyCertChainFilePath) {
            this.keyCertChainFilePath = keyCertChainFilePath;
        }

        public String determineDefaultKeyCertChainFilePath() {
            Assert.hasText(this.keyCertChainFilePath, "[Assertion failed chat server keyCertChainFilePath] - it must not be null or empty");
            return this.keyCertChainFilePath;
        }

        public String getKeyFilePath() {
            return keyFilePath;
        }

        public void setKeyFilePath(String keyFilePath) {
            this.keyFilePath = keyFilePath;
        }

        public String determineDefaultKeyFilePath() {
            Assert.hasText(this.keyFilePath, "[Assertion failed chat server keyFilePath] - it must not be null or empty");
            return this.keyFilePath;
        }

    }

    public void afterPropertiesSet() {
        determineDefaultPort();
        determineDefaultWebsocketPath();
        determineDefaultMaxFrameSize();
        determineDefaultMaxContentLength();
        determineDefaultReaderIdleTime();
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124
    • 125
    • 126
    • 127
    • 128
    • 129
    • 130
    • 131
    • 132
    • 133
    • 134
    • 135
    • 136
    • 137
    • 138
    • 139
    • 140
    • 141
    • 142
    • 143
    • 144
    • 145
    • 146
    • 147
    • 148
    • 149
    • 150
    • 151
    • 152
    • 153
    • 154
    • 155
    • 156
    • 157
    • 158
    • 159
    • 160
    • 161
    • 162
    • 163
    • 164
    • 165
    • 166
    • 167
    • 168
    • 169
    • 170
    • 171
    • 172
    • 173
    • 174
    • 175
    • 176
    • 177
    • 178
    • 179
    • 180
    • 181
    • 182
    • 183
    • 184
    • 185
    • 186
    • 187
    • 188
    • 189
    • 190
    • 191
    • 192
    • 193
    • 194
    • 195
    • 196
    • 197
    • 198
    • 199
    • 200
    • 201
    • 202
    • 203
    • 204
    • 205
    • 206
    • 207
    • 208
    • 209
    • 210
    • 211
    • 212
    • 213
    • 214
    • 215

    yml 配置示例：

    quan: 
  im:
    port: 10000   # 配置chat服务端口
    websocket-path: /chat   # 配置chat服务websocket访问的uri
    reader-idle-time: 1800 #允许连接空闲的时间，单位：秒。超时后强制下线

    • 1
    • 2
    • 3
    • 4
    • 5
    定义聊天服务类（ChatServer）

    用于实现客户端与服务器建立连接，状态维护

    package cn.javaquan.tools.chat.server;

import cn.javaquan.tools.chat.autoconfigure.ChatProperties;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.ImmediateEventExecutor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.util.Assert;

import java.net.InetSocketAddress;

/**
 * 默认的聊天服务
 *
 * @author javaquan
 * @since 1.0.0
 */
public class ChatServer {

    private static final Log logger = LogFactory.getLog(ChatServer.class);

    private final ChannelGroup channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
    private final EventLoopGroup group = new NioEventLoopGroup();
    private Channel channel;

    public ChannelFuture start(InetSocketAddress address, ChatProperties properties) {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(group)
                .channel(NioServerSocketChannel.class)
                .childHandler(createInitializer(channelGroup, properties));
        ChannelFuture future = bootstrap.bind(address);
        future.syncUninterruptibly();
        channel = future.channel();
        return future;
    }

    protected ChannelInitializer<Channel> createInitializer(ChannelGroup group, ChatProperties properties) {
        return new ChatServerInitializer(group, properties);
    }

    public void destroy() {
        if (channel != null) {
            channel.close();
        }
        channelGroup.close();
        group.shutdownGracefully();
    }

    public void start(ChatProperties properties) {
        ChannelFuture future = this.start(new InetSocketAddress(properties.getPort()), properties);
        addShutdownHook(this);
        future.addListener((listener) -> {
            Assert.isTrue(listener.isSuccess(), logMessageFormat(properties.getPort(), "error"));
            logger.info(logMessageFormat(properties.getPort(), "success"));
        });
    }

    /**
     * Registers a new virtual-machine shutdown hook.
     *
     * @param chatServer
     */
    private void addShutdownHook(ChatServer chatServer) {
        Runtime.getRuntime().addShutdownHook(new Thread(chatServer::destroy));
    }

    private String logMessageFormat(Integer port, String state) {
        return String.format("%s started %s on port(s): %s", this.getClass().getSimpleName(), state, port);
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    定义聊天服务配置初始化类（ChatServerInitializer）

    主要用于初始化聊天服务应用到的处理器。

    package cn.javaquan.tools.chat.server;

import cn.javaquan.tools.chat.autoconfigure.ChatProperties;
import cn.javaquan.tools.chat.context.ClientInboundHandler;
import cn.javaquan.tools.chat.context.TextWebSocketFrameHandler;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.group.ChannelGroup;
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 io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

/**
 * 初始化服务配置
 *
 * @author javaquan
 */
public class ChatServerInitializer extends ChannelInitializer<Channel> {
    private final ChannelGroup group;
    private final ChatProperties properties;

    public ChatServerInitializer(ChannelGroup group, ChatProperties properties) {
        this.group = group;
        this.properties = properties;
    }

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new HttpObjectAggregator(properties.getMaxContentLength()));
        pipeline.addLast(new IdleStateHandler(properties.getReaderIdleTime(), 0, 0, TimeUnit.SECONDS));

        pipeline.addLast(new ClientInboundHandler(group, properties.getWebsocketPath()));
        pipeline.addLast(new TextWebSocketFrameHandler());

        pipeline.addLast(new WebSocketServerProtocolHandler(properties.getWebsocketPath(), null, true, properties.getMaxFrameSize()));
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46

    用户上线、下线处理

    客户端绑定服务处理类（ClientInboundHandler）

    主要用于处理用户上线、下线状态处理。

    package cn.javaquan.tools.chat.context;

import cn.javaquan.tools.chat.core.ChannelPool;
import cn.javaquan.tools.chat.core.support.AuthorizationProcessor;
import cn.javaquan.tools.chat.util.SpringUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;

/**
 * 客户端用户状态处理
 *
 * @author javaquan
 */
@Sharable
public class ClientInboundHandler extends ChannelInboundHandlerAdapter {

    private static final Log logger = LogFactory.getLog(ClientInboundHandler.class);

    private final ChannelGroup group;
    private final String websocketPath;

    public ClientInboundHandler(ChannelGroup group, String websocketPath) {
        this.group = group;
        this.websocketPath = websocketPath;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpRequest) {
            FullHttpRequest request = (FullHttpRequest) msg;
            String uri = request.uri();
            Map<String, String> queryParams = paramsParser(uri);
            online(ctx.channel(), queryParams);
            request.setUri(websocketPath);
        }
        super.channelRead(ctx, msg);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                logger.info(String.format("用户[%s]闲置时间超过最大值，将关闭连接！", ChannelPool.getSessionState(ctx.channel())));
                ctx.channel().close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        group.add(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        group.remove(channel);
        offline(channel);
    }

    /**
     * 异常时调用
     *
     * @param ctx
     * @param cause
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error("服务器错误", cause);
        offline(ctx.channel());
        // 发生异常之后关闭连接（关闭channel）
        ctx.channel().close();
    }

    /**
     * url参数解析
     *
     * @param uriParams
     * @return
     * @throws URISyntaxException
     */
    private Map<String, String> paramsParser(String uriParams) throws URISyntaxException {
        URI uri = new URI(uriParams);
        Map<String, String> paramsMap = new HashMap<>();

        String queryParam = uri.getQuery();
        String[] queryParams = queryParam.split("&");

        for (String param : queryParams) {
            String[] urlParam = param.split("=");
            paramsMap.put(urlParam[0], urlParam[1]);
        }

        return paramsMap;
    }

    /**
     * 用户上线
     *
     * @param channel
     * @param urlParams url参数
     */
    private void online(Channel channel, Map<String, String> urlParams) {
        String userId = urlParams.get("userId");
        String authorization = urlParams.get("authorization");
        AuthorizationProcessor authorizationProcessor = SpringUtils.getBean(AuthorizationProcessor.class);

        if (!authorizationProcessor.checkAuth(authorization)) {
            channel.close();
            logger.info(String.format("用户[%s]凭证校验失败，连接被服务器拒绝", userId));
            return;
        }

        logger.info(String.format("用户[%s]上线", userId));

        channel.attr(ChannelPool.SESSION_STATE).set(userId);
        ChannelPool.addChannel(userId, channel);

        /// TODO 若用户上线，则通知好友已上线。kafka发送上线事件
    }

    /**
     * 用户离线
     *
     * @param channel
     */
    private void offline(Channel channel) {
        ChannelPool.removeChannel(channel);

        logger.info(String.format("用户[%s]下线", ChannelPool.getSessionState(channel)));

        /// TODO 若用户下线，则通知好友已下线。kafka发送下线事件
    }

}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124
    • 125
    • 126
    • 127
    • 128
    • 129
    • 130
    • 131
    • 132
    • 133
    • 134
    • 135
    • 136
    • 137
    • 138
    • 139
    • 140
    • 141
    • 142
    • 143
    • 144
    • 145
    • 146
    • 147
    • 148
    • 149
    • 150
    • 151
    • 152

    用户消息发送、接收处理

    定义一个文本消息处理器（TextWebSocketFrameHandler）

    用于将用户发送的文本消息转换为服务端使用的模版消息。
    通过模版将消息转发给接收者。

    package cn.javaquan.tools.chat.context;

import cn.javaquan.tools.chat.core.MessageHandlerFactory;
import cn.javaquan.tools.chat.core.message.MessageTemplate;
import cn.javaquan.tools.chat.util.JsonUtils;
import cn.javaquan.tools.chat.util.SpringUtils;
import cn.javaquan.tools.chat.core.support.IMessageHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

/**
 * 消息处理器
 *
 * @author javaquan
 */
public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    @Override
    public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        MessageTemplate messageTemplate = messageConvertor(msg);
        messageHandler(ctx, messageTemplate);
    }

    /**
     * 消息处理
     * 
    
     * 根据消息类型处理消息
     * 
    
     * 需要自定义实现{@link IMessageHandler}接口。
     *
     * @param ctx
     * @param messageTemplate
     */
    private void messageHandler(ChannelHandlerContext ctx, MessageTemplate messageTemplate) {
        MessageHandlerFactory messageHandlerFactory = SpringUtils.getBean(MessageHandlerFactory.class);
        messageHandlerFactory.getService(messageTemplate.getType()).handler(ctx, messageTemplate);
    }

    /**
     * 将字符串信息转换为模版信息格式
     *
     * @param msg
     * @return
     */
    private MessageTemplate messageConvertor(TextWebSocketFrame msg) {
        return JsonUtils.parseObject(msg.text(), MessageTemplate.class);
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50

    用户登录凭证校验

    定义一个凭证处理器接口（AuthorizationProcessor）

    将处理器定义成接口，主要目的是将组件与业务解耦。
    因为不同的业务，实现的权限业务都可能不一样。
    只需业务端实现该接口，当权限校验不通过时，组件内部就会拒绝客户端连接。

    package cn.javaquan.tools.chat.core.support;


/**
 * 授权凭证处理器
 *
 * @author wangquan
 */
public interface AuthorizationProcessor {

    /**
     * 检查权限
     *
     * @param authorization 登录凭证
     * @return
     */
    boolean checkAuth(String authorization);

}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20

    定义 ChatAutoConfiguration 自动化配置类

    ChatAutoConfiguration 是 quan-chat 组件中最重要的一项配置，通过该配置来定义组件是否生效。
    当引入 quan-chat 组件时，不需要对组件进行扫描。服务启动时会自动发现该配置。
    通过该配置初始化聊天服务所依赖的相关功能。若未按照配置要求配置属性，quan-chat 组件引入将无效。

    package cn.javaquan.tools.chat.autoconfigure;

import cn.javaquan.tools.chat.ChatServerApplication;
import cn.javaquan.tools.chat.core.ChannelPool;
import cn.javaquan.tools.chat.core.support.AbstractAuthorizationCheckProcessor;
import cn.javaquan.tools.chat.core.support.AuthorizationProcessor;
import cn.javaquan.tools.chat.server.ChatServer;
import cn.javaquan.tools.chat.server.SecureChatServer;
import io.netty.channel.Channel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.AnyNestedCondition;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;

import java.io.File;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * im聊天sdk配置
 *
 * @author javaquan
 * @since 1.0.0
 */
@AutoConfiguration
@EnableConfigurationProperties(ChatProperties.class)
public class ChatAutoConfiguration {

    @Import(ChatServerApplication.class)
    @Configuration(proxyBeanMethods = false)
    @Conditional(ChatCondition.class)
    protected static class ChatConfiguration {

        @ConditionalOnProperty(prefix = "quan.im.ssl", name = "enabled", havingValue = "false", matchIfMissing = true)
        @ConditionalOnMissingBean
        @Bean
        ChatServer chatServer() {
            return new ChatServer();
        }

        @ConditionalOnMissingBean
        @Bean
        ChannelPool channelPool() {
            Map<String, Channel> channelContainer = new ConcurrentHashMap<>();
            return new ChannelPool(channelContainer);
        }

        @ConditionalOnMissingBean
        @Bean
        AuthorizationProcessor authorizationProcessor() {
            return new AbstractAuthorizationCheckProcessor();
        }

    }

    static class ChatCondition extends AnyNestedCondition {

        ChatCondition() {
            super(ConfigurationPhase.PARSE_CONFIGURATION);
        }

        @ConditionalOnProperty(prefix = "quan.im", name = "port")
        static class PortProperty {

        }

        @ConditionalOnProperty(prefix = "quan.im.ssl", name = "enabled", havingValue = "true")
        @ConditionalOnMissingBean
        @Bean
        SslContext sslContext(ChatProperties properties) throws Exception {
            ChatProperties.Ssl ssl = properties.getSsl();
            File keyCertChainFile = new File(ssl.determineDefaultKeyCertChainFilePath());
            File keyFile = new File(ssl.determineDefaultKeyFilePath());
            return SslContextBuilder.forServer(keyCertChainFile, keyFile).build();
        }

        @ConditionalOnProperty(prefix = "quan.im.ssl", name = "enabled", havingValue = "true")
        @ConditionalOnMissingBean
        @Bean
        ChatServer secureChatServer(SslContext context) {
            return new SecureChatServer(context);
        }
    }

}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93

    定义 ChatServerApplication 服务启动类

    当引入 quan-chat 组件时，并正确配置 ChatProperties 属性，服务启动时则会自动扫描 ChatServerApplication 类，用于启动 聊天服务端。

    package cn.javaquan.tools.chat;

import cn.javaquan.tools.chat.autoconfigure.ChatProperties;
import cn.javaquan.tools.chat.server.ChatServer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;

/**
 * chat服务启动
 *
 * @author javaquan
 * @since 1.0.0
 */
public class ChatServerApplication implements ApplicationRunner {

    @Autowired
    private ChatServer chatServer;
    @Autowired
    private ChatProperties properties;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        properties.afterPropertiesSet();
        chatServer.start(properties);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27

    参考资料

    如果本文对您解决问题有帮助，欢迎到Gitee或Github点个star 🤝

    quan-chat 工具文档：https://doc.javaquan.cn/pages/tools/chat/
    quan-chat 工具开源地址：https://gitee.com/quan100/quan/tree/main/quan-tools/quan-chat

  • 相关阅读:
    React - setState 原理
    基于Flask_admin库，编写个人工作平台详述。
    深度学习目标检测模型综述
    Tailwindcss 提取组件
    C程序设计（谭浩强）第五版课后题答案 第一章
    R语言编写自定义函数对数据进行标准化、使用keras包构建深度学习自动编码器（autoencoder）、使用MSE指标评估自动编码器的效能（重构误差）
    Git的初步认识
    最长字段和问题
    从 1.5 开始搭建一个微服务框架——日志追踪 traceId
    html静态网站基于动漫主题网站网页设计与实现共计10个页面——二次元漫画
  • 原文地址：https://blog.csdn.net/w_quan/article/details/134495134