メールマーケティングサービスを構築する目的は何ですか？ 内部ビュー、パート2

完全なメールマーケティングサービスを構築するのはどれくらい難しいですか？ このために何を想定する必要がありますか？ 開発者の心を探求する過程で、どんな落とし穴に遭遇する可能性がありますか？



一緒に考えてみましょう。 いくつかの記事で、私が1年以上にわたって自分のメールニュースレターサービスをどのように行っているか、どのような教訓を自分のために学んだか、そしてこれらすべてで次に何をするつもりかについてお話します。

記事が問題の技術面のみを考慮していることを直ちに予約してください。

最初の部分はここで読むことができます 。

自分について簡単に

私は5年間Pythonで書いています。主にPostgreSQLのDjangoを使用しています。jQuery+ KnockoutJSレベルでJavaScriptを準備できます。Reactで書くこともあります。 空き時間には、UpWorkと自分のインターネットプロジェクトでフリーランスの仕事をしますが、そのうちの1つについては今から話します。 私はこのプロジェクトに約1年半携わっています。

この記事の目的は何ですか？

最初の部分では、製品（Python、Django、PostgreSQL）の主な技術として使用する技術、電子メール追跡の仕組み、およびサービスに関する統計情報について簡単に説明しました。

このパートでは、次のことを伝えます。

1.非同期タスクを使用してユーザーを苦痛から守る方法。 Celeryを使用して複数のタスクを同時に実行することを検討してください。
2.リソースが非常に限られているサーバーで、数百万のレコードを含むテーブルを使用して作業を編成した方法。 非正規化されたテーブルが適切な場合がある理由を説明します。
3.モノリシックDjangoプロジェクトの機能の一部を個別のマイクロサービスに移行し、これに使用したテクノロジーを最小限の時間コストで迅速に管理した方法。

それでは始めましょう。

Pythonプロジェクトの非同期タスク、またはユーザーが待たない理由

遅かれ早かれ、特に同期モードでユーザーブラウザーからの要求を処理するWebベースのアプリケーションについて話している場合、成長するプロジェクトはすべてパフォーマンスの問題に直面します。

最も簡単な例

pythonプロジェクトに典型的なテクノロジースタックを使用するとしましょう：uwsgiの新しいビルド、nginx、Webサーバー+ pythonアプリケーション。 Nginxは静的ファイルを提供し、ルートをuwsgiに転送します。 uwsgi構成では、N人のワーカーがいます。

これらはすべて、大量の「重い」リクエスト（大きなファイルの読み込みなど）に遭遇するまでスムーズに機能します。これには、harakiri uwsgiに近い、またはそれより長い処理時間が必要になります。 その後、uwsgiワーカーはユーザーからのリクエストを処理する時間がなくなり、ユーザーはnginx 502からエラーを受け取り、「サービス」に関するコメントを誓います。

もちろん、トルネード、asyncio + aiohttpなどのソリューションに頼ることもできますが、これは少し異なるものです。既に同期Djangoアプリケーションがあり、書き換えを破棄するのは非常に残念です。

好奇心reader盛な読者は、nginxに複数のアップストリームを使用することでDjangoとaiohttpを組み合わせることができると言うでしょうが、今はもっと標準的なアプローチを検討したいと思います。

現在、同期Pythonプロジェクトの最も標準的なアプローチは、ライブラリを使用してCelery非同期タスクを処理することです。



以下のすべての情報は、バージョン3.xで有効です。 私はそれを使用します。 Celeryのバージョン4では、多くが変更されました。

このライブラリにまだ慣れていない読者のために、以下の方法を知っているという簡単なコメント：

1. Celeryが非同期に実行する（またはbeatを使用する必要があるときに実行するために実行する）タスク（通常の機能）を作成できます。

システム電子メールを送信する単純なタスクの例：

from celery import current_app ... @current_app.task() def send_service_message(subject, recipients, template, html_template, context): html_mail( subject=subject, template=template, html_template=html_template, recipients=recipients, context=context ) 

このタスクの課題は次のとおりです。

 from app.message.tasks import send_service_message ... send_service_message.delay( _(u'%s:  ' % subscriber.list.project.domain), [user.email], '', 'site/subscriber/email/subscription_notification.html', context ) 

2. Djangoを使用すると、これらのタスクを管理するための非常に便利なインターフェイスが得られます。これにより、たとえば、実行時の各タスクの起動パラメーターを定義できます。

3.このシステム全体は、ほとんどの場合、RabbitMQ（メッセージブローカーとして使用）に基づいて実行され、ほとんどの場合、supervisordを使用して制御されます。 また、データベースはブローカー（最も遅いオプションですが、開発には適しています）、Redis（プロトコルの維持が非常に難しいため、Celeryの作成者は使用を推奨しません）、Amazon SQSとして使用できます。

私のプロジェクトでの使用例

