如何从零开始写一个Python MQ Client

忘是亡心i 2022-09-14 02:58 172阅读 0赞

本文转载已经过授权，原文链接：

[如何从零开始写一个Python MQ Client\_bianlifengtech的博客-CSDN博客1.背景公司的主要开发语言是Java，算法部门主要使用的语言为Python。算法应用经常需要订阅业务系统产生的各种消息，但是业务部门使用的消息队列却为开源的Qmq。Qmq原生并没有提供对Python的支持，因此需要编写一个Python QmqClient。本文虽然是以Qmq为原型，但是其他大多数的MQ Client基本上是相同的套路，还是具有普适意义的。2.需求分析从功能角度看，Client可以分为Consumer以及Producer两个模块。由于算法需求主要为订阅消息，所以选择优先完成Con.![favicon32.ico][]https://blog.csdn.net/bianlifengtech/article/details/120351427?spm=1001.2014.3001.5501][Python MQ Client_bianlifengtech_-CSDN_1._Java_Python_Qmq_Qmq_Python_Python QmqClient_Qmq_MQ Client_2._Client_Consumer_Producer_Con._favicon32.ico_https_blog.csdn.net_bianlifengtech_article_details_120351427_spm_1001.2014.3001.5501]

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

## **1.背景** ##

公司的主要开发语言是Java，算法部门主要使用的语言为Python。算法应用经常需要订阅业务系统产生的各种消息，但是业务部门使用的消息队列却为开源的Qmq。Qmq原生并没有提供对Python的支持，因此需要编写一个Python Qmq Client。本文虽然是以Qmq为原型，但是其他大多数的MQ Client基本上是相同的套路，还是具有普适意义的。

## **2.需求分析** ##

从功能角度看，Client可以分为Consumer以及Producer两个模块。由于算法需求主要为订阅消息，所以选择优先完成Consumer模块，再完成Producer模块。

从职责角度分析，Client可以分为以下四个部分：

1.  网络
2.  序列化和反序列化
3.  线程控制
4.  交互逻辑

本文从上面几个部分分别入手，逐个讲解分析实现网络、序列化反序列化、线程控制、交互逻辑四大部分需要了解的原理性知识和可能遇到的问题以及实现过程中如何解决这些问题，从而帮助大家更好的实现自己的Client。

## **3.网络** ##

此部分的讲解网络处理，主要涉及数据包的处理，response响应回调，网络连接管理和框架选择。

## **3.1 数据包处理** ##

在网络编程的数据包处理中，我们主要面临两个核心问题：

`确定request所对应的response`

这个问题的意思是，客户端发送一个请求给服务端，服务端会回一个响应。但实际中客户端会发送很多请求给服务端，服务端也会回复很多响应给客户端。但是我们要将客户端回复的响应与客户端发送的请求要对应上。我们有两种途径解决这个问题：

第一种为串行化发送request，串行化接收response， request和response的对应是依赖顺序的。典型的例如redis的所采用的方式，比如向redis server发送get key1, get key2, get key3，那么redis server回复的响应是 value of key1, value of key2, value of key3。

第二种为通过request id映射。此种需要在报文中定义一个字段用来存放此request请求的id，服务端响应此request的response报文中也有一个id字段，并且此字段和request中的id字段相等。这样我们就能够去确定request-response 之间的对应关系了：发送request的时候生成一个id，并有一个map维护id <-> request之间的关系，当收到response的时候从response里取出id去map里获取request。 这里Qmq使用的是第二种方式。两种方式并无绝对的优劣，只是选择不同而已。

`粘包/半包/拆包`

粘包和半包的说法并不准确，因为TCP是面向流的，何来粘和半一说呢。面向流是什么意思呢？我们可以类比河流，河里的水是源源不断地，它没有明确的界限。但是在上层的应用中我们发送一个数据，往往是有界限的。比如我们这里的MQ，那么一般我们把一条消息作为一个单元，发送端发送了两条消息，我们要从TCP这条河流里读取出两条消息

处理此问题一般有下列几种方法：

*  消息定长，报文大小固定，不够的空格补全。
 *  使用特殊分隔符。例如FTP协议、redis报文协议等则采用回车换行符来作为报文的符。
 *  在消息头中包含表示信息总长度（或者消息体总长度）的字段，通过解析消息头中消息体的长度配合缓冲区来解决粘包/拆包问题。

了解了一般性的解决方案，我们来看看Qmq他是怎么解决这两个问题的呢？qmq使用自定义格式的报文协议来进行网络传输。自定义报文分为消息头header和消息体body，对于不同目的的请求request， body的编码解码格式不同。qmq的报文格式如下：

`qmq报文格式`

![7dda9a1bd931b2a1aeb769b55764140f.png][]

[favicon32.ico]: /images/20220828/f5cce289dab246a2b23f3d249178cd2b.png
[Python MQ Client_bianlifengtech_-CSDN_1._Java_Python_Qmq_Qmq_Python_Python QmqClient_Qmq_MQ Client_2._Client_Consumer_Producer_Con._favicon32.ico_https_blog.csdn.net_bianlifengtech_article_details_120351427_spm_1001.2014.3001.5501]: https://blog.csdn.net/bianlifengtech/article/details/120351427?spm=1001.2014.3001.5501
[7dda9a1bd931b2a1aeb769b55764140f.png]: /images/20220828/cd320a1774e24b17b5f9fee1d5b9dd58.png