# Tomcat
server:
  port: 9201
  netty:
    port: 10201
    application:
      name: yhy-netty-server

# Spring
spring:
  application:
    # 應(yīng)用名稱
    name: soc-dmoasp-system
  profiles:
    # 環(huán)境配置
    active: dev
  cloud:
    nacos:
      discovery:
        # 服務(wù)注冊(cè)地址
        server-addr: nacos-registry:8858
      config:
        # 配置中心地址
        server-addr: nacos-registry:8858
        file-extension: yml
        # 共享配置
        shared-configs:
          - data-id: application.${spring.cloud.nacos.config.file-extension}
            refresh: true
          - data-id: soc-dmoasp-redission.${spring.cloud.nacos.config.file-extension}
          - data-id: soc-dmoasp-druid.${spring.cloud.nacos.config.file-extension}

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-reactor-netty</artifactId>
</dependency>

import com.alibaba.cloud.nacos.NacosDiscoveryProperties;
import com.alibaba.nacos.api.PropertyKeyConst;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.soc.dmoasp.system.server.handler.WebSocketIdleStateHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.net.InetAddress;
import java.util.Properties;

/**
 * Netty服務(wù)
 * @author dsw
 * @date 2024/11/18 17:34
 */
@Component
public class NettyServer implements CommandLineRunner {

    Logger log = LoggerFactory.getLogger(NettyServer.class);

    @Autowired
    private NacosDiscoveryProperties nacosDiscoveryProperties;


    @Value("${server.netty.port}")
    private Integer nettyPort;
    @Value("${server.netty.application.name}")
    private String nettyName;

    EventLoopGroup bossGroup;

    EventLoopGroup workGroup;

