RabbitMQ高阶使用消息推送

本文涉及的产品
云数据库 MongoDB,独享型 2核8GB
推荐场景:
构建全方位客户视图
简介: RabbitMQ高阶使用消息推送

1 从打车开始说起

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


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

1.1 需要解决的问题

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

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

1.2 消息推送

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

2 消息推送

2.1 什么是消息推送

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

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

2.2 方案介绍

2.2.1 ajax短轮询

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

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

2.2.2 长轮询

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

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

2.2.3 WebSocket

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

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

2.3 WS实现消息推送

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

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

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

2.3.1 架构介绍

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

  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进行处理

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;
    }
}


相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
消息中间件 存储 监控
【图解RabbitMQ-3】消息队列RabbitMQ介绍及核心流程
【图解RabbitMQ-3】消息队列RabbitMQ介绍及核心流程
405 0
|
消息中间件 Java Maven
消息中间件系列教程(12) -RabbitMQ-消息确认机制
消息中间件系列教程(12) -RabbitMQ-消息确认机制
78 0
|
小程序 API 开发工具
小程序消息推送入门 (订阅消息推送)
小程序消息推送入门 (订阅消息推送)
620 0
|
消息中间件 运维 算法
RabbitMQ高阶使用延时任务
RabbitMQ高阶使用延时任务
165 0
|
消息中间件 存储 JSON
RabbitMQ消息模型之FanoutExchange消息模型实战
RabbitMQ消息模型之FanoutExchange消息模型实战
RabbitMQ消息模型之FanoutExchange消息模型实战
|
消息中间件 Java Kafka
RabbitMQ安装以及消息模型使用攻略
🍅程序员小王的博客:程序员小王的博客 🍅 欢迎点赞 👍 收藏 ⭐留言 📝 🍅 如有编辑错误联系作者,如果有比较好的文章欢迎分享给我,我会取其精华去其糟粕 🍅java自学的学习路线:java自学的学习路线
230 0
RabbitMQ安装以及消息模型使用攻略
|
消息中间件 前端开发 JavaScript
Websocket Stomp+RabbitMQ实现消息推送
Websocket Stomp+RabbitMQ实现消息推送
556 0
Websocket Stomp+RabbitMQ实现消息推送
|
消息中间件 Linux
消息队列:第五章:RabbitMQ的使用
消息队列:第五章:RabbitMQ的使用
177 0
消息队列:第五章:RabbitMQ的使用
|
消息中间件 测试技术
RabbitMQ消息模型之DirectExchange消息模型实战
RabbitMQ消息模型之DirectExchange消息模型实战
RabbitMQ消息模型之DirectExchange消息模型实战
|
消息中间件 Java Kafka
Springboot最简单的实战介绍 整合kafka-生产者与消费者(消息推送与订阅获取)
Springboot最简单的实战介绍 整合kafka-生产者与消费者(消息推送与订阅获取)
781 0
Springboot最简单的实战介绍 整合kafka-生产者与消费者(消息推送与订阅获取)
下一篇
无影云桌面