【SpringBoot+自定义注解+Aop+长轮询】实现简单的实时消息推送

喜欢ヅ旅行 2024-03-26 12:25 40阅读 0赞

#### 文章目录 ####

*  前言
 *  一、应用场景
 *  二、使用步骤
 *   *  1.引入依赖
     *  2.数据库建立
     *  3.编写业务代码
     *  4.Token相关
     *  5.编写队列相关方法
     *  6.自定义注解+AOP
     *  7.长轮询
 *  功能测试
 *  总结
 *  推荐

--------------------

## 前言 ##

`小白摸索，大佬勿喷`

我们平常在网站上都可以看到右上角的消息上未读消息数量，以及有消息时右下角实时弹出的例子

我公司项目刚好也有这个需求，但是由于甲方要求不能使用第三方服务，于是便有了本文的解决思路

--------------------

## 一、应用场景 ##

该思路适用于一些无法使用第三方服务的场景，如果可以使用第三方服务仍旧是使用第三方的服务比较好

## 二、使用步骤 ##

### 1.引入依赖 ###

这里我用到的依赖有：`aspectj、mybatis-plus、jjwt、hutool`等，大家可以自行选择

<dependency>
                <groupId>com.baomidou</groupId>
                <artifactId>mybatis-plus-boot-starter</artifactId>
                <version>3.3.2</version>
            </dependency>
            <dependency>
                <groupId>com.baomidou</groupId>
                <artifactId>mybatis-plus-generator</artifactId>
                <version>3.3.2</version>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-freemarker</artifactId>
                <version>2.7.3</version>
            </dependency>
            <dependency>
                <groupId>org.aspectj</groupId>
                <artifactId>aspectjrt</artifactId>
                <version>1.9.6</version>
            </dependency>
            <dependency>
                <groupId>org.aspectj</groupId>
                <artifactId>aspectjweaver</artifactId>
                <version>1.9.6</version>
            </dependency>
            <dependency>
                <groupId>io.jsonwebtoken</groupId>
                <artifactId>jjwt</artifactId>
                <version>0.9.1</version>
            </dependency>
    
            <dependency>
                <groupId>cn.hutool</groupId>
                <artifactId>hutool-all</artifactId>
                <version>5.7.20</version>
            </dependency>

### 2.数据库建立 ###

`消息表：`

