[ DevOps ] - AWS 기반 대규모 아키텍트 설계 - 데이터복제 이벤트 처리 Internal HTTP

이번에는 Flask의 main쪽을 수정을 해 보면서, 내부 HTTP연결을 통해서 두 서비스간에 소통을 하는 방법에 대해 간단히 알아보겠습니다.

 

그래고 저번시간에 코드도 잘못된 것이 있으므로 최종으로 코드의 틀을 완성하고 테스팅하는 시간이라고 보시면 됩니다.

 

/boss/main.py

from flask import Flask, jsonify, abort
from flask_sqlalchemy import SQLAlchemy
from flask_cors import CORS
from dataclasses import dataclass
from producer import publish
import requests

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:root@db/main'

# 추가적으로 메모리를 필요로 하는 작업을 비활성화 한다.
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

CORS(app)

db = SQLAlchemy(app)

@dataclass
class Shop(db.Model):
    id: int
    shop_name: str
    shop_address: str
    
    id = db.Column(db.Integer, primary_key=True, autoincrement=False)
    shop_name = db.Column(db.String(200))
    shop_address = db.Column(db.String(200))


@dataclass
class Order(db.Model):
    id: int
    shop: str
    address: str
    
    id = db.Column(db.Integer, primary_key=True, autoincrement=False)
    shop = db.Column(db.Integer)
    address = db.Column(db.String(200))
    

@app.route('/api/shop')
def index():
    return jsonify(Shop.query.all())

@app.route('/api/order/<int:id>/deliver_finish', methods=['POST'])
def deliver_finish(id):
    publish('order_deliver_finish', id)
    return jsonify({'message': 'success'})

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5001)

우선 :8001 /api/shop은 boss의 mysql에 저장된 shop의 목록을 확인하는 것입니다. 이르 만든 이유는 order서비스에서 DB를 수정하거나 없애면 이게 RabbitMQ를 통해서 반영이 되는지 테스팅하기 위함입니다.

 

그리고 :8001 /api/order/<int:id>/deliver_finish는 사장님들이 배달완료 버튼을 누른것입니다. 나중에 이것을 하면 RabbitMQ를 통해서 order의 DB에서 진짜로 배달완료로 바뀌었는지 확인해 보는 작업을 해볼겁니다,

 

이를 하기 위해서는 첫번쨰는 할건 없지만, 두번째 배달 확전 기능을 구현하기 위해서는, consumer.py를 order에서도 boss와 같이 작성해 주어야 합니다.

 

/boss/consumer.py

from pika import URLParameters, BlockingConnection
import os, django, json

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'order.settings')
django.setup()

from user_order.models import Order, Shop


# 파라미터 설정
params = URLParameters('amqps://vjmczupx:MOf1ZIEhCOj58rh-oSQd0Mrg_BpiVfUD@dingo.rmq.cloudamqp.com/vjmczupx')

# 커넥션 ( 인스턴스에 접근 하는 것 )
connection = BlockingConnection(params)

# 통로 ( 채널 )을 뚫어주는 것
channel = connection.channel()

# 나는 order로 보낸 메세지만 받겠다라는 으미
channel.queue_declare(queue='order')

def callback(ch, method, properties, body):
    print('Received in order')
    id = json.loads(body)
    order = Order.objects.get(id=id)
    order.deliver_finish = 1
    order.save()
    print('order deliver finished')
     
# channel을 basic한 consume을 하겠다는 의미 ( order로 보낸 메세지만 받겠다 )
channel.basic_consume(queue='order', on_message_callback=callback, auto_ack=True)

print('Started consuming')

channel.start_consuming()

channel.close()

또한 consumer로 들어온 정보가 DJANGO의 settings.py의 적용을 받기 위한 설정을 진행해 주었습니다. 그리고 callback은 boss와 비슷한데, body로 받아온 id값을 활용하여 order필드를 찾고, deliver_finish를 1(True)로 바꾸고 저장하는 과정으로 짜주면 됩니다.

 

이제 간단히 postman으로 확인해 보겠습니다.

우선 docker container를 다 실행해 주었습니다.

GET :8000 /api/shop

우선 shop의 목록은 확인이 잘 됩니다.

 

POST :8000 /api/order

또한 잘 order목록이 추가되고 RabbitMQ를 통해서 메세징이 찍히는 것을 보실 수 있습니다.

 

GET :8000 /api/order

그리고 전체 order목록도 잘 확인됨을 보실 수 있습니다.

 

그리고 바로 boss로 가서 이 바뀐 shop정보가 확인되는지 하나만 일단 확인해 보겠습니다.

GET :8001 /api/shops

잘 보시면 위와 다르긴 한데, RabbitMQ를 적용하고 3번만 데이터를 추가했기 때문입니다. 나중에는 DB를 초기화하고 이를 진행하게 된다면, 둘의 동기화가 완벽하게 이루어 질겁니다.

 

그 다음으로는 배달 완료를 눌러보겠습니다. 그럼 아래와 같이 성공 메세지가 뜨게 됩니다.

POST :8001 /api/order/2/deliver_finish

이는 2번 배달 목록을 완료 버튼을 눌러버린 것입니다. 바로 order로 가서 이게 진짜로 반영 되었는지 마지막으로 확인해 보겠습니다.

 

GET :8000 /api/order

다음과 같이 2번 배달 목록의 deliver_finish가 true로 바뀐 것을 보실 수 있습니다.

 

다음부터는 이를 인프라적 관점에서 좀 더 깊게 파고 들어가보겠습니다.