百度360必应搜狗淘宝本站头条
php redisredis 命令oracle位图索引redis 锁redis 主从
当前位置：网站首页 > 技术教程 > 正文

基于Netty+redis实现websocket集群方案

mhr18 2024-11-13 11:07 20 浏览 0 评论

一 背景

最近公司在做标注系统，有些用户之间需要进行消息的推送，比如管理员分配了一个任务给标注员时，需要给标注员推送一条消息

之前项目中用的spring-boot自带的websocket,spring-boot-starter-websocket集成，代码臃肿，性能不是很好

这次打算用netty来实现websocket，因为应用后期可能部署集群，用redis的sub pub功能来实现集群（用ZK也可以实现）

二 实现

1 当用户连接上来时，需要保存用户与channel的映射关系

2 当A用户需要向B用户推送一条消息的时候，需要根据发布到所有节点，看看B用户在那台实例上

三 代码实现

代码结构：netty包下，如下图所示：

一 、netty服务端启动类：

package com.minivision.label.management.netty;
 
import com.minivision.label.management.util.SysThreadPool;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
 
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
 
/**
 * @program: label-management-system-backend
 * @description: netty工具类
 * @author: wangjinwei
 * @create: 2022-01-04 09:34
 **/
@Component
@Slf4j
public class WebSocketServer {
    /**
     * webSocket协议名
     */
    private static final String WEBSOCKET_PROTOCOL = "WebSocket";
 
    /**
     * 端口号
     */
    @Value("${netty.server.port:8782}")
    private int port;
 
    private String path = "/home";
 
    private EventLoopGroup bossGroup;
    private EventLoopGroup workGroup;
    private LoggingHandler LOGGING_HANDLER;
 
 
    /**
     * 启动
     *
     * @throws InterruptedException
     */
    private void start() throws InterruptedException {
 
 
        // bossGroup就是parentGroup，是负责处理TCP/IP连接的
        bossGroup = new NioEventLoopGroup();
        // workerGroup就是childGroup,是负责处理Channel(通道)的I/O事件
        workGroup = new NioEventLoopGroup();
        //日志打印
        LOGGING_HANDLER = new LoggingHandler(LogLevel.INFO);
        ServerBootstrap sb = new ServerBootstrap();
        //设置全连接队列大小
        sb.option(ChannelOption.SO_BACKLOG, 1024);
        sb.group(workGroup, bossGroup)
                .channel(NioServerSocketChannel.class)
                .localAddress(this.port)
                .childHandler(new WebSocketInitializer(path,WEBSOCKET_PROTOCOL,LOGGING_HANDLER));
 
        // 配置完成，开始绑定server，通过调用sync同步方法阻塞直到绑定成功
        ChannelFuture channelFuture = sb.bind(port).sync();
        log.info("【Netty服务启动成功========端口：" + port + "】");
        log.info("Server started and listen on:{}", channelFuture.channel().localAddress());
        // 成功绑定到端口之后,给channel增加一个 管道关闭的监听器并同步阻塞,直到channel关闭,线程才会往下执行,结束进程。
        channelFuture.channel().closeFuture().sync();
 
    }
 
    /**
     * 释放资源
     *
     * @throws InterruptedException
     */
    @PreDestroy
    public void destroy() throws InterruptedException {
        if (bossGroup != null) {
            bossGroup.shutdownGracefully().sync();
        }
        if (workGroup != null) {
            workGroup.shutdownGracefully().sync();
        }
    }
 
    @PostConstruct()
    public void init() {
        // 需要开启一个新的线程来执行netty server 服务器
        SysThreadPool.getThread().execute(() -> {
            try {
                start();
            } catch (InterruptedException e) {
                log.error("【Netty服务启动失败】" + e.getMessage(), e);
                Thread.currentThread().interrupt();
            }
        });
    }
}

二、handler初始化类：

package com.minivision.label.management.netty;
 
import com.minivision.label.management.netty.handler.AuthHandler;
import com.minivision.label.management.netty.handler.LoginRequestHandler;
import com.minivision.label.management.netty.handler.ServerHeartBeatHandler;
import com.minivision.label.management.netty.handler.WebSocketHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
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.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
 
import java.util.concurrent.TimeUnit;
 