私のプロジェクトでは、ユーザーにメールニュースレターとトランザクションメールサービスを提供しています。
したがって、潜在的に2種類の非同期タスクがあります。

1.長いタスク。 郵便物はそれぞれ20から25000通の手紙で、手紙の送信に必要なタイムアウトのために6-7時間以内に実行できます。
2.短いタスク。 ユーザーがサイトに登録し、そのバックエンドが登録確認を記載したメールを私のサービスのAPIに送信するリクエストを送信しました。数秒後（またはそれ以上の速さ）に、ユーザーは受信トレイにこの手紙を見るはずです。

1つのキュー内ですべてのタスクの計画と実行を忘れると（つまり、デフォルトで）、すでに実行されている長いタスクが既にある場合、すべての短いタスクはその完了を待ちます。
6〜7時間後にトランザクションレターを受け取ることはオプションではないことは明らかであるため、タスクの並列処理に複数のキューを使用することになります。

これを行うには、プロジェクトでキュー自体を構成します。

 # settings.py ... CELERY_QUEUES = ( Queue('celery', Exchange('celery'), routing_key='celery'), Queue('bulk', Exchange('bulk'), routing_key='bulk'), Queue('transactional', Exchange('transactional'), routing_key='transactional'), ) CELERY_ROUTES = { 'app.campaign.tasks.start_campaign': { 'queue': 'bulk' }, 'app.message.tasks.send_service_message': { 'queue': 'transactional' }, 'app.message.api.tasks.send_message': { 'queue': 'transactional' } } 

私のプロジェクトでは、3つのキューを使用します。

1.トランザクション-トランザクションレターを送信するためのキュー
2.バルク-ニュースレターを送信するためのキュー
3.セロリ-残りのタスクの順番。

したがって、supervisor.dでは、構成は次の形式を取ります。

 [program:celery] command=<PROJECT_ROOT>/venv/bin/celery worker -A conf.celery -l INFO --concurrency=4 --pidfile=/var/run/celery/celery.pid -Q celery ... [program:bulk] command=<PROJECT_ROOT>/venv/bin/celery worker -A conf.celery -l INFO --concurrency=4 --pidfile=/var/run/celery/bulk.pid -Q bulk ... [program:transactional] command=<PROJECT_ROOT>/venv/bin/celery worker -A conf.celery -l INFO --concurrency=4 --pidfile=/var/run/celery/transactional.pid -Q transactional ... 

ここでは多くの設定オプションが省略されていることが理解されます。 要するに、各ターンでスーパーバイザーでプロセスを実行する必要があるということです。

特定のキューにタスクを設定する方法は次のとおりです。

 send_message.apply_async(kwargs={'mailer': self, 'message': message}, queue='transactional') 

その結果、複数のタスクを同時に処理でき、ユーザーは満足しているように見えます。サービスは十分に高速に動作し、uwsgiワーカーは無料です。

タスクステータスの追跡

タスクのステータスの表示は非常に簡単です。

 celery -A conf.celery inspect scheduled celery -A conf.celery inspect active celery -A conf.celery inspect reserved 

また、より便利な監視のために、 Flowerソリューションを使用できます。

私は自分自身をセロリの超専門家とは考えていません。読者がそれを使った経験を共有してくれたら嬉しいです。

大きなテーブルの非正規化-長所と短所

遅かれ早かれ、データを蓄積します。 そして、彼らは何とかして働かなければなりません。
私のサービスは1か月あたり約40万件のメールを処理します（これは非常に小さいです）が、90日以上前のメールに関するデータを消去していますが、追跡データを含むテーブルには約200万件のレコードが残っています。

これらのデータは統計に表示する必要があり、それらを操作する必要があり、スパマーを特定するためにそれらを分析する必要があります。

最初は、SQLのすべての標準に従って構築されたテーブルが超正規化されました。 しかし、現実には、ORM Djangoはテーブル間のかなり複雑なJOINのために、あなたとあなたのプロジェクトの生活を困難にすることがよくあります。
さらに、特に初心者デベロッパーの場合、クエリセットを繰り返すときにクエリがいくつ発生するかは必ずしも明らかではありません。

