На диаграмме выше показана текущая архитектура, которую я пытаюсь реализовать.В настоящее время я запускаю zookeeper и kakfa (впервые для обоих) на удаленном сервере, и это приложение-служба уведомлений создается с использованием flask, который выступает в роли потребителя (kafka-python), а другое приложение App1..Appn будет выступать в качестве производителя.Ну, я могу независимо управлять потребителем и производителем, которые работают нормально.Но я пытался связать потребителя со службой уведомлений, ничего не происходит.
app.py
def send_notification_from_consumer(payload):
# sending notification using firebase
if __name__ == '__main__':
app.run(host='127.0.0.1', port=8000, debug=True)
consumer.py
from kafka import KafkaConsumer
import json
from app import send_notification_from_consumer as snc
def notification_consumer():
print("---inside consumer-->")
consumer = KafkaConsumer(
'send-notification',
bootstrap_servers=['remote_server_ip:9092'],
auto_offset_reset='earliest',
enable_auto_commit=True,
group_id='my-group',
value_deserializer=lambda m: json.loads(m.decode("utf-8")))
for message in consumer:
print("----->",message.value)
snc(message.value)
producer.py
from kafka import KafkaProducer
import json
producer = KafkaProducer(bootstrap_servers=['remote_server_ip:9092'],
value_serializer=lambda m: json.dumps(m).encode('utf-8'))
def on_send_success(record_metadata):
print(record_metadata.topic)
print(record_metadata.partition)
print(record_metadata.offset)
def on_send_error(excp):
log.error('I am an errback', exc_info=excp)
for i in range(1):
name = {"phone_number_list": ["1234567890", "1234567891"],
"payload": {"title":"Hello from another ",
"description":"Test notification by notification service",
"banner_image":"", "id":123, "action":{"type":"web", "button_label":"", "link":""},
"app_name":"app_name"}
}
producer.send('send-notification', name).add_callback(on_send_success).add_errback(on_send_error)
Можно ли вызвать потребителя без использования HTTP-запроса, потому что я просто вызываю send_notification_from_consumer в kafkaпотребитель.Как правильно это сделать?Как мне запустить потребителя внутри колбы.
Я тоже пробовал это решение Но я получил эту ошибку.Может ли это решение масштабироваться?
RuntimeError: Working outside of request context.