    @Override
    public void run(String... args) throws Exception {
        log.info("初始化netty配置開(kāi)始");
        //netty 服務(wù)端啟動(dòng)的端口不可和Springboot啟動(dòng)類(lèi)的端口號(hào)重復(fù)
        this.start();
        //關(guān)閉服務(wù)器的時(shí)候同時(shí)關(guān)閉Netty服務(wù)
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                this.destroy();
            } catch (InterruptedException e) {
                log.error(e.getMessage());
            }
        }));
    }

    @Async
    public void start() throws InterruptedException {
        try {
            bossGroup = new NioEventLoopGroup(1);
            workGroup = new NioEventLoopGroup(10);
            ServerBootstrap bootstrap = new ServerBootstrap();
            // bossGroup輔助客戶端的tcp連接請(qǐng)求, workGroup負(fù)責(zé)與客戶端之前的讀寫(xiě)操作
            bootstrap.group(bossGroup, workGroup)
            // 指定Channel
            .channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline
                                    //HTTP解碼
                                    .addLast(new HttpServerCodec())
                                    .addLast(new ChunkedWriteHandler())
                                    //HTTP段聚合
                                    .addLast(new HttpObjectAggregator(1024*1024))
                                    //將HTTP協(xié)議轉(zhuǎn)成ws協(xié)議
                                    .addLast(new WebSocketServerProtocolHandler("/socket"))

                            ;

                        }
                    });
            registerNamingService(nettyName,nettyPort);
            // 配置完成，開(kāi)始綁定server，通過(guò)調(diào)用sync同步方法阻塞直到綁定成功
            ChannelFuture future = bootstrap.bind(nettyPort).sync();
            if (future.isSuccess()) {
                log.info("Server started and listen on:{}", future.channel().localAddress());
                log.info("啟動(dòng) Netty Server");
            }
        } catch (InterruptedException e) {
            log.error("netty異常：{}", e.getMessage());
        }
    }
    /**
     * 將Netty服務(wù)注冊(cè)進(jìn)Nacos
     *
     * @param nettyName 服務(wù)名稱
     * @param nettyPort 服務(wù)端口號(hào)
     */
    private void registerNamingService(String nettyName, Integer nettyPort) {
        try {
            Properties properties = new Properties();
            properties.setProperty(PropertyKeyConst.SERVER_ADDR, nacosDiscoveryProperties.getServerAddr());
            properties.setProperty(PropertyKeyConst.NAMESPACE, nacosDiscoveryProperties.getNamespace());
            properties.setProperty(PropertyKeyConst.USERNAME, nacosDiscoveryProperties.getUsername());
            properties.setProperty(PropertyKeyConst.PASSWORD, nacosDiscoveryProperties.getPassword());
            NamingService namingService = NamingFactory.createNamingService(properties);
            InetAddress address = InetAddress.getLocalHost();
            // 定義服務(wù)實(shí)例信息
            Instance instance = new Instance();
            instance.setIp(address.getHostAddress());
            instance.setPort(nettyPort);
            instance.setWeight(1.0);
            instance.setHealthy(true);
            namingService.registerInstance(nettyName, nacosDiscoveryProperties.getGroup(), instance);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 釋放資源
     */
    @PreDestroy
    public void destroy() throws InterruptedException {
        if (bossGroup != null) {
            bossGroup.shutdownGracefully().sync();
        }
        if (workGroup != null) {
            workGroup.shutdownGracefully().sync();
        }
        log.info("關(guān)閉Netty");

    }

}

# spring配置
  spring:
    cloud:
      gateway:
        discovery:
          locator:
            lowerCaseServiceId: true
            enabled: true
        routes:
          # 系統(tǒng)模塊
          - id: soc-dmoasp-system
            uri: lb://soc-dmoasp-system
            predicates:
              - Path=/system/**
            filters:
              - StripPrefix=1
          # netty服務(wù)
          - id:  netty-server
            uri: lb:ws://soc-netty-server
            predicates:
              - Path=/netty-server/**
            filters:
              - StripPrefix=1
#不需要進(jìn)行權(quán)限校驗(yàn)的uri
security:
  ignore:
    whites:
      - /auth/logout
      - /auth/login
      - /auth/register
      - /*/v2/api-docs
      - /csrf
      #netty連接地址
      - /netty-server/**

@Component
@RefreshScope
public class AuthFilter implements GlobalFilter, Ordered
{
    private static final Logger log = LoggerFactory.getLogger(AuthFilter.class);

    // 排除過(guò)濾的 uri 地址，nacos自行添加
    @Autowired
    private IgnoreWhiteProperties ignoreWhite;

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain)
    {
        ServerHttpRequest request = exchange.getRequest();
        ServerHttpRequest.Builder mutate = request.mutate();

        String url = request.getURI().getPath();
        // 跳過(guò)不需要驗(yàn)證的路徑
        if (StringUtils.matches(url, ignoreWhite.getWhites()))
        {
            return chain.filter(exchange);
        }
        ......
    }
}

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.soc.dmoasp.common.core.constant.CacheConstants;
import com.soc.dmoasp.common.core.constant.TokenConstants;
import com.soc.dmoasp.common.core.enums.NettyMsgEnum;
import com.soc.dmoasp.common.core.utils.JwtUtils;
import com.soc.dmoasp.common.core.utils.StringUtils;
import com.soc.dmoasp.common.redis.service.RedisService;
import com.soc.dmoasp.system.server.vo.NettyResult;
import io.jsonwebtoken.Claims;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;

/**
 * netty鑒權(quán)處理
 * @author dsw
 * @date 2024/11/18 17:55
 */
public class AuthHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    Logger log = LoggerFactory.getLogger(AuthHandler.class);

    private final RedisTemplate<String, Object> redisTemplate;

    private final RedisService redisService;

    public AuthHandler(RedisTemplate<String, Object> stringObjectRedisTemplate, RedisService redisService) {
        redisTemplate = stringObjectRedisTemplate;
        this.redisService = redisService;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) throws Exception {
        try {
            JSONObject clientMessage = JSON.parseObject(textWebSocketFrame.text());
            //獲取code 只判斷code為鑒權(quán)的消息類(lèi)型
            Integer code = clientMessage.getInteger("code");
            if (NettyMsgEnum.AUTH_MESSAGE.getCode().equals(code)) {
                //獲取token
                String token = clientMessage.getString("token");
                if (StringUtils.isEmpty(token))
                {
                    ctx.channel().writeAndFlush(NettyResult.authFail("令牌不能為空"));
                    ctx.close();
                }
                // 如果前端設(shè)置了令牌前綴，則裁剪掉前綴
                if (StringUtils.isNotEmpty(token) && token.startsWith(TokenConstants.PREFIX))
                {
                    token = token.replaceFirst(TokenConstants.PREFIX, StringUtils.EMPTY);
                }
                //JWT校驗(yàn)
                Claims claims = JwtUtils.parseToken(token);
                if (claims == null)
                {
                    ctx.channel().writeAndFlush(NettyResult.authFail("令牌已過(guò)期或驗(yàn)證不正確"));
                    ctx.close();
                }
                String userkey = JwtUtils.getUserKey(claims);
                //從Redis中查看是否有這個(gè)Token沒(méi)有則不能登錄
                boolean islogin = redisService.hasKey(getTokenKey(userkey));
                if (!islogin)
                {
                    ctx.channel().writeAndFlush(NettyResult.authFail("登錄狀態(tài)已過(guò)期"));
                    ctx.close();
                }
                //獲取用戶保存至Socket連接會(huì)話中
                String userId = JwtUtils.getUserId(claims);
                AttributeKey<String> userIdKey = AttributeKey.valueOf("userId");
                ctx.channel().attr(userIdKey).setIfAbsent(userId);
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("userId",userId);
                log.info("有新的Socket客戶端鏈接 userId :{}", userId);
                //將連接信息保存至Redis中key為userId value為ctx.channel().id()
                redisTemplate.opsForHash().put(CacheConstants.getUserChannelKey(),userId,ctx.channel().id());
                ctx.channel().writeAndFlush(NettyResult.success(NettyMsgEnum.AUTH_MESSAGE.getCode(), "鑒權(quán)成功", jsonObject));
                //鑒權(quán)完成后移除AuthHandler消息監(jiān)聽(tīng)
                ctx.pipeline().remove(AuthHandler.class);
            } else {
                ctx.channel().writeAndFlush(NettyResult.authFail("請(qǐng)先鑒權(quán)，在發(fā)送其他類(lèi)型請(qǐng)求！"));
                ctx.close();
            }
        } catch (Exception e) {
            log.error(e.getMessage());
            ctx.channel().writeAndFlush(NettyResult.authFail("鑒權(quán)失敗"));
            ctx.close();
        }
    }

    /**
     * 獲取緩存key
     */
    private String getTokenKey(String token)
    {
        return CacheConstants.LOGIN_TOKEN_KEY + token;
    }
}