その結果、かなり単純なクエリを計算するには：

 SELECT d.date, SUM(CASE WHEN me.status = 'SENT' THEN 1 ELSE 0 END) AS sent_count, SUM(CASE WHEN me.status = 'OPENED' THEN 1 ELSE 0 END) AS opened_count, SUM(CASE WHEN me.status = 'REDIRECTED' THEN 1 ELSE 0 END) AS redirected_count, SUM(CASE WHEN me.status = 'HARDBOUNCED' THEN 1 ELSE 0 END) AS hardbounced_count, SUM(CASE WHEN me.status = 'SOFTBOUNCED' THEN 1 ELSE 0 END) AS softbounced_count, SUM(CASE WHEN me.status = 'UNSUBSCRIBED' THEN 1 ELSE 0 END) AS unsubscribed_count FROM ( SELECT TO_CHAR(date_trunc('day', ('2017-03-01'::DATE - offs)), 'YYYY-MM-DD') AS date FROM generate_series(0, 30, 1) AS offs ) d LEFT OUTER JOIN MESSAGE_MESSAGEEVENT me ON (d.date=to_char(date_trunc('day', me.date_created), 'YYYY-MM-DD') AND status IN ('SENT', 'OPENED', 'REDIRECTED', 'HARDBOUNCED', 'SOFTBOUNCED', 'UNSUBSCRIBED') AND id IN (...)) GROUP BY date ORDER BY date 

約100秒かかりました。 恐ろしい時間ですね。 これはいくつかの理由で起こりました。

1.最初に、これを絶対に行わないでください： AND id IN（...）配列に約100,000個の要素がある場合:)
2.第二に、 d.date = to.char（date_trunc（ 'day'、me.date_created）、 'YYYY-MM-DD'）よりもd.date = me.date_created :: dateと書く方がはるかに良いです。
3.第三に、表には他のいくつかのキーの外部キーがありました。 JOINの動作が非常に遅いため。

何に来たの

1.外部キー（部分的に非正規化されたテーブル）に加えて、関連レコードの明示的な値のフィールドを追加しました。
2.リクエストを書き直しました。

 SELECT d.date, SUM(CASE WHEN me.status = 'SENT' THEN 1 ELSE 0 END) AS sent, SUM(CASE WHEN me.status = 'OPENED' THEN 1 ELSE 0 END) AS opened, SUM(CASE WHEN me.status = 'HARDBOUNCED' THEN 1 ELSE 0 END) AS hardbounced, SUM(CASE WHEN me.status = 'SOFTBOUNCED' THEN 1 ELSE 0 END) AS softbounced, SUM(CASE WHEN me.status = 'UNSUBSCRIBED' THEN 1 ELSE 0 END) AS unsubscribed, SUM(CASE WHEN me.status = 'REDIRECTED' THEN 1 ELSE 0 END) AS redirected FROM ( SELECT ('2017-03-01'::DATE - offs)::date AS date FROM generate_series(0, 30, 1) AS offs ) d LEFT OUTER JOIN MESSAGE_MESSAGEEVENT me ON ( d.date=me.date_created::date AND me.date_created >= '2017-02-01 00:00:00' AND me.date_created <= '2017-03-01 23:59:59' AND true = true AND me.project_id = ... ) GROUP BY date ORDER BY date; 

その結果、この要求の処理時間は3秒に短縮されましたが、これはすでにかなり耐えられます。 説明する：

 GroupAggregate (cost=61552.47..82232.11 rows=200 width=11) (actual time=2445.092..2986.524 rows=31 loops=1) -> Merge Left Join (cost=61552.47..73614.51 rows=491920 width=11) (actual time=2445.048..2857.069 rows=69314 loops=1) Merge Cond: ((('2017-03-01'::date - offs.offs)) = ((me.date_created)::date)) -> Sort (cost=62.33..64.83 rows=1000 width=4) (actual time=0.127..0.174 rows=31 loops=1) Sort Key: (('2017-03-01'::date - offs.offs)) Sort Method: quicksort Memory: 26kB -> Function Scan on generate_series offs (cost=0.00..12.50 rows=1000 width=4) (actual time=0.034..0.069 rows=31 loops=1) -> Materialize (cost=61490.14..62719.94 rows=98384 width=15) (actual time=2444.913..2695.888 rows=69312 loops=1) -> Sort (cost=61490.14..61736.10 rows=98384 width=15) (actual time=2444.908..2539.290 rows=69312 loops=1) Sort Key: ((me.date_created)::date) Sort Method: external merge Disk: 2152kB -> Bitmap Heap Scan on message_messageevent me (cost=11442.78..51647.59 rows=98384 width=15) (actual time=1922.446..2253.982 rows=69312 loops=1) Recheck Cond: (project_id = ...) Filter: ((date_created >= '2017-02-01 00:00:00+01'::timestamp with time zone) AND (date_created <= '2017-03-01 23:59:59+01'::timestamp with time zone)) -> Bitmap Index Scan on message_messageevent_b098ad43 (cost=0.00..11418.19 rows=236446 width=0) (actual time=1870.742..1870.742 rows=284445 loops=1) Index Cond: (project_id = ...) Total runtime: 2988.107 ms 

このソリューションの欠点は何ですか？

まず、テーブルはより多くのディスク容量を占有します。 第二に、非正規化されたテーブルの文字列値が関連する必要があります。 つまり 親テーブルの文字列識別子を変更する場合は、子の非正規化テーブルでも文字列識別子を変更する必要があります。 幸いなことに、Djangoの移行により、これを非常に簡単に実装できます。

