通过Redis消息队列实现大文件处理

Posted 2023-04-16

参考技术A

一、故事背景
1、读取离线文件数据，再通过【离线数据】作为条件，查询第三方接口，返回最终的结果，再入库。
2、 业务逻辑是很简单， 读取文件、查询接口、返回数据集、入库 四步。
3、业务特性：第三方接口调用400毫秒(ms) 。
如果用普通单线程去跑算500毫秒一个请求，一天也就跑8W多数据量，20多亿的数据不知道跑到猴年马月了。

二、处理方案
A) 初步方案采用ganymed-ssh2(文件都存储在Linux服务器上) 来读文件，Redis来存储消息、多线程来提升处理能力。
B) 流程图：

三、呈现问题

四、优化问题

最终流程图：

1、 通过Redis做一个计数器 每读取一行记录数值，即使服务终止后，先从Redis读取这个数值
再通过cat指定行数开始读数据即可。

2、 通过取模拆Key 分片到不同小Key存储 ，降低单个节点存储压力，也充分利用了存储资源。

3、Redis Push 提供了批量方式(leftPushAll) ，可以指定读取行数再批量入库，而pop并没有提供批量 只能一个一个pop。

4、消费者通过多线程pop、再分发到线程去处理。

五、总结问题

Kafka，Mq和Redis作为消息队列使用

参考技术A kafka是个日志处理缓冲组件，在大数据信息处理中使用。和传统的消息队列相比较简化了队列结构和功能，以流形式处理存储（持久化）消息（主要是日志）。日志数据量巨大，处理组件一般会处理不过来，所以作为缓冲层的kafka，支持巨大吞吐量。为了防止信息丢失，其消息被调用后不直接丢弃，要多存储一段时间，等过期时间过了才丢弃。这是mq和redis不能具备的。主要特点如下:巨型存储量: 支持TB甚至PB级别数据。高吞吐，高IO：一般配置的服务器能实现单机每秒100K以上消息的传输。消息分区，分布式消费：能保消息顺序传输。 支持离线数据处理和实时数据处理。Scale out：支持在线水平扩展，以支持更大数据处理量

redis只是提供一个高性能的、原子操作内存键值对，具有高速访问能力，可用做消息队列的存储，但是不具备消息队列的任何功能和逻辑，要作为消息队列来实现的话，功能和逻辑要通过上层应用自己实现。

我们以RabbitMQ为例介绍。它是用Erlang语言开发的开源的消息队列，支持多种协议，包括AMQP，XMPP, SMTP, STOMP。适合于企业级的开发。
MQ支持Broker构架，消息发送给客户端时需要在中心队列排队。对路由，负载均衡或者数据持久化都有很好的支持。

还有ActiveMq，ZeroMq等。功能基本上大同小异。并发吞吐TPS比较，ZeroMq 最好，RabbitMq 次之， ActiveMq 最差。

原文：