/**
 * netty消息類(lèi)型枚舉
 * @author dsw
 * @date 2024/11/18 17:58
 */
public enum NettyMsgEnum {

    AUTH_MESSAGE(1001, "鑒權(quán)消息","Auth-Netty"),
    //{'code':1003,'data':{'unreadCount':0}}
    NOTICE_MESSAGE(1003, "公告通知消息","Notice-Netty"),
    HEART_MESSAGE(1006, "心跳消息","Heart-Netty"),
    ERROR_MESSAGE(-1, "錯(cuò)誤",null);

    private final Integer code;

    private final String info;

    private final String strategyName;

    NettyMsgEnum(Integer code, String info, String strategyName){
        this.code = code;
        this.info = info;
        this.strategyName = strategyName;
    }

    public static NettyMsgEnum getByCode(Integer code) {
        for (NettyMsgEnum msgEnum : values()) {
            if (msgEnum.getCode().equals(code)) {
                return msgEnum;
            }
        }
        return ERROR_MESSAGE;
    }

    public Integer getCode() {
        return code;
    }

    public String getInfo() {
        return info;
    }

    public String getStrategyName() {
        return strategyName;
    }
}

import com.alibaba.fastjson2.JSON;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

import java.io.Serializable;

/**
 * netty響應(yīng)實(shí)體
 * @author dsw
 * @date 2024/11/18 18:02
 */
public class NettyResult  implements Serializable {

    private static final long serialVersionUID = 1L;

    private Integer code;

    private String message;

    private Object data;

    public NettyResult(Integer code, String message, Object data) {
        this.code = code;
        this.message = message;
        this.data = data;
    }


    public static TextWebSocketFrame fail(String message) {
        return new TextWebSocketFrame(JSON.toJSONString(new NettyResult(-1, message, null)));
    }

    public static TextWebSocketFrame authFail(String message) {
        return new TextWebSocketFrame(JSON.toJSONString(new NettyResult(-2, message, null)));
    }

    public static TextWebSocketFrame success( String message) {
        return new TextWebSocketFrame(JSON.toJSONString(new NettyResult(200, message, null)));
    }

    public static TextWebSocketFrame success(Integer code, Object data) {
        return new TextWebSocketFrame(JSON.toJSONString(new NettyResult(code,null, data)));
    }

    public static TextWebSocketFrame success(Integer code, String message, Object data) {
        return new TextWebSocketFrame(JSON.toJSONString(new NettyResult(code,message, data)));
    }

    public Integer getCode() {
        return code;
    }

    public String getMessage() {
        return message;
    }

    public Object getData() {
        return data;
    }
}

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * 空閑檢測(cè)
 * @author dsw
 * @date 2024/11/18 11:47
 */