/**
 * @program: label-management-system-backend
 * @description: 初始化handler
 * @author: wangjinwei
 * @create: 2022-01-04 15:50
 **/
@Slf4j
public class WebSocketInitializer extends ChannelInitializer<SocketChannel> {
 
 
    private String path;
 
    /**
     * webSocket协议名
     */
    private String protocol;
 
    private LoggingHandler loggingHandler;
 
    public WebSocketInitializer(String path, String protocol, LoggingHandler loggingHandler) {
        this.path = path;
        this.protocol = protocol;
        this.loggingHandler = loggingHandler;
    }
 
    @Override
    protected void initChannel(SocketChannel ch) {
        log.info("收到新连接：" + ch.localAddress());
        // websocket协议本身是基于Http协议的，所以需要Http解码器
        ch.pipeline().addLast("http-codec", new HttpServerCodec());
        //netty中的日志handler
        ch.pipeline().addLast(loggingHandler);
        // 以块的方式来写的处理器
        ch.pipeline().addLast("http-chunked", new ChunkedWriteHandler());
        //HttpObjectAggregator的作用是将请求分段再聚合,参数是聚合字节的最大长度，将HTTP消息的多个部分合成一条完整的HTTP消息 说明：
        // 1、http数据在传输过程中是分段的，HttpObjectAggregator可以将多个段聚合 2、这就是为什么，当浏览器发送大量数据时，就会发送多次http请求
        ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
        /*
         * 说明：
         * 1、对应webSocket，它的数据是以帧（frame）的形式传递
         * 2、浏览器请求时 ws://localhost:58080/xxx表示请求的uri
         * 3、核心功能是将http协议升级为ws协议，保持长连接
         * 这个的作用主要是用来解决HTTP握手等问题。虽然可以自己实现，但是推荐采用这个默认的handler，它能够解决很多未知的问题。
         */
        ch.pipeline().addLast(new WebSocketServerProtocolHandler(path, protocol, true, 65536 * 10));
        // 进行设置心跳检测 0表示不监控
        ch.pipeline().addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS));
        //心跳事件
        ch.pipeline().addLast(new ServerHeartBeatHandler());
        // 配置通道处理 来进行业务处理
        ch.pipeline().addLast(new WebSocketHandler());
        //登录处理
        ch.pipeline().addLast(new LoginRequestHandler());
        //后面互发消息可以插拔式校验
        ch.pipeline().addLast(new AuthHandler());
    }
}

三、业务handler:

package com.minivision.label.management.netty.handler;
 
import com.minivision.label.management.netty.protocol.Packet;
import com.minivision.label.management.netty.protocol.PacketCodeC;
import com.minivision.label.management.netty.util.SessionUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
 
/**
 * @program: label-management-system-backend
 * @description: netty业务类
 * @author: wangjinwei
 * @create: 2022-01-04 09:34
 **/
@Component
@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
 
    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        if (ctx.channel().isWritable()) {
            log.info("channel is writable try to continue flush....");
            ctx.flush();
        }
        ctx.fireChannelWritabilityChanged();
    }
 
    /**
     * 客户端与服务器建立连接时触发
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("与客户端建立连接，通道开启！channelActive 被调用" + ctx.channel().id().asLongText());
    }
 
    /**
     * 客户端与服务器关闭连接时触发
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        String channelLongId = ctx.channel().id().asLongText();
        log.info("channelInactive 被调用" + channelLongId);
        // 删除通道
        SessionUtil.unBindSession(ctx.channel());
    }
 
    /**
     * 服务器接收客户端的数据消息
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        String content = msg.text();
        log.info("服务器收到消息：{}", content);
        Packet packet = PacketCodeC.INSTANCE.getPacket(content);
        if (packet == null) {
            log.error("WebSocketHandler.channelRead0 decode error content={}", content);
            ctx.channel().close();
        } else {
 
            ctx.fireChannelRead(packet);
        }
    }
 
    /**
     * description token校验
     *
     * @param isAppClient 是否是APP连接
     * @param token       token
     * @return boolean
     * @author wangjinwei
     * @date 2019年2月18日下午7:49:32
     */
    private boolean checkToken(String token, Boolean isAppClient) {
        return true;
    }
 
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("", cause);
        super.exceptionCaught(ctx, cause);
    }
}
 
 
package com.minivision.label.management.netty.handler;
 
