Cloud Pub/Sub – Sử dụng trong Python như thế nào?

pip install google-auth google-auth-httplib2 google-api-python-client
pip install --upgrade google-cloud-pubsub

import os
import datetime
import sys
import time
import json
import  math
import random as rd
import pandas as pd
from google.oauth2 import service_account
from google.cloud import pubsub_v1
from google.cloud import bigquery

ROOT = 'D:machine_learning'
ROOT_CERT = os.path.join(ROOT, 'certificates')
ROOT_DATA = os.path.join(ROOT, 'datas')

project_id = "seminar-demo"
topic_name = "demo"
#//D:machine_learnincertificatesseminar-demo-c0b16d489728.json
credentials = service_account.Credentials.from_service_account_file(
    os.path.join(ROOT_CERT, 'seminar-demo-c0b16d489728.json'))

scoped_credentials = credentials.with_scopes(
    ['https://www.googleapis.com/auth/pubsub',
     'https://www.googleapis.com/auth/cloud-platform'])

class Sender:
    def __init__(self, data):
        self.__data = data
        self.__publisher = pubsub_v1.PublisherClient(credentials=credentials)
        # The `topic_path` method creates a fully qualified identifier
        # in the form `projects/{project_id}/topics/{topic_name}`
        self.__topic_path = self.__publisher.topic_path(project_id, topic_name)
    def __publish(self, message):
        future = self.__publisher.publish(self.__topic_path, data=message)
        return future

    def run(self):
        for idx, item in self.__data.iterrows():
            if idx > 0 and idx % 100 == 0:
                time.sleep(30)
            data = json.dumps(item.to_dict())
            data = data.encode('utf-8')
            # When you publish a message, the client returns a future.
            future = self.__publish(data)
            print('Published {} of message ID {}.'.format(data, future.result()))

class Receiver:
    def __init__(self):
        self.__subscriber = pubsub_v1.SubscriberClient(credentials=credentials)
        self.__subscription_name = 'sub_demo'
        # The `subscription_path` method creates a fully qualified identifier
        # in the form `projects/{project_id}/subscriptions/{subscription_name}`
        self.__subscription_path = self.__subscriber.subscription_path(
            project_id, self.__subscription_name)

    def __callback(self, message):
        print('Received message: {}'.format(message))
        message.ack()
    def run(self):
        self.__subscriber.subscribe(self.__subscription_path, callback=self.__callback)
        # The subscriber is non-blocking. We must keep the main thread from
        # exiting to allow it to process messages asynchronously in the background.
        print('Listening for messages on {}'.format(self.__subscription_path))
        while True:
            time.sleep(60)

self.__publisher = pubsub_v1.PublisherClient(credentials=credentials)

self.__subscriber = pubsub_v1.SubscriberClient(credentials=credentials)

def main():
    # read file csv
    # example : D:machine_learningdatasdata.csv
    data = pd.read_csv(os.path.join(ROOT_DATA, 'data.csv'), encoding='unicode_escape')
    sender = Sender(data)
    sender.run()

def main():
    receiver = Receiver()
    receiver.run()