public class WebSocketIdleStateHandler extends IdleStateHandler {

    Logger log = LoggerFactory.getLogger(WebSocketIdleStateHandler.class);

    /**
     * 默認(rèn)的讀空閑時(shí)間
     */
    private static final int DEFAULT_READER_IDLE_TIME = 10;

    /**
     * 默認(rèn)10秒讀空閑斷開(kāi)客戶端
     */
    public WebSocketIdleStateHandler() {
        super(DEFAULT_READER_IDLE_TIME, 0, 0, TimeUnit.SECONDS);
    }

    /**
     * 指定心跳時(shí)間（秒）
     *
     * @param readerIdleTimeSeconds 讀空閑時(shí)間
     * @param writerIdleTimeSeconds 寫(xiě)空閑時(shí)間
     * @param allIdleTimeSeconds    讀寫(xiě)空閑時(shí)間
     */
    public WebSocketIdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {
        super(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds, TimeUnit.SECONDS);
    }

    /**
     * 指定心跳時(shí)間及時(shí)間單位
     *
     * @param readerIdleTime 讀空閑時(shí)間
     * @param writerIdleTime 寫(xiě)空閑時(shí)間
     * @param allIdleTime    讀寫(xiě)空閑時(shí)間
     * @param unit           時(shí)間單位
     */
    public WebSocketIdleStateHandler(long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit) {
        super(readerIdleTime, writerIdleTime, allIdleTime, unit);
    }

    /**
     * 當(dāng)空閑事件觸發(fā)時(shí)執(zhí)行
     */
    @Override
    protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
        //如果是讀空閑
        if (evt.state().equals(IdleState.READER_IDLE)) {
            Channel channel = ctx.channel();
            log.debug("服務(wù)端未檢測(cè)到客戶端【{}】的心跳包，強(qiáng)制關(guān)閉客戶端！", channel.id());
            channel.close();
        }
        super.channelIdle(ctx,evt);
    }
}

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.soc.dmoasp.common.core.constant.CacheConstants;
import com.soc.dmoasp.common.core.enums.NettyMsgEnum;
import com.soc.dmoasp.system.server.config.NettyConfig;
import com.soc.dmoasp.system.server.strategy.NettyStrategy;
import com.soc.dmoasp.system.server.strategy.NettyStrategyFactory;
import com.soc.dmoasp.system.server.vo.NettyResult;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;


/**
 * webSocket處理
 * @author dsw
 * @date 2024/11/18 10:20
 */
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    Logger log = LoggerFactory.getLogger(WebSocketHandler.class);

    private final RedisTemplate<String, Object> redisTemplate;
    
    private final NettyStrategyFactory nettyStrategyFactory;

    public WebSocketHandler(RedisTemplate<String, Object> redisTemplate, NettyStrategyFactory nettyStrategyFactory) {
        this.redisTemplate = redisTemplate;
        this.nettyStrategyFactory = nettyStrategyFactory;
    }

    /**
    * webSocket連接創(chuàng)建后調(diào)用
    */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        // 添加到channelGroup 通道組
        NettyConfig.getChannelGroup().add(ctx.channel());
    }
    /**
    * 讀取數(shù)據(jù)
    */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) throws Exception {
        AttributeKey<String> userIdKey = AttributeKey.valueOf("userId");
        String userId = ctx.channel().attr(userIdKey).get();
        log.info("收到消息 userId：{} message:{}",userId,frame.text());
        // 接收客戶端的消息
        JSONObject pullMessage = JSON.parseObject(frame.text());
        Integer code = pullMessage.getInteger("code");
        // 獲取消息類(lèi)型
        NettyStrategy nettyStrategy = nettyStrategyFactory.getNettyStrategy(NettyMsgEnum.getByCode(code));
        // 處理消息
        TextWebSocketFrame pushMessage = nettyStrategy.execute(pullMessage);
        // 返回處理結(jié)果給客戶端
        ctx.channel().writeAndFlush(pushMessage);
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        AttributeKey<String> userIdKey = AttributeKey.valueOf("userId");
        String userId = ctx.channel().attr(userIdKey).get();
        log.info("用戶下線了 userId:{}",userId);
        // 刪除通道
        removeUserId(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.channel().writeAndFlush(NettyResult.fail("系統(tǒng)錯(cuò)誤：" + cause.getMessage()));
    }

    /**
    * 刪除用戶與channel的對(duì)應(yīng)關(guān)系
    */
    private void removeUserId(ChannelHandlerContext ctx) {
        AttributeKey<String> userIdKey = AttributeKey.valueOf("userId");
        String userId = ctx.channel().attr(userIdKey).get();
      if(StringUtils.isNotBlank(userId)){
         redisTemplate.opsForHash().delete(CacheConstants.getUserChannelKey(),userId);
      }
   }
}