import com.alibaba.fastjson.JSON;
import com.minivision.admin.center.open.api.TokenVerifyServiceApi;
import com.minivision.admin.center.open.api.dto.TokenVerifyReq;
import com.minivision.admin.center.open.api.dto.TokenVerifyResp;
import com.minivision.label.management.dto.LoginUser;
import com.minivision.label.management.netty.protocol.LoginRequestPacket;
import com.minivision.label.management.netty.protocol.LoginResponsePacket;
import com.minivision.label.management.netty.util.Attributes;
import com.minivision.label.management.netty.util.SessionUtil;
import com.minivision.label.management.util.SpringUtil;
import com.minivision.maiot.common.base.dto.ResultDTO;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
 
import java.util.stream.Collectors;
 
/**
 * @program: label-management-system-backend
 * @description:登陆请求处理
 * @author: wangjinwei
 * @create: 2022-01-04 09:34
 **/
@Slf4j
public class LoginRequestHandler extends SimpleChannelInboundHandler<LoginRequestPacket> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, LoginRequestPacket loginRequestPacket) throws Exception {
        log.info("收到客户端请求-----------------");
        LoginResponsePacket loginResponsePacket = new LoginResponsePacket();
        LoginUser loginUser = getLoginUser(loginRequestPacket.getMtk());
        if (loginUser != null) {
            ctx.channel().attr(Attributes.LOGINUSER).set(loginUser);
            SessionUtil.bindSession(loginUser, ctx.channel());
            log.info("LoginRequestHandler---userId={},userName={} : 登录成功!", loginUser.getUserId(), loginUser.getUserName());
 
            loginResponsePacket.setSuccess(true);
            loginResponsePacket.setUserId(loginUser.getUserId());
            loginResponsePacket.setUserName(loginUser.getUserName());
            //ctx.fireChannelRead(loginRequestPacket.retain()); 这样可以传递到下一个handler继续处理
        } else {
            loginResponsePacket.setSuccess(false);
            loginResponsePacket.setReason("账号密码校验失败");
            log.info("LoginRequestHandler--- {}, 登录失败!" + loginRequestPacket.getMtk());
        }
        ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(loginResponsePacket)));
 
        if (loginUser == null) {
            log.info("LoginRequestHandler--- {}, 登录失败!关闭channel" + loginRequestPacket.getMtk());
            ctx.channel().close();
        }
 
    }
 
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        SessionUtil.unBindSession(ctx.channel());
    }
 
    private LoginUser getLoginUser(String mtk) {
        TokenVerifyReq tokenVerifyReq = new TokenVerifyReq();
        tokenVerifyReq.setToken(mtk);
        TokenVerifyServiceApi tokenVerifyServiceApi = SpringUtil.getBean(TokenVerifyServiceApi.class);
        ResultDTO<TokenVerifyResp> tokenVerifyResp = tokenVerifyServiceApi.verifyToken(tokenVerifyReq);
        TokenVerifyResp resData = tokenVerifyResp.getResData();
        if (tokenVerifyResp.success() && resData.getStatus() == TokenVerifyResp.ReplyStatus.OK) {
            // 将认证后的信息传到后端服务
            LoginUser loginUser = new LoginUser();
            loginUser.setUserId(resData.getUserId());
            loginUser.setUserName(resData.getUserName());
            loginUser.setRoleIds(resData.getRoles().stream().map(TokenVerifyResp.UserRoleInfo::getRoleId).collect(Collectors.toList()));
            return loginUser;
        }
        return null;
    }
}
 
 
 
package com.minivision.label.management.netty.handler;
 
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
 
/**
 * @program: label-management-system-backend
 * @description: 心跳
 * @author: wangjinwei
 * @create: 2022-01-04 16:06
 **/
