RabbitMQ高阶使用消息推送

阳光穿透心脏的1/2处 2024-03-16 19:44 32阅读 0赞

#### 目录 ####

*  1 从打车开始说起
 *   *  1.1 需要解决的问题
     *  1.2 消息推送
 *  2 消息推送
 *   *  2.1 什么是消息推送
     *  2.2 方案介绍
     *   *  2.2.1 ajax短轮询
         *  2.2.2 长轮询
         *  2.2.3 WebSocket
     *  2.3 WS实现消息推送
     *   *  2.3.1 架构介绍
         *  2.3.2 暂存数据
         *   *  2.3.2.1 什么是MongoDB
             *  2.3.2.2 插入数据
             *  2.3.2.3 查询数据
         *  2.4.1 轮询任务
         *   *  2.4.1.1 代码实现

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

## 1 从打车开始说起 ##

![在这里插入图片描述][f4e11270147241bfb2fd894de57a1365.png]

![在这里插入图片描述][d8dc4f74df6a4370ac7943bffef0e808.png]

> 我们把滴滴打车的流程简化下

![在这里插入图片描述][8e42ecdcf2254019a4d1139323a4596a.png]

1.  登录app后点击打车开始进行打车
2.  打车服务开始为司机派单
3.  司机接单后开始给来接驾
4.  上车乘客后处于行程中
5.  行程结束后完成本次打车服务

### 1.1 需要解决的问题 ###

> 我们需要实现派单服务，用户发送打车订单后需要进行进行派单，如果在指定时间内没有找到司机就会收到派单超时的通知，并且能够实时查看当前排队的抢单人数

![在这里插入图片描述][890b5c0c3c5345a5abeb4de736cd63d1.png]

> 下面我们来介绍下涉打车涉及到的一些问题

### 1.2 消息推送 ###

> 我们需要将我们的异步处理结构返回到客户端，我们的客户端是使用的websocket连接的，因为websocket是点对点连接的，连接到一台固定的通知服务后，只能从这一台通知服务来获取数据，因为我们的通知服务允许分布式部署，这个问题改如何解决？

![在这里插入图片描述][c5afbfd47acb4f56a3f148efa9ddaa3f.png]

## 2 消息推送 ##

### 2.1 什么是消息推送 ###

> 当我们使用http协议探知服务器上是否有内容更新，就必须频繁的从客户端到服务器端进行确认，而http一下的这些标准会成为一个瓶颈：

*  一条连接上只可以发送一个请求
 *  请求只能从客户端开始，客户端不可以接收除了响应以外的指令
 *  请求 / 响应首部未经过压缩就直接进行传输，首部的信息越多，那么延迟就越大。
 *  发送冗长的首部，每次互相发送相同的首部造成的浪费越多
 *  可以任意选择数据压缩格式，非强制压缩发送

### 2.2 方案介绍 ###

#### 2.2.1 ajax短轮询 ####

> Ajax（异步的javascript与xml技术）是一种有效利用javascript和dom的操作，以达到局部web页面的提花和加载的异步通信手段，和以前的同步通信相比，他只更新一部分页面，相应中传输饿数据量会因此的减少

![在这里插入图片描述][6310e3f44bbf47b6a89cea87d0b52f49.png]

Ajax轮询的原理是，让浏览器每隔一段时间就发送一次请求，询问服务器是否有新消息，而利用ajax实时的从服务器获取内容，有可能导致大量的请求产生。
    
    	特点：实现简单、短连接、数据同步不及时、对服务器资源会造成一定压力，此模式广泛应用于：扫描登录、扫码支付、天气更新等（腾讯、京东、阿里一直都在沿用此技术并日渐成熟和稳定）

#### 2.2.2 长轮询 ####

> 原理和Ajax轮询差不多，都是采用轮询的方式，不过采用的是阻塞模型

![在这里插入图片描述][15af539415f44e529ca22a1fb3cd71c4.png]

也就是说，当客户端发起连接后，如果服务器端内容没有更新，将响应至于挂起状态，一直不回复response给客户端，知道有内容更新，再返回响应。
    
    	虽然可以做到实时更新，但是为了保留响应，一次连接饿持续时间也变长了，期间为了维持连接会消费更多的资源，需要有很高的并发，也就是说同时接待客户的能力
    
    	从上面两种方式中，其实可以看出是再不断的建立http连接，然后等待服务器处理，可以体现出了http的特点：**被动性**，即：请求只能由客户端发起，服务器端不能主动联系客户端。
    
    	特点：无需浏览器或APP端任何单独插件支持、长连接，减少网络(三次)握手和四次挥手、对服务器资源要求较高等，此模式常用于实时消息轮播、金融数据即时刷新、数据图表实时刷新等，JAVA服务器端一般采用Servlet3支持的异步任务、延时结果（DeferedResult）等手段实现。

#### 2.2.3 WebSocket ####

> WebSocket 是 HTML5 开始提供的一种在单个 TCP 连接上进行全双工通讯的协议

WebSocket 使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在 WebSocket API 中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输

![在这里插入图片描述][29d18c9eef9843bfa397ef4cc807a7bf.png]

### 2.3 WS实现消息推送 ###

> 通过上面我们实现了延时任务处理以及派单排队，但是如果将我们的异步处理结果推送给客户端呢