CREATE TABLE `t_message` (
      `id` int NOT NULL AUTO_INCREMENT,
      `content` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '内容',
      `from_uid` int DEFAULT NULL COMMENT '发起人id',
      `to_uid` int DEFAULT NULL COMMENT '接收人id',
      PRIMARY KEY (`id`)
    ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

`部门表：`

CREATE TABLE `t_dept` (
      `id` int NOT NULL AUTO_INCREMENT,
      `dept_name` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '部门名称',
      `dept_code` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '部门编码',
      `leader_uid` int DEFAULT NULL COMMENT '部门领导id',
      PRIMARY KEY (`id`)
    ) ENGINE=InnoDB AUTO_INCREMENT=21 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

`用户表：`

CREATE TABLE `t_user` (
      `id` int NOT NULL AUTO_INCREMENT,
      `username` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '用户名',
      `password` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '密码',
      `real_name` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '真实姓名',
      PRIMARY KEY (`id`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

### 3.编写业务代码 ###

> 为了方便演示我这里使用的是MybatisPlus生成后自带的方法去添加，@RealTimeMessage为自定义注解，下面会提到

/**
         * 添加部门
         *
         * @param dept 部门实体类
         * @return String
         */
        @RealTimeMessage("添加部门")
        @PostMapping("/add")
        public String addDept(@RequestBody Dept dept) {
        
            boolean save = deptService.save(dept);
            if (save) {
        
                return "SUCCESS";
            } else {
        
                return "FAILED";
            }
        }

### 4.Token相关 ###

`JwtUtils：`

@Component
    public class JwtUtils {
        
    
        private final String KEY = "abcdefghijklmnopqrstuvwxyz";
    
        /**
         * 创建token
         *
         * @param user 用户实体
         * @return String
         */
        public String createToken(User user) {
        
            Map<String, Object> claims = BeanUtil.beanToMap(user);
            JwtBuilder jwtBuilder = Jwts.builder()
                    .setClaims(claims)
                    .setIssuedAt(new Date())
                    .signWith(SignatureAlgorithm.HS512, KEY);
            return jwtBuilder.compact();
        }
    
        /**
         * 解析token
         *
         * @param token token
         * @return User
         */
        public User parseToken(String token) {
        
            Claims claims;
            try {
        
                claims = Jwts.parser()
                        .setSigningKey(KEY)
                        .parseClaimsJws(token)
                        .getBody();
            } catch (JwtException e) {
        
                claims = null;
                e.printStackTrace();
            }
            return BeanUtil.toBean(claims, User.class);
        }
    
    }

`TokenService及实现类：`

// TokenService
    public interface TokenService {
        
    	/**
    	* 获取当前登录用户的信息
    	* @return User 
    	*/
        User getUserByToken();
    }
    
    
    // TokenService的实现类
    @Service
    public class TokenServiceImpl implements TokenService {
        
    
        @Resource
        private JwtUtils jwtUtils;
    
        @Override
        public User getUserByToken() {
        
            HttpServletRequest request = getRequest();
            String token = request.getHeader("token");
            return jwtUtils.parseToken(token);
        }
    
        public HttpServletRequest getRequest(){
        
            return ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
        }

### 5.编写队列相关方法 ###

@Component
    public class MessageQueue {
        
    
        /**
         * 存放消息的map集合
         */
        private static final Map<User, Queue<Message>> MSG_MAP = new ConcurrentHashMap<>();
        private static final Map<User, Integer> RETURN_TIME = new ConcurrentHashMap<>();
    
        /**
         * 注册当前请求用户到map中
         *
         * @param user 当前用户
         */
        public void register(User user) {
        
            MSG_MAP.put(user, new ConcurrentLinkedDeque<>());
            RETURN_TIME.put(user, 0);
        }
    
        /**
         * 销毁已注册的用户
         *
         * @param user 当前用户
         */
        public void remove(User user) {
        
            MSG_MAP.remove(user);
            RETURN_TIME.remove(user);
        }
    
        /**
         * 将消息存入对应的接收方队列中
         *
         * @param message 要接收的消息
         */
        public void push(Message message) {
        
            MSG_MAP.keySet().forEach(a -> {
        
                synchronized (a) {
        
                    if (a.getId().equals(message.getToUid())) {
        
                        MSG_MAP.get(a).add(message);
                        a.notifyAll();
                    }
                }
            });
        }
    
        /**
         * 推送消息
         *
         * @param user 当前用户实体
         * @return List<Message>
         */
        public List<Message> pollAll(User user) {
        
            Map<User,Queue<Message>> map1 = MSG_MAP;
            Map<User,Integer> map2 = RETURN_TIME;
            List<Message> messageList = new ArrayList<>();
            Queue<Message> messageQueue = MSG_MAP.get(user);
            //如果队列不为空，则将消息添加到对应的队列中  否则将该用户注册到map中
            if (messageQueue != null) {
        
                messageQueue.forEach(a -> {
        
                    Message poll = messageQueue.poll();
                    if (poll != null) {
        
                        messageList.add(poll);
                    }
                });
            } else {
        
                register(user);
            }
    
            //如果消息为空则将服务阻塞 不响应前端请求 直到map中有新消息则返回消息 
            //此时需要与前端配合设置阻塞时间 可根据实际需求更改
            if (messageList.isEmpty() && RETURN_TIME.get(user) < 3) {
        
                try {
        
                    Integer time = RETURN_TIME.get(user);
                    RETURN_TIME.put(user, time + 1);
                    synchronized (user) {
        
                        user.wait(10000);
                    }
                    return pollAll(user);
                } catch (InterruptedException e) {
        
                    Thread.currentThread().interrupt();
                }
            }
            return messageList;
    
        }
    
    }

### 6.自定义注解+AOP ###

> 自定义注解，value用来存放操作名称

@Target(ElementType.METHOD)
    @Retention(RetentionPolicy.RUNTIME)
    public @interface RealTimeMessage {
        
    
        String value();
    }

> 编写切面去拦截添加了自定义注解的方法

@Aspect
    @Component
    public class RealTimeMessageAop {
        
    
        @Resource
        private MessageService messageService;
    
        @Resource
        private UserService userService;
    
        @Resource
        private DeptService deptService;
    
        @Resource
        private TokenService tokenService;
    
    	//
        @Resource
        private MessageQueue messageQueue;
    
        @Pointcut("@annotation(com.example.validator.anno.RealTimeMessage)")
        public void realTimeMessage() {
        
        }
    
        @Around("realTimeMessage()")
        public Object check(ProceedingJoinPoint point) throws Throwable {
        
            Object obj;
            //获取当前请求用户对象
            User user = tokenService.getUserByToken();
            //获取自定义注解
            Method method=((MethodSignature)point.getSignature()).getMethod();
            Signature signature = point.getSignature();
            Method realMethod = point.getTarget().getClass().getDeclaredMethod(signature.getName(), method.getParameterTypes());
            RealTimeMessage anno = realMethod.getAnnotation(RealTimeMessage.class);
            //获取前端传参，并转为map集合
            Object[] args = point.getArgs();
            Dept dept = BeanUtil.toBean(args[0], Dept.class);
            //继续执行原有方法
            obj = point.proceed();
            //获取返回消息
            String result = String.valueOf(obj);
            //如果请求成功则弹出消息 否则不做操作
            if ("SUCCESS".equals(result)) {
        
                //添加消息到数据库中
                Message build = Message.builder()
                        .content(anno.value())
                        .fromUid(user.getId())
                        .toUid(dept.getLeaderUid())
                        .build();
                messageService.save(build);
                //发送实时消息
                messageQueue.push(build);
            }
            return obj;
        }
    
    }

此处我想实现的效果为当管理员添加部门，并指定该部门的领导时，将通知实时发送给管理员设置的领导

### 7.长轮询 ###

`该接口登陆后由前端调用，到达约定时间或返回结果后，前端再次调用该接口实现长轮询`

@RestController
    @RequestMapping("/message")
    public class MessageController {
        
    
        @Resource
        private MessageQueue messageQueue;
    
        @Resource
        private TokenService tokenService;
    
    	/**
    	* 拉取实时消息
    	* @return List<Message> 
    	*/
        @GetMapping("/realTimeMsg")
        public List<Message> getRealTimeMsg() {
        
            //获取当前请求用户信息
            User user = tokenService.getUserByToken();
            try {
        
                //注册该用户
                messageQueue.register(user);
                //推送消息
                return messageQueue.pollAll(user);
            } catch (Exception e) {
        
                throw new IllegalStateException();
            } finally {
        
                //销毁注册
                messageQueue.remove(user);
            }
        }

## 功能测试 ##

自此我们的代码已经全部编写完成，接下来就启动项目看一下效果

然后打开我们的接口调试工具把接口与数据填进去  
![添加部门接口调试][b5365ff12a8c4e4da4453e58d489e121.png]

![拉取实时消息接口调试][56ef32edcc9b4f8c872a61c59babdc5c.png]

我们还需要`Token`的参与，为了方便演示我们就直接在生成所需要的Token

![Token生成单元测试][Token]  
接下来我们把管理员Token填入添加单位的接口的`Header`里  
![添加管理员Token到添加部门接口][Token 1]

然后我们把任命领导的Token填入拉取消息的`Header`里

![添加领导Token到拉取消息接口][Token 2]

我们`首先调用拉取实时消息的接口`，接口在没有消息的时候会一直阻塞，直到到达约定时间，如果到了约定时间仍无新消息则直接返回空列表，效果如下：  
![请求阻塞][687228bdec3b47108351fe2a22f14b22.png]![调用拉取实时消息接口][d9f72784846d42a986dfbdd29375ecd5.png]

接着我们`重新调用拉取实时消息的接口，并发送添加部门的请求`，这时我们可以看到消息已经推送过来了，效果如下：  
![调用添加部门接口][8311deaed4074c9ea5240bd53c439570.png]  
![拉取实时消息接口][1ad4ec08398a47b1964115a887222238.png]

自此我们的全部功能就已经完成啦

--------------------

## 总结 ##

本文提供了一种简单实时消息推送的一种思路，在使用时可根据实际需求进行修改，该思路主要工作在后端实现，前端只需要轮番调用拉取消息的接口即可

--------------------

## 推荐 ##

关注博客和公众号获取最新文章

[Bummon’s Blog][Bummon_s Blog] ｜ [Bummon’s Home][Bummon_s Home] ｜ [公众号][Link 1]

[b5365ff12a8c4e4da4453e58d489e121.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/98d0bd41fdd14a0f8fa4510ac13ace32.png
[56ef32edcc9b4f8c872a61c59babdc5c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/ae17be6c976a400fbe4a6be50a2a7ae2.png
[Token]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/6e1eaae13c3a4ebda2a29c81e81e6a36.png
[Token 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/928dc8e43ec045be94e09883745c86ba.png
[Token 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/1190c9c744c34d64806ecf5920626308.png
[687228bdec3b47108351fe2a22f14b22.png]: https://img-blog.csdnimg.cn/687228bdec3b47108351fe2a22f14b22.png
[d9f72784846d42a986dfbdd29375ecd5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/0533e253bb884686b5ef090d09622ef2.png
[8311deaed4074c9ea5240bd53c439570.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/b0abe0365a5f47b09b05a5110b0fcfe2.png
[1ad4ec08398a47b1964115a887222238.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/15a5481dc0a94f599f8e59256a44bbb9.png
[Bummon_s Blog]: https://blog.bummon.com/
[Bummon_s Home]: https://www.bummon.com/
[Link 1]: https://www.bummon.com/wechat/