import com.alibaba.fastjson2.JSONObject;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

/**
 * Netty接收消息處理策略類(lèi)
 * @author dsw
 * @date 2024/5/27 10:21
 */
public interface NettyStrategy {

    /**
     * 執(zhí)行添加數(shù)值
     *
     * @return
     */
    TextWebSocketFrame execute(JSONObject message);
}

import com.soc.dmoasp.common.core.enums.NettyMsgEnum;
import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * Netty策略工廠
 * @author dsw
 * @date 2024/11/18 10:20
 */
@Component
public class NettyStrategyFactory {

    Logger log = org.slf4j.LoggerFactory.getLogger(NettyStrategyFactory.class);

    /**
     * 通過(guò)Spring容器的方式注入
     */
    @Autowired
    private Map<String, NettyStrategy> nettyStrategy;

    /**
     * 獲取對(duì)應(yīng)策略類(lèi)
     * @param
     */
    public NettyStrategy getNettyStrategy(NettyMsgEnum nettyMsgEnum){
        if(!nettyStrategy.containsKey(nettyMsgEnum.getStrategyName())){
            log.warn("沒(méi)有對(duì)應(yīng)的消息策略");
            throw new RuntimeException("沒(méi)有對(duì)應(yīng)的消息策略");
        }
        return nettyStrategy.get(nettyMsgEnum.getStrategyName());
    }
}

import com.alibaba.fastjson2.JSONObject;
import com.soc.dmoasp.common.core.enums.NettyMsgEnum;
import com.soc.dmoasp.system.server.strategy.NettyStrategy;
import com.soc.dmoasp.system.server.vo.NettyResult;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.springframework.stereotype.Component;

/**
 * Netty心跳消息
 * @author dsw
 * @date 2024/5/27 10:25
 */
@Component("Heart-Netty")
public class HeartStrategyImpl implements NettyStrategy {

    @Override
    public TextWebSocketFrame execute(JSONObject message) {
        String data = message.getString("data");
        if ("ping".equals(data)) {
            return NettyResult.success(NettyMsgEnum.HEART_MESSAGE.getCode(), null, "pong");
        }
        return NettyResult.fail("消息格式不正確");
    }
}

import com.soc.dmoasp.common.core.constant.CacheConstants;
import com.soc.dmoasp.common.redis.configure.FastJson2JsonRedisSerializer;
import com.soc.dmoasp.system.server.receiver.PushMsgRedisReceiver;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

import java.util.Collections;
import java.util.List;

/**
 * redisReceiver 配置
 * @author dsw
 * @date 2024/11/18 12:07
 */
@Configuration
public class RedisReceiverConfig {

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory,
                                                                       MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        List<PatternTopic> topics = Collections.singletonList(
            PatternTopic.of(CacheConstants.Topic.SYS_SOCKET_PUSH_TOPIC)
        );
        // 添加訂閱者監(jiān)聽(tīng)類(lèi)，數(shù)量不限.PatternTopic定義監(jiān)聽(tīng)主題,這里監(jiān)聽(tīng)test-topic主題
        container.addMessageListener(listenerAdapter, topics);
        return container;
    }

    @Bean
    @SuppressWarnings(value = { "unchecked", "rawtypes" })
    public MessageListenerAdapter listenerAdapter(PushMsgRedisReceiver pushMsgRedisReceiver) {
        MessageListenerAdapter adapter = new MessageListenerAdapter(pushMsgRedisReceiver);
        FastJson2JsonRedisSerializer serializer = new FastJson2JsonRedisSerializer(Object.class);
        adapter.setSerializer(serializer);
        return adapter;
    }
}

import com.soc.dmoasp.common.core.constant.CacheConstants;
import com.soc.dmoasp.common.core.exception.Asserts;
import com.soc.dmoasp.common.redis.dto.NettyMessage;
import com.soc.dmoasp.system.server.config.NettyConfig;
import com.soc.dmoasp.system.server.vo.NettyResult;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Objects;

/**
 * Netty發(fā)送消息接收者
 * @author dsw
 * @date 2024/11/18 12:07
 */