モノリシックプロジェクト機能のマイクロサービスへの移行

多かれ少なかれ大規模なプロジェクトを成長させることの問題の1つは、維持することがますます難しくなることです。 コードベースは成長し、テストの数は増加しています。 さらに、プロジェクトの一部でミスをすると、他の人がそのために苦しむ可能性があります。 たとえば、ランダムなSyntaxErrorはプロジェクト全体を圧倒します。
また、既に説明した大量のクエリに関連するパフォーマンスの問題は、ますます重要になっています。

解決策は、機能の一部を個別のマイクロサービスに分離することです。

1.この機能に関連するコードベースとテストを別のプロジェクトに入れます。
2.個別のuwsgiインスタンスと、nginxの個別のアップストリームまたはサーバーを作成します。
3. RabbitMQやHTTP APIなどのブローカーを使用してメインプロジェクトに接続します。
...
4.利益！

マイクロサービスアーキテクチャの長所：

1.いずれかの機能部分のエラーによりプロジェクトが失敗する危険性はなくなりました。
2.他のプロジェクトで再利用できます。 特に、あらゆる種類の短縮リンク、画像圧縮機などのアプリケーションマイクロサービスに関連しています。
3.サービスの依存関係を互いに独立して更新できます。 たとえば、メインプロジェクトでDjangoのレガシーバージョンを使用する場合、重要なコードベースを適合させる必要なく、マイクロサービスで新しいバージョンを使用できます。

私の場合、プロジェクトを管理するために簡単な管理パネルを実装する必要がありました。 ユーザーによるサービスの使用に関するかなり重いレポートが含まれていたため、個別に動作する必要があり、メインプロジェクトのuwsgiワーカーを攻撃したくありませんでした。

私が使用した技術：

1.メインプロジェクトの一部としてHTTP APIを構築するためのDjango Rest Framework （以降-DRF）。
2.管理システムのクライアント-ReactJS、NodeJS、Babel、ES6、およびフロントエンドの世界から私には理解できない言葉を書くために。

ビルドAPI

DRFを使用すると、既存のプロジェクトで便利なREST APIを迅速に構築できます。 私の場合、次の要件を満たすインターフェースを作成する必要がありました。

1. APIハンドルへのアクセスは安全でなければなりません。 管理者権限を持つ特定のユーザーのみが管理システムにアクセスできます。
2.インターフェイスは、クライアントSPAアプリケーションに便利な対話方法を提供する必要があります。
3.すぐに使用できるように、ペンはオブジェクトのリスト、特定のオブジェクトに関する情報の取得、オブジェクトの削除などに機能する必要があります。 一般に、APIはクライアント側で通常のCRUDインターフェイスを構築するのに十分なはずです。

ログイン

クライアントからの要求を承認するために、DRFの既成のソリューション-トークン認証を使用しました。

それを有効にするために必要なすべて：

1.「rest_framework.authtoken」をINSTALLED_APPSに追加します。
2.移行を実行します。
3.管理システムへのアクセス権が必要なユーザーのトークンを作成します。

次に、必要なビューセットにミックスする共通のミックスインを作成します。

 # permissions.py from rest_framework.permissions import IsAdminUser class IsProjectAdminUser(IsAdminUser): def has_permission(self, request, view): return request.user.is_authenticated and request.user.is_admin 

 # views.py from rest_framework.authentication import TokenAuthentication from app.contrib.api.permissions import IsProjectAdminUser class AdminAuthMixin(object): authentication_classes = [TokenAuthentication] permission_classes = [IsProjectAdminUser] 

さらに、このミックスインを使用して、承認をViewSetに接続するだけです。

 ... from app.contrib.api.views import AdminAuthMixin, MultiSerializerViewSet ... class CampaignsViewSet(AdminAuthMixin, MultiSerializerViewSet): queryset = Campaign.objects.filter(date_archived__isnull=True) filter_class = CampaignsFilter serializers = { 'list': CampaignsListSerializer, 'retrieve': CampaignDetailSerializer, } 

顧客インタラクション

axiosライブラリは、クライアント側でAPIとやり取りするのに最適でした。

APIリクエストの例：

 axios.get('http://example.com/api/entrypoint/', {params: this.state.query_params}) .then(response => { const items = response.data.results.map(obj => obj); this.setState({ is_loading: false, items: items, count: response.data.count }); }); 

Reactアプリケーションのコンポーネントのコード全体を提供しません。なぜなら、 それはかなり標準です。

PS

以下の記事では、logstash、kibana、elasticsearchに基づいた大規模プロジェクトで、苦労せずにログを整理する方法、およびHelpScoutおよびGitBookソリューションに基づくクライアントのドキュメントとサポートに触れる方法について説明します。

ご清聴ありがとうございました！