Google PubSub Python clien内存泄漏

Question

我正在使用Python将消息无限期地发布到googlepubsub。我正在分批发布消息。我已将批处理大小设置为20。我在5个过程中发布消息。在

我用1个CPU和3.75内存在GCE中执行下面的脚本。在运行了一个多小时后，GCE内存不足，发布程序崩溃了。在

import json

from multiprocessing import Process
from google.cloud import pubsub

_PROJECT = 'google-project-id'
_TOPIC = 'sample'
_PROCESS_COUNT = 5


def publish(message_str, ps_batch):
    while True:
        ps_batch.publish(message_str)

def main():
    json_message = {'id': 1,'first_name': 'Karim','last_name': 'Bourdice','email': 'kbourdice0@purevolume.com','gender': 'Male','ip_address': '76.108.207.137','field1': '1MCwggd9g267ZBfq8p6KUR8RzXh1UARcBM','field2': '14MbkGizhAuYY9fjzA7UQ1x5sHfPEearMy','field3': '1BC7imHVSwT2xKmZBE6xVwGExXte4M87L7','field4': '1H2CPcuv9JvmnKV3Q9zxtZqZkvn86yyqFz','field5': '1GkanWfbWXxmriDt4vdDq5pb2wySL1nDiZ','field6': '1VKh8m3VcEt5xkfqj9BcCJfuL5nVc3nKZ','field7': '1NuruyCxMeL6YM8ouzD9aqAqFDfHzb6pak','field8': '1DB3Ac7uR5NJziSfJVLM1G2F6tW27SkZqN','field9': '1LdM4nGXPQkSKPTLkZFTbKHN9sAUpBGtSf','field10': '1LLCjdpVYy97u7kBdhHJcgycAZbdZ3R1xF','field11': '16FaG4qFomWMmv4Zv4F8MMUBkcJcEHHijY','field12': '1PXkwtCc4CWSp8BkecAvQubMs12GCvqdVu','field13': '1BqnENPGGDXn9S1FHn2pvwLtDDYu1QTjHL','field14': '1KckpoksBagHxANikRrRBC428srpxAqY5C','field15': '17Z3yPdFgkxEn29JaT89HshvVF4RbtMsMq','field16': '14QpaoakehbRnckiQd7Zte2nSRRGX695Wk','field17': '1wGqgYDSmfE9Q1ZH67GwLqNQmH2u7viRs','field18': '134KjLvr8ugwZK19wzL3V9i8fvmaBBjHQM','field19': '1JrUeWAvufRHXbMLSQeGLkjHJNx3XFM77A','field20': '19Botyc2ikY7V9XqRkJbnoJHWk2eVRxni5','field29': '1DjHXt1yecWxjuNPDMQDaAiCjBvDoEwZr1','field21': '1AZejYQWuuQi9DsGp27tHt7PdCRyBtLD8L','field22': '1AzK3J6vwwh8Xog5TJA48Uh223m5LwqRxz','field23': '12rSBAhFvgFMWbssYmSF3zCpDxmpSPxJV6','field24': '1C1QxVG8QphiRD2uzF6Mg86vRdwTARXpeW','field25': '1GRCqpUZ3oejbic59z16XWUYbw8GYdiq6f','field26': '1CH2yzHjehfuyVW9vw9NPTWX3yW5v2vTFg','field27': '1CCoa5aQPJ4Ya1FDKWrMceRReyptKcWV1N','field28': '1LJSmvzGWwDBLc1YTPaT2k1uAkiX8GjGpz'}
    message_str = json.dumps(json_message)
    batch_size = 20

    ps = pubsub.Client()
    ps_topic = ps.topic(_TOPIC)
    ps_batch = ps_topic.batch(max_messages = batch_size)
    print("Pushing messages to {0}".format(_TOPIC))

    processes = []
    for i in range(1, _PROCESS_COUNT):
        p = Process(target=publish, args=(message_str, ps_batch))
        p.start()
        processes.append(p)

    for p in processes:
        p.join()


if __name__ == "__main__":
    main()

这是stackdriver gce内存使用情况图表。

您可以看到，在运行上述python脚本之后，使用的内存在稳步增加。最后它坠毁了。请注意，我在这个GCE中单独运行这个python脚本，而不是任何其他应用程序

有没有解决这个问题的方法？如何避免内存泄漏？在

Answer 1

您不必自己清除batch.message_ids，而是可以不时调用batch.commit()（而不是纯粹依赖于“在计数器滚动时隐式提交”语义）。在

Answer 2

正如Luke和Tres提到的，这是googlepython客户端的一个问题，他们同意修复这个问题。在

参考https://issuetracker.google.com/u/0/issues/62606697

Answer 3

看看代码，我很确定问题出在这里： https://github.com/GoogleCloudPlatform/google-cloud-python/blob/master/pubsub/google/cloud/pubsub/topic.py#L550

您有一个单独的批处理发布消息，但是返回的消息\u id只是无限期地添加到这个列表中，直到您OOM为止。在

这是个虫子。：—）您可以通过每隔一段时间将batch.message_ids设置为空列表来解决这个问题。在