@Slf4j
public class ServerHeartBeatHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        //超时事件
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent idleEvent = (IdleStateEvent) evt;
            if (idleEvent.state() == IdleState.READER_IDLE) {
                //关闭通道连接
                log.info("ServerHeartBeatHandler.reader_idle");
            } else if (idleEvent.state() == IdleState.WRITER_IDLE) {
                //写
                log.info("ServerHeartBeatHandler.writer_idle");
            }
        }
        super.userEventTriggered(ctx, evt);
    }
}
 
 
 
 
package com.minivision.label.management.netty.handler;
 
import com.minivision.label.management.netty.util.SessionUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
 
 
/**
 * @program: label-management-system-backend
 * @description:权限可插拔
 * @author: wangjinwei
 * @create: 2021-12-28 10:52
 **/
@Slf4j
public class AuthHandler extends ChannelInboundHandlerAdapter {
 
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("AuthHandler.channelRead start;");
        if (!SessionUtil.hasLogin(ctx.channel())) {
            ctx.channel().close();
        } else {
            ctx.pipeline().remove(this);
            super.channelRead(ctx, msg);
        }
    }
 
 
}

四、协议接口：

package com.minivision.label.management.netty.protocol.command;
 
 
/**
 * @program: label-management-system-backend
 * @description: 消息类型
 * @author: wangjinwei
 * @create: 2021-12-28 14:07
 **/
public interface Command {
    /**
     * 登录请求
     */
    Byte LOGIN_REQUEST = 1;
    /**
     *登录响应
     */
    Byte LOGIN_RESPONSE = 2;
    /**
     * 客户端 消息请求
     */
    Byte MESSAGE_REQUEST = 3;
    /**
     * 服务端消息 响应
     */
    Byte MESSAGE_RESPONSE = 4;
 
 
}
 
package com.minivision.label.management.netty.protocol;
 
import lombok.Data;
 
import static com.minivision.label.management.netty.protocol.command.Command.LOGIN_REQUEST;
 
/**
 * @program: label-management-system-backend
 * @description: 登陆请求消息
 * @author: wangjinwei
 * @create: 2021-12-28 08:57
 **/
@Data
public class LoginRequestPacket extends Packet {
 
    private String mtk;
 
    @Override
    public Byte getCommand() {
        return LOGIN_REQUEST;
    }
}
 
 
package com.minivision.label.management.netty.protocol;
 
import lombok.Data;
 
import static com.minivision.label.management.netty.protocol.command.Command.LOGIN_RESPONSE;
 
/**
 * @program: label-management-system-backend
 * @description: 登录响应消息体
 * @author: wangjinwei
 * @create: 2021-12-28 14:12
 **/
@Data
public class LoginResponsePacket extends Packet {
 
    /**
     * 失败原因
     */
    private String reason;
    /**
     * 是否成功
     */
    private boolean success;
    /**
     * 用户id
     */
    private String userId;
    /**
     * 用户名
     */
    private String userName;
 
    @Override
    public Byte getCommand() {
 
        return LOGIN_RESPONSE;
    }
}
 
 
package com.minivision.label.management.netty.protocol;
 
import lombok.Data;
import lombok.NoArgsConstructor;
 
import static com.minivision.label.management.netty.protocol.command.Command.MESSAGE_REQUEST;
 
/**
 * @program: label-management-system-backend
 * @description: 客户端消息请求
 * @author: wangjinwei
 * @create: 2021-12-28 16:15
 **/
@Data
@NoArgsConstructor
public class MessageRequestPacket extends Packet {
 
    private String toUserId;
    private String message;
 
    public MessageRequestPacket(String toUserId, String message) {
        this.message = message;
        this.toUserId = toUserId;
    }
 
    @Override
    public Byte getCommand() {
        return MESSAGE_REQUEST;
    }
}
 
package com.minivision.label.management.netty.protocol;
 
 
import lombok.Data;
import lombok.experimental.Accessors;
 
import static com.minivision.label.management.netty.protocol.command.Command.MESSAGE_RESPONSE;
 
/**
 * @program: label-management-system-backend
 * @description: 客户端消息响应
 * @author: wangjinwei
 * @create: 2021-12-28 17:09
 **/
@Accessors(chain = true)
@Data
public class MessageResponsePacket extends Packet {
 
    private String fromUserId;
    private String fromUserName;
    private Object message;
 
    @Override
    public Byte getCommand() {
        return MESSAGE_RESPONSE;
    }
}
 
 
package com.minivision.label.management.netty.protocol;
 