我们就需要使用消息推送技术，需要完成一下功能

*  将消息推送到指定的用户
 *  对于未上线用户需要暂存数据，上线后推送

![在这里插入图片描述][7248db3284034e2f8717236699208a1d.png]

#### 2.3.1 架构介绍 ####

> 因为websocket是点对点的，而服务间调用是轮询的，无法实现微服务之间点对点的消息推送，我们使用定时任务来实现消息推送

![在这里插入图片描述][59bbd89771b246e69a45c14d616a27a6.png]

1.  调用接口先将消息暂存到MongoDB中
2.  轮询任务首先拉取当前在线人员列表
3.  轮询任务通过在线人员列表到MongoDB中拉取在线用户的通知消息
4.  将消息通过WS推送到指定的用户

#### 2.3.2 暂存数据 ####

> 通过MongoDB将我们的消息数据暂存到数据库中，可以完成对于未上线消息暂存以及对分布式websocket的数据调度

##### 2.3.2.1 什么是MongoDB #####

MongoDB是一个介于关系数据库和非关系数据库之间的产品，是非关系数据库当中功能最丰富，最像关系数据库的。它支持的数据结构非常松散，是类似json的bson格式，因此可以存储比较复杂的数据类型。Mongo最大的特点是它支持的查询语言非常强大，其语法有点类似于面向对象的查询语言，几乎可以实现类似关系数据库单表查询的绝大部分功能，而且还支持对数据建立索引。

##### 2.3.2.2 插入数据 #####

@Override
    public void addMessage(PushMessage message) {
        
        mongoTemplate.save(message);
    }

##### 2.3.2.3 查询数据 #####

@Override
    public List<PushMessage> getMessageByUserNames(List<String> userNameList) {
        
        Query query = new Query(Criteria.where("username").in(userNameList));
        return mongoTemplate.findAllAndRemove(query, PushMessage.class);
    }

#### 2.4.1 轮询任务 ####

> 轮询任务就是不断的搜索检查是否有新的消息，然后交给WS进行处理

![在这里插入图片描述][cf24b2e5d41641fbad275ad5bfc26e5b.png]

##### 2.4.1.1 代码实现 #####

> 使用pull方式将MongoDB中的在线用户的暂存消息取出来，推送给在线用户

/**
     * 定时任务 推送暂存消息
     */
    @Component
    public class ScheduledTask {
        
    
        private static final Logger logger = LoggerFactory.getLogger(ScheduledTask.class);
    
        @Autowired
        private PushService pushService;
    
        private static final ExecutorService executorService = Executors.newFixedThreadPool(10);
    
        @Autowired
        private WebSocketServer webSocketServer;
    
        @PostConstruct
        public void init() {
        
            executorService.execute(() -> {
        
                autoPushMessage();
            });
    
        }
    
    
        /**
         * 自动推送消息
         */
        public void autoPushMessage() {
        
            //轮询并发送消息
            PollingRound.pollingPull(() -> {
        
                //获取最新需要推送的消息
                List<PushMessagePO> pushMessagesList = getPushMessages();
                //校验消息
                if (null != pushMessagesList && !pushMessagesList.isEmpty()) {
        
                    logger.debug("推送消息线程工作工作中,推送数据条数:{}", pushMessagesList.size());
                    //推送消息
                    webSocketServer.pushMessage(pushMessagesList);
                    return PollingRound.delayLoop(100);
                }
                logger.debug("推送消息线程工作工作中,推送数据条数:{}", 0);
                return PollingRound.delayLoop(1000);
            });
        }
    
        public List<PushMessagePO> getPushMessages() {
        
            List<String> userNameList = webSocketServer.getInLineAccountIds();
            if (null != userNameList && !userNameList.isEmpty()) {
        
                //在MongoDB中获取当前在线用户的暂存消息
                List<PushMessagePO> pushMessageList = pushService.getMessageByAccountIds(userNameList);
                //返回消息
                return pushMessageList;
            }
            return null;
        }
    }

[f4e11270147241bfb2fd894de57a1365.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/459244e8238e4fd5a0b0618ca62536da.png
[d8dc4f74df6a4370ac7943bffef0e808.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/2c8a017e652f4bb78bee34b3cf321fc6.png
[8e42ecdcf2254019a4d1139323a4596a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/417b2ff879ae41769f4a9612948633b6.png
[890b5c0c3c5345a5abeb4de736cd63d1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/9e50afd26a9d4613b14759c619079ef8.png
[c5afbfd47acb4f56a3f148efa9ddaa3f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/94f6a614029b449c87e5fec252201031.png
[6310e3f44bbf47b6a89cea87d0b52f49.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/ee35e5dc1dbb49d5aefa5d7c303cb7b0.png
[15af539415f44e529ca22a1fb3cd71c4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/5b2b3957dda440819322a95aff51ecbd.png
[29d18c9eef9843bfa397ef4cc807a7bf.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/c7aa6662e3ac479b84fa9169f6dc9f41.png
[7248db3284034e2f8717236699208a1d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/2d2e39e39253455e9ee616e974a0f0d1.png
[59bbd89771b246e69a45c14d616a27a6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/01424a366f9e4d53b4303ce5814f6dbb.png
[cf24b2e5d41641fbad275ad5bfc26e5b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/d42dafe6a4f24fd59ad2942a9f5fc184.png