@Component
public class PushMsgRedisReceiver {
    Logger log = LoggerFactory.getLogger(PushMsgRedisReceiver.class);

    @Autowired
    private RedisTemplate<String,Object> stringObjectRedisTemplate;

    /**
     * RedisTopic訂閱
     * @param nettyMessage netty消息
     * @param topic netty消息對(duì)應(yīng)的topic
     */
    public void handleMessage(NettyMessage nettyMessage, String topic) {

        Object channelId = stringObjectRedisTemplate.opsForHash().get(CacheConstants.getUserChannelKey(), nettyMessage.getUserId());
        if (Objects.isNull(channelId)) {
            log.warn("推送消息失敗,用戶不在線！ userId:{},msg:{}",nettyMessage.getUserId(),nettyMessage.getMessage());
            Asserts.fail("推送消息失敗，用戶不在線！");
        }
        Channel channel = NettyConfig.getChannelGroup().find((ChannelId) channelId);
        if(channel!=null){
            channel.writeAndFlush(NettyResult.success(nettyMessage.getCode(),nettyMessage.getMessage()));
            log.info("推送消息成功！ userId:{},msg:{}",nettyMessage.getUserId(),nettyMessage.getMessage());
        }else {
            log.warn("推送消息失敗,沒(méi)有找到Channel！ userId:{},msg:{}",nettyMessage.getUserId(),nettyMessage.getMessage());
        }
    }
}

/**
 * Redis消息發(fā)布訂閱 發(fā)布消息
 * @param channel 通道ID
 * @param message 消息
 */
@Async
public void convertAndSend(String channel, Object message) {
    redisTemplate.convertAndSend(channel, message);
}

import com.alibaba.fastjson2.JSONObject;
import com.soc.dmoasp.common.core.constant.CacheConstants;
import com.soc.dmoasp.common.redis.dto.NettyMessage;
import com.soc.dmoasp.common.redis.dto.PushSocketMsgDTO;
import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Set;

/**
 * Netty發(fā)送消息給服務(wù)端
 * @author dsw
 * @date 2024/11/18 11:47
 */
@Service
public class PushSocketMsgService {

    Logger log = org.slf4j.LoggerFactory.getLogger(PushSocketMsgService.class);

    @Autowired
    private RedisTemplate<String, Object> stringObjectRedisTemplate;

    @Autowired
    private RedisService redisService;

    /**
     * 給所有用戶發(fā)送消息
     * @param msgDTO
     */
    public void pushMsgToAll( PushSocketMsgDTO msgDTO) {
        Set<Object> keys = stringObjectRedisTemplate.opsForHash().keys(CacheConstants.getUserChannelKey());
        keys.forEach(key -> this.pushMsgToUser(msgDTO.getCode(), key.toString(), msgDTO.getMessage()));
    }

    /**
     * 給指定用戶發(fā)送消息
     * @param msgDTO
     */
    public void pushMsgToUserList(PushSocketMsgDTO msgDTO) {
        for(Long userId : msgDTO.getUserIdList()){
            this.pushMsgToUser(msgDTO.getCode(), userId.toString(), msgDTO.getMessage());
        }
    }

    protected void pushMsgToUser(Integer code, String userId, JSONObject message) {
        //推送到其他負(fù)載處理
        NettyMessage nettyMessage = new NettyMessage();
        nettyMessage.setUserId(userId);
        nettyMessage.setCode(code);
        nettyMessage.setMessage(message);
        redisService.convertAndSend(CacheConstants.Topic.SYS_SOCKET_PUSH_TOPIC, nettyMessage);
        log.info("推送消息成功！ userId:{},message:{}", userId, message);
    }
}

import com.alibaba.fastjson2.JSONObject;
import lombok.Data;

/**
 * Neety消息發(fā)布VO
 * @author dsw
 * @date 2024/11/18 13:49
 */
@Data
public class NettyMessage {

    private Integer code;

    private String userId;

    private JSONObject message;

}

import com.alibaba.fastjson2.JSONObject;
import lombok.Data;

/**
 * 發(fā)送socket消息DTO
 * @author dsw
 * @date 2024/11/18 13:39
 */
@Data
public class PushSocketMsgDTO {
    /**
     * 消息類(lèi)型
     * 詳情看 NettyMsgEnum
     */
    private Integer code;
    /**
     * 用戶ID
     */
    private List<Long> userIdList;
    /**
     * 消息體
     */
    private JSONObject message;
}