/**
 * @program: label-management-system-backend
 * @description: 消息体的父类
 * @author: wangjinwei
 * @create: 2021-12-28 10:43
 **/
public abstract class Packet {
    /**
     *版本
     **/
    private Byte version = 1;
 
    /**
     * 消息类型 --登录请求，登录响应、消息请求，消息响应
     *
     * @return
     */
    public abstract Byte getCommand();
}
 
 
package com.minivision.label.management.netty.protocol;
 
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
 
import java.util.HashMap;
import java.util.Map;
 
import static com.minivision.label.management.netty.protocol.command.Command.*;
 
/**
 * @program: label-management-system-backend
 * @description: 消息解析
 * @author: wangjinwei
 * @create: 2021-12-28 14:07
 **/
@Slf4j
public class PacketCodeC {
 
    public static final PacketCodeC INSTANCE = new PacketCodeC();
 
    /**
     * 协议版本类型
     */
    private final Map<Byte, Class<? extends Packet>> packetTypeMap;
 
    private PacketCodeC() {
        packetTypeMap = new HashMap<>();
        packetTypeMap.put(LOGIN_REQUEST, LoginRequestPacket.class);
        packetTypeMap.put(LOGIN_RESPONSE, LoginResponsePacket.class);
        packetTypeMap.put(MESSAGE_REQUEST, MessageRequestPacket.class);
        packetTypeMap.put(MESSAGE_RESPONSE, MessageResponsePacket.class);
 
 
    }
 
    private Class<? extends Packet> getRequestType(byte command) {
        return packetTypeMap.get(command);
    }
 
    public Packet getPacket(String content) {
        try {
            JSONObject jsonObject = JSON.parseObject(content);
            Byte command = jsonObject.getByte("command");
            if (command == null) {
                log.error("PacketCodeC.getPacket command is null content={}", content);
                return null;
            }
            Class<? extends Packet> classType = getRequestType(command);
            if (classType == null) {
                return null;
            }
            Packet packet = jsonObject.toJavaObject(classType);
            return packet;
        } catch (Exception e) {
            log.error("PacketCodeC.getPacket", e);
        }
        return null;
    }
}

五、用户session

package com.minivision.label.management.netty.util;
 
import com.minivision.label.management.dto.LoginUser;
import io.netty.util.AttributeKey;
 
 
/**
 * @program: label-management-system-backend
 * @description:
 * @author: wangjinwei
 * @create: 2021-12-27 16:34
 **/
public interface Attributes {
 
    /**
     * 登陆标志
     */
    AttributeKey<Boolean> LOGIN = AttributeKey.newInstance("login");
    /**
     * 登陆用户信息
     */
    AttributeKey<LoginUser> LOGINUSER = AttributeKey.newInstance("LoginUser");
}
 
package com.minivision.label.management.netty.util;
 
import com.minivision.label.management.dto.LoginUser;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import lombok.extern.slf4j.Slf4j;
 
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
 
/**
 * @program: label-management-system-backend
 * @description: 登陆常量类
 * @author: wangjinwei
 * @create: 2021-12-28 11:09
 **/
@Slf4j
public class SessionUtil {
    /**
     * 用户channel对应关系
     */
    private static final Map<String, Channel> userIdChannelMap = new ConcurrentHashMap<>();
 
    /**
     * 群组消息
     */
    private static final Map<String, ChannelGroup> groupIdChannelGroup = new ConcurrentHashMap<>();
 
    /**
     * 绑定用户与通道
     * @param loginUser
     * @param channel
     */
    public static void bindSession(LoginUser loginUser, Channel channel) {
        userIdChannelMap.put(loginUser.getUserId(), channel);
        channel.attr(Attributes.LOGINUSER).set(loginUser);
    }
 
    /**
     * 获取用户
     * @param channel
     * @return
     */
    public static LoginUser getSession(Channel channel) {
        return channel.attr(Attributes.LOGINUSER).get();
 
    }
 
    public static boolean hasLogin(Channel channel) {
        return getSession(channel) != null;
    }
 
    public static Channel getChannel(String userId) {
        return userIdChannelMap.get(userId);
    }
 
    public static void bindChannelGroup(String groupId, ChannelGroup channelGroup) {
        groupIdChannelGroup.put(groupId, channelGroup);
    }
 
    public static ChannelGroup getChannelGroup(String groupId) {
        return groupIdChannelGroup.get(groupId);
    }
 
    /**
     * 解除绑定关系
     * @param channel
     */
    public static void unBindSession(Channel channel) {
        if (hasLogin(channel)) {
            LoginUser session = getSession(channel);
            userIdChannelMap.remove(session.getUserId());
            channel.attr(Attributes.LOGINUSER).set(null);
            log.info(session + " 退出登录!");
        }
 
    }
 
    public static Map<String, Channel> getUserIdChannelMap() {
        return userIdChannelMap;
    }
}

六、redis 实现集群：直接调用RedisClient#convertAndSend方法

package com.minivision.label.management.redis;
 
import com.minivision.label.management.service.PushService;
import org.springframework.beans.factory.annotation.Autowired;
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 static com.minivision.label.management.common.constant.LabelConstants.TASK_MSG_WS_TOPIC;
 
/**
 * @Description Redis消息监听者容器:
 * @author wangjinwei
 * @date 2021/4/29 15:02
 */
@SuppressWarnings({"all"})
@Configuration
public class RedisConfig {
 
 
    @Autowired
    private PushService pushService;
 
    /**
     * redis消息监听器容器
     * 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器
     * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理
     *
     * @param connectionFactory
     * @param listenerAdapter
     * @return
     */
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory
            , MessageListenerAdapter TopicAdapter, MessageListenerAdapter messageListenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //添加消息监听器
        container.addMessageListener(TopicAdapter, new PatternTopic(TASK_MSG_WS_TOPIC));
        return container;
    }
 
 
    /**
     * 消息监听器适配器，绑定消息处理器
     *
     * @param receiver
     * @return
     */
    @Bean
    MessageListenerAdapter TopicAdapter() {
        return new MessageListenerAdapter(new TopicListener(pushService));
    }
 
}
 
 
package com.minivision.label.management.redis;
 
import com.alibaba.fastjson.JSON;
import com.minivision.label.management.netty.protocol.Packet;
import com.minivision.label.management.service.PushService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
 
/**
 *
 * @author wangjinwei<br>
 * @version 1.0<br>
 * @Description: 监听消息<br>
 * @CreateDate 2021/4/27 9:13
 */
@Slf4j
public class TopicListener implements MessageListener {
 
    private PushService pushService;
 
    public TopicListener(PushService pushService) {
        this.pushService = pushService;
    }
 
    @Override
    public void onMessage(Message message, byte[] bytes) {
        String msg = message.toString();
        RedisWebsocketMsg<Packet> redisWebsocketMsg = JSON.parseObject(msg, RedisWebsocketMsg.class);
        pushService.pushMsg(redisWebsocketMsg.getReceiver(), JSON.toJSONString(redisWebsocketMsg.getContent()));
    }
}
 
 
 
 
package com.minivision.label.management.redis;
 
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
 
/**
 * Description：redis客户端操作
 *
 * @author wangjinwei
 * @date 2021/4/28
 */
@Slf4j
@Component
public class RedisClient {
 
 
    @Autowired
    private StringRedisTemplate redisTemplate;
 
 
    /**
     * @return void
     * @Description:发送群消息
     * @param: topicName
     * @param: redisWebsocketMsg
     * @author wangjinwei
     * @date 2021/4/30 11:13
     */
    public void convertAndSend(String topicName, RedisWebsocketMsg redisWebsocketMsg) {
        redisTemplate.convertAndSend(topicName, JSON.toJSONString(redisWebsocketMsg));
    }
 
 
}
 
package com.minivision.label.management.redis;
 
import lombok.Data;
import lombok.experimental.Accessors;
 
import java.io.Serializable;
import java.util.List;
 
/**
 * websocket redis广播消息
 * @author wangjinwei <br>
 * @version 1.0<br>
 * @CreateDate 2021/6/23 <br>
 */
@Accessors(chain = true)
@Data
public class RedisWebsocketMsg<T> implements Serializable {
    /**
     * 序列号
     */
    private static final long serialVersionUID = -1L;
    /**
     * 消息接收者
     */
    private List<String> receiver;
    /**
     * 内容
     */
    private T content;
}

七、最终暴露给其他模块调用的接口：pushService#sendMsg接口

package com.minivision.label.management.service;
 
import com.minivision.label.management.redis.RedisWebsocketMsg;
 
import java.util.List;
 
/**
 * @program: label-management-system-backend
 * @description:
 * @author: wangjinwei
 * @create: 2021-12-27 15:28
 **/
public interface PushService {
 
    /**
     * 推送给指定通道
     * @param userId channelId
     * @param msg msg
     */
    void pushMsgToOne(String userId, String msg);
 
    /**
     * 推送给指定用户
     * @param userIds channelId
     * @param msg msg
     */
    void pushMsg(List<String> userIds, String msg);
 
    /**
     * @Description: 将消息推送给单个用户
     * @param: redisWebsocketMsg
     * @return void
     * @author wangjinwei
     * @date 2021/4/30 10:40
     */
    void sendMsg(RedisWebsocketMsg redisWebsocketMsg);
}
 
 
 
 
 
package com.minivision.label.management.service.impl;
 
import com.minivision.label.management.netty.util.SessionUtil;
import com.minivision.label.management.redis.RedisClient;
import com.minivision.label.management.redis.RedisWebsocketMsg;
import com.minivision.label.management.service.PushService;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
 
import java.util.List;
 
import static com.minivision.label.management.common.constant.LabelConstants.TASK_MSG_WS_TOPIC;
 
/**
 * ━━━━━━佛祖保佑━━━━━━
 *                  ,;,,;
 *                ,;;'(    社
 *      __      ,;;' ' \   会
 *   /'  '\'~~'~' \ /'\.)  主
 * ,;(      )    /  |.     义
 *,;' \    /-.,,(   ) \    码
 *     ) /       ) / )|    农
 *     ||        ||  \)
 *     (_\       (_\
 * ━━━━━━永无BUG━━━━━━
 * @program: label-management-system-backend
 * @description:
 * @author: wangjinwei
 * @create: 2021-12-27 15:30
 **/
@Service
@Slf4j
public class PushServiceImpl implements PushService {
 
    @Autowired
    private RedisClient redisClient;
 
    @Override
    public void pushMsgToOne(String userId, String msg) {
        Channel channel = SessionUtil.getChannel(userId);
        if (channel == null) {
            log.info("未找到通道信息，数据不推送");
            return;
        }
        if (!channel.isActive()) {
            log.info("通道已经关闭，数据不推送");
            //移除通道信息
            SessionUtil.unBindSession(channel);
        }
        long start = System.currentTimeMillis();
        log.info("PushServiceImpl.pushMsgToOne userId={}, msg={},start={}", userId, msg, start);
        channel.writeAndFlush(new TextWebSocketFrame(msg));
        log.info("PushServiceImpl.pushMsgToOne userId={}, msg={},end={}", userId, msg, System.currentTimeMillis());
    }
 
    @Override
    public void pushMsg(List<String> userIds, String msg) {
        if (CollectionUtils.isEmpty(userIds)) {
            return;
        }
        userIds.forEach(userId -> {
            pushMsgToOne(userId, msg);
        });
    }
 
    /**
     * @Description: 将消息推送给单个用户
     * @param: redisWebsocketMsg
     * @return void
     * @author wangjinwei
     * @date 2021/4/30 10:40
     */
    @Override
    @SuppressWarnings("all")
    public void sendMsg(RedisWebsocketMsg redisWebsocketMsg) {
        long start = System.currentTimeMillis();
        //给其他订阅了主题的节点发消息,因为本节点没有
        redisClient.convertAndSend(TASK_MSG_WS_TOPIC, redisWebsocketMsg);
    }
 
}

相关推荐

甲骨文签署多项大型云协议，其一未来可贡献超300亿美元年收入

IT之家7月1日消息，根据甲骨文Oracle当地时间6月30日向美国证券交易委员会(SEC)递交的FORM8-K文件，该企业在始于2025年6月1日的202...

甲骨文获TEMU巨额合同，后者大部分基础设施将迁移至Oracle云

IT之家6月23日消息，Oracle甲骨文创始人、董事长兼首席技术官LarryEllison（拉里埃里森）在本月早些时候的2025财年第四财季和全财年财报电话会议上表示，Oracle...

Spring Boot 自定义数据源设置，这些坑你踩过吗?

你在使用SpringBoot进行后端开发的过程中，是不是也遇到过这样的问题：项目上线后，数据库连接总是不稳定，偶尔还会出现数据读取缓慢的情况，严重影响了用户体验。经过排查，发现很大一部分原因竟然...

一个开箱即用的代码生成器（一个开箱即用的代码生成器是什么）

今天给大家推荐一个好用的代码生成器，名为renren-generator，该项目附带前端页面，可以很方便的选择我们所需要生成代码的表。首先我们通过git工具克隆下来代码（地址见文末），导入idea。...

低代码建模平台-数据挖掘平台（低代码平台的实现方式）

现在来看一下数据连接。·这里是管理数据连接的空间，点击这里可以新增一个数据连接。·输入连接名称，然后输入url，是通过gdbc的方式去连接的数据库，目前是支持mysql、oracle以及国产数据库达梦...

navicat 17.2.7连接oracle数据库提示加载oracle库失败

系统：macOS15.5navicat版本：navicatpremiumlite17.2.7连接oracle测试报错：加载oracle库失败【解决办法】：放达里面找到程序，显示简介里面勾选“使...

开源“Windows”ReactOS更新：支持全屏应用

IT之家6月17日消息，ReactOS团队昨日（6月16日）在X平台发布系列推文，公布了该系统的最新进展，包括升级Explorer组件，支持全屏应用，从Wine项目引入了...

SSL 推出采用全模拟内置混音技术的模拟调音台Oracle

英国调音台传奇品牌SolidStateLogic宣布推出Oracle——一款采用全模拟内置混音技术的调音台，在紧凑的AWS尺寸机箱内集成了大型调音台的功能。该调音台提供24输入和...

47道网络工程师常见面试题，看看有没有你不会的!

你们好，我的网工朋友。网络工程师面试的时候，都会被问到什么？这个问题其实很泛，一般来说，你肯定要先看明白岗位需求写的是什么。基本上都是围绕公司需要的业务去问的。但不可否认的是，那些最基础的概念，多少也...

汉得信息:发布EBS系统安装启用JWS的高效解决方案

e公司讯，从汉得信息获悉，近日，微软官方宣布InternetExplorer桌面应用程序将于2022年6月15日正式停用。目前大部分客户都是使用IE浏览器打开EBS的Form界面，IE停用后，只能使...

36.9K star ! 推荐一个酷炫低代码开发平台!功能太强!

前言最近在逛github，看看能不能搜罗到一些对自己有帮助的开源软件。不经意间看到一个高star的java开源项目：jeecg-boot。进入在线演示版一看，感叹实在是太牛了！此开源项目不管是给来学习...

Linux新手入门系列:Linux下jdk安装配置

本系列文章是把作者刚接触和学习Linux时候的实操记录分享出来，内容主要包括Linux入门的一些理论概念知识、Web程序、mysql数据库的简单安装部署，希望能够帮到一些初学者，少走一些弯路。注意：L...

手把手教你在嵌入式设备中使用SQLite3

摘要：数据库是用来存储和管理数据的专用软件，使得管理数据更加安全，方便和高效。数据库对数据的管理的基本单位是表(table)，在嵌入式linux中有时候它也需要用到数据库，听起来好难，其实就是几个函数...

JAVA语言基础（java语言基础知识）

一、计算机的基本概念什么是计算机？计算机（Computer)全称：电子计算机，俗称电脑。是一种能够按照程序运行、自动高速处理海量数据的现代化智能电子设备。由硬件和软件组成、没有安装过任何软件的计算机称...

再见 Navicat!一款开源的 Web 数据库管理工具!

大家好，我是Java陈序员。在日常的开发工作中，常常需要与各种数据库打交道。而为了提高工作效率，常常会使用一些可视化工具进行操作数据库。今天，给大家介绍一款开源的数据库管理工具，无需下载安装软件，基...

取消回复欢迎 你 发表评论:

一周热门
最近发表
标签列表