.NET 6 と Daprを使った分散サービス開発 その10 PubSub

# DaprでPubSub(パブリッシュとサブスクライブ)

前回は、Daprに備わっているBinding (バインディング)を用いて、SMTPのOutput Bindingを設定し、メールを送信する方法について触れました。

**今までの前提の上でバインディングの機能を追加しますので、このページから読まれている方は、前提として以下を読んできてください。**

----
https://qiita.com/kazumihirose/items/aa34373369ce818e8027

https://qiita.com/kazumihirose/items/ff13fa45bf797e6eb878

https://qiita.com/kazumihirose/items/58d0127212068563fe27

https://qiita.com/kazumihirose/items/45b0ea36dd50d63f1d86

https://qiita.com/kazumihirose/items/701fcbfb397fe8670bf7

https://qiita.com/kazumihirose/items/016dde8ca9b35d1030cc

https://qiita.com/kazumihirose/items/e996e3cc34c511bc1124

https://qiita.com/kazumihirose/items/f7811bf3e5abe405a551

https://qiita.com/kazumihirose/items/57e3aeaab65a1a68977b

## DaprとRabbitMQを通じてPub/Sub

様々なアプリケーションを構成するにあたって非同期メッセージングは、結合度を下げ、且つ並列処理とスケーラービリティを確保する上で、とても大事な技術です。

今まで行ってきた、RabbitMQでのBindingでトリガする事と何が違うのか？と思われるかもしれませんが、BindingはQueueだけで無く、主にDaprの外で稼働する様々なサービスとの相互運用性に重きを置いています。

よって、RabbitMQだけでなく、例えばスケジューラもそうですし、Twitterで指定されたハッシュタグでのTweetでトリガするBindingなどがあるのは、そういった背景からです。

今回のPub / Subによるメッセージング処理は、むしろサービス間の非同期メッセージングに重きを置いています。
今まで行ってきたService Invocation(サービス起動）が同期での呼び出しであれば、非同期はこちらのPub/Subを使ってくださいねという意図があります。

例えば、以下の図のようにECサイトをイメージしてもらうとわかりやすいですね。CartでOrderされたら、Shipping(配送手続きAPI）とEmail(メールでの注文確認API）に同時に送るみたいなケースです。

この場合、PublisherはCart で、サブスクライバーはShippingとEmailの二つになり、どちらにDapr経由で呼び出されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/47ac92c7-1b5d-e1db-23aa-d7d1c3d4a039.png)

### プロジェクトを作成、追加

今回は、新たに二つの.NET Web API プロジェクトを追加しました。PubServiceとSubServiceです。プロジェクトのルートフォルダから以下を実行してください。今回は意図的にPubServiceをコンソールプログラムとしてプロジェクトを作成しています。

```bash
dotnet new console -o PubService
dotnet new webapi -o SubService
dotnet new webapi -o SubService2
dotnet sln add PubService SubService SubService2
```

もちろん、忘れずにtye.yamlにもプロジェクトを追加しておき、Daprサイドカーを起動するようにしましょう。

```bash:tye.yaml
# tye application configuration file
# read all about it at https://github.com/dotnet/tye
#
# when you've given us a try, we'd love to know what you think:
#    https://aka.ms/AA7q20u
#
name: dapr
extensions:
- name: dapr

  # log-level configures the log level of the dapr sidecar
  log-level: debug

  # config allows you to pass additional configuration into the dapr sidecar
  # config will be interpreted as a named k8s resource when deployed, and will be interpreted as
  # a file on disk when running locally at `./components/myconfig.yaml`
  #
  # config: myconfig

  # components-path configures the components path of the dapr sidecar
  components-path: "./components/"

  # If not using the default Dapr placement service or otherwise using a placement service on a nonstandard port,
  # you can configure the Dapr sidecar to use an explicit port.
  # placement-port: 6050
services:
- name: service-a
  project: ServiceA/ServiceA.csproj
- name: service-b
  project: ServiceB/ServiceB.csproj
- name: app
  project: App/App.csproj
- name: stateservice
  project: StateService/StateService.csproj
- name: worker-service
  project: WorkerService/WorkerService.csproj
- name: pub-service
  project: PubService/PubService.csproj
- name: sub-service
  project: SubService/SubService.csproj
- name: sub-service2
  project: SubService2/SubService2.csproj

# This may conflict with the redis instance that dapr manages.
#
# Doing a `docker ps` can show if its already running. If that's the case
# then comment out out when running locally. 
# - name: redis
#   image: redis
#   bindings: 
#   - port: 6379

```

### componentsにファイルを追加

こちらも、慣れてきたと思います。componentsフォルダに設定を追加します。
今回は、Daprを用いたPub/Subを行います。

今回はRedisかRabbitMQのどちらか好きな方を使って試してみましょう。
いずれの場合においても、同じpubsub.yamlとして定義を行います。今回はRabbitMQ環境があったので、RabbitMQで動作させています。

#### Redisを使ってPubSubを行う場合

Redisを使った場合のPub/Sub設定例です。

```yaml:pubsub.yaml
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: daprpubsub
spec:
  type: pubsub.redis
  version: v1
  metadata:
  - name: redisHost
    value: localhost:6379
  - name: redisPassword
    value: ""
scopes:
  - pub-service
  - sub-service
  - sub-service2
```

#### RabbitMQを使ってPubSubを行う場合

もう一つはRabbitMQを用いた場合の設定例です。

```yaml:pubsub.yaml
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: daprpubsub
spec:
  type: pubsub.rabbitmq
  version: v1
  metadata:
  - name: host
    value: "amqp://localhost:5672"
  - name: durable
    value: "false"
  - name: deletedWhenUnused
    value: "false"
  - name: autoAck
    value: "false"
  - name: reconnectWait
    value: "0"
  - name: concurrency
    value: parallel
scopes:
  - sub-service
  - sub-service2
```

### Subscriptionの定義

subscription.yamlとしてもう一つYAMLファイルを追加します。

```yaml:subscription.yaml
apiVersion: dapr.io/v1alpha1
kind: Subscription
metadata:
  name: apppubsub
spec:
  topic: AppStatus
  route: /appstatus
  pubsubname: daprpubsub
scopes:
  - sub-service
  - sub-service2
```

この設定ファイルでは、Subscriptionの定義を行うことができます。
pubsubname: daprpubsubは、pubsub.yamlのmetadataのnameです。そして、route /appstatusでPOSTされるパス、scopesでsub-serviceとsub-service2がサブスクライブしています。

### 起動チェック

起動確認も兼ねて、いつものtye runで起動してみましょう。

```bash
tye run
```

起動したら、Tye Dashboardを見てみましょう。
http://127.0.0.1:8000/ にアクセスすると、いつも通り起動中のサービスとDaprサイドカーの状況が表示されると思います。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/25baf13a-c3ec-d207-8b0f-2c05a82a00c0.png)

ここでは、sub-service-daprのログをみてみます。(サービスの方じゃなくて、Daprのサイドカー側です。）

ログを追いかけると、以下のような感じでdaprpubsubの起動ログが残っていると思います。ブラウザからページ内、検索でrabbitmqなどを探してみると把握しやすいと思います。

以下はログを抜き出しだ結果です。

```bash
[sub-service-dapr]: msg="found component. name: daprpubsub, type: pubsub.rabbitmq/v1" 
[sub-service-dapr]: msg="loading component. name: daprpubsub, type: pubsub.rabbitmq/v1" 
[sub-service-dapr]: msg="rabbitmq pub/sub: connectionCount: current=0 reference=0"
[sub-service-dapr]: msg="rabbitmq pub/sub: connected with connectionCount=1"
[sub-service-dapr]: msg="component loaded. name: daprpubsub, type: pubsub.rabbitmq/v1" 
[sub-service-dapr]: msg="rabbitmq pub/sub: subscribe to topic/queue 'AppStatus/sub-service-AppStatus'"
[sub-service-dapr]: msg="rabbitmq pub/sub: declaring exchange 'AppStatus' of kind 'fanout'"
[sub-service-dapr]: msg="rabbitmq pub/sub: declaring queue 'sub-service-AppStatus'"
[sub-service-dapr]: msg="rabbitmq pub/sub: binding queue 'sub-service-AppStatus' to exchange 'AppStatus'"
```

#### RabbitMQでも確認

ログを見た事には理由があります、DaprのPubSubは、Subscriptionで定義したように、定義CRDに基づいて、設定を自動的に行ってくれるのです。

RabbitMQで確認してみましょう。
http://localhost:15672/#/exchanges にアクセスします。

特に設定していないにも関わらずAppStatusという名称のExchangeがfanoutで構成されています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/f572c70c-798a-9bcb-91cc-bebef0ea9519.png)

Exchangeの詳細も確認すると、sub-service-AppStatusとsub-service2-AppStatusの二つのQueueが自動で作成され、AppStatus ExchangeにBindingされています。

(ここでの、BindingはRabbitMQのExchangeにQueueをバインドするバインディングです。DaprのBindingでは無いので、ご注意を)


今回は、RabbitMQで行っていますが、Redisでも同様で自動設定されると思われます。
このように、定義してPubSubを各サービス間で設定できるのは、非常に便利に思えます。これは、後にKubernatesにデプロイメントした際に便利な機能です。

### SubServiceとSubService2プロジェクトの編集

**双方とも同じ内容** になります。
まず、WeatherForecastController.csとWeatherForecastr.csは使用しませんので削除してください。

#### Dapr.AspNetCoreパッケージを追加

今までは、dotnet add package や Visual Studio からNuGetパッケージマネージャーで追加したりなどしてきました。今回は、パッケージコンソールから追加してみましょう。やり方はいずれの方法でもよく、つまりは以下のライブラリがプロジェクトに追加できればOKです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/de191702-0cf0-3b7d-f02b-b4d8312985f9.png)

```powershell
Install-Package Dapr.AspNetCore
Install-Package Dapr.Client
```

プロジェクトが多いので、既定のプロジェクトをSubServiceとSubService2に切り替えるのを忘れないでください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/0ef9f547-c5ba-3fbb-850a-b03d9afef16a.png)

#### Program.csの修正

Program.csからも、今までと同じように//app.UseHttpsRedirection();としてコメントしてください。
Dapr自体はPOSTできればOKなのですが、SDKで結構よしなにしてくれる仕組みがありますので、それらを有効にします。

```csharp:Program.cs
var builder = WebApplication.CreateBuilder(args);

// Add services to the container.
builder.Services.AddControllers().AddDapr(); // 拡張メソッド追加

// Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();

var app = builder.Build();
app.UseCloudEvents(); // クラウドイベントフォーマット対応

// Configure the HTTP request pipeline.
if (app.Environment.IsDevelopment())
{
    app.UseSwagger();
    app.UseSwaggerUI();
}

//app.UseHttpsRedirection(); // Dapr内通信の為、HTTPのみに変更

app.UseAuthorization();

app.MapControllers();
app.MapSubscribeHandler(); // サブスクライブハンドラー有効化

app.Run();


```

#### コントローラー追加

次に、それぞれのプロジェクトにコントローラを追加しましょう。
コントローラー側では、拡張メソッドによってアノテーションでサブスクライブ設定が可能になっています。

プロジェクトSubService側

```csharp:AppStatusController.cs
using Dapr;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;

namespace SubService.Controllers
{
    [Route("[controller]")]
    [ApiController]
    public class AppStatusController : ControllerBase
    {

        private readonly ILogger<AppStatusController> _logger;

        public AppStatusController(ILogger<AppStatusController> logger)
        {
            _logger = logger;
        }

        // サブスクライブ設定
        [Topic("apppubsub", "AppStatus")]
        [HttpPost(Name = "PostAppStatus")]
        public void PostAsync([FromBody] string message)
        {
            Console.WriteLine("Subscriber received : " + message);
        }

    }
}

```

プロジェクトSubService2側

```csharp:AppStatusController.cs
using Dapr;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;

namespace SubService2.Controllers
{
    [Route("[controller]")]
    [ApiController]
    public class AppStatusController : ControllerBase
    {

        private readonly ILogger<AppStatusController> _logger;

        public AppStatusController(ILogger<AppStatusController> logger)
        {
            _logger = logger;
        }

        // サブスクライブ設定
        [Topic("apppubsub", "AppStatus")]
        [HttpPost(Name = "PostAppStatus")]
        public void PostAsync([FromBody] string message)
        {
            Console.WriteLine("Subscriber received : " + message);
        }

    }
}
```

以下のような状態になっていると思います。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/a1c5cf24-7620-9253-3797-557ffe30e2d2.png)

### PubServiceプロジェクト側の編集

今回は、良い機会なのでConsoleプロジェクトとしてホストして、同じようにDaprをサイドカーとしてローディングしています。今までと特に違いはありませんが、このようにコンソールプログラムを常駐させる事もできますし、任意のコンテナを常駐させる事もできます。

#### PubServiceにクライアントSDK追加

こちらでも、PubServiceプロジェクトにクライアントライブラリを追加しておきましょう。

```bash
$ dotnet add package Dapr.Client
$ dotnet add package Dapr.AspNetCore
```

#### Program.csの編集

２秒毎に、ランダムで生成した数値をPublishEventAsyncで送るシンプルなものです。

```csharp:Program.cs
using Dapr.Client;

namespace PubService
{
    class Program
    {
        static async Task Main(string[] args)
        {
            string PUBSUB_NAME = "daprpubsub";
            string TOPIC_NAME = "AppStatus";

            while (true)
            {
                System.Threading.Thread.Sleep(2000);
                Random random = new Random();
                int stateId = random.Next(1, 1000);
                CancellationTokenSource source = new CancellationTokenSource();
                CancellationToken cancellationToken = source.Token;
                using var client = new DaprClientBuilder().Build();

                await client.PublishEventAsync(PUBSUB_NAME, TOPIC_NAME, stateId.ToString(), cancellationToken);
                Console.WriteLine("Published data: " + stateId);
            }
        }
    }
}

```

### 検証

今回はtye runで起動しますが、編集箇所も多かったですし、ちょっとオプションを追加します。 --watchオプションを追加する事で、tyeを起動した状態でプロジェクトを編集すると、自動でリビルドした上でプロセスをリロードしてくれます。コーディングやデバッグしながらの作業で、とても便利です。

```bash
tye run --watch
.....
起動のログ
.....
.....
[15:47:37 INF] watch: pub-service_e203bdc7-c Exited
[15:47:37 INF] watch: File changed: C:\Users\kahiro\DaprQiita\PubService\Program.cs
[15:47:37 INF] Build Watcher: Builds requested; waiting 250ms for more...
[15:47:37 INF] Build Watcher: Getting requests...
[15:47:37 INF] Build Watcher: Processing 1 requests...
[15:47:37 INF] Build Watcher: Building project C:\Users\kahiro\DaprQiita\PubService\PubService.csproj...
[15:47:37 INF] Build Watcher: Waiting for builds to complete...
[15:47:39 INF] Built C:\Users\kahiro\DaprQiita\PubService\PubService.csproj with exit code 0
[15:47:39 INF] Build Watcher: Done with requests; waiting for more...
[15:47:40 INF] pub-service_e203bdc7-c running on process id 20756 bound to http://localhost:59051, https://localhost:59052
[15:47:40 INF] Replica pub-service_e203bdc7-c is moving to a ready state
[15:47:40 INF] watch: pub-service_e203bdc7-c Started
[15:47:41 INF] Selected process 20756.
[15:47:41 INF] Listening for event pipe events for pub-service_e203bdc7-c on process id 20756
```

#### PubService側ログ

PubService側のログを見てみましょう。慣れてきたと思いますが、以下のViewから各プロジェクトのログを参照できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/c8506780-a82f-9882-14a2-394a07578162.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/414f0311-8aa4-c0eb-ff6d-a693d35b05a2.png)

#### PubServiceのDapr側ログ

Dapr側サイドカーでも、"rabbitmq pub/sub: publishing message to AppStatus" app_id=pub-service という感じでログが残っています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/fc99e715-a233-da59-e232-4b4ba3e22af6.png)

#### RabbiqMQ側管理コンソール

RabbiqMQ側の管理コンソールからExchangeの状態を確認します。以下のようにメッセージがExchangeに入り、バインドされたQueueへFanoutで渡されています。

http://localhost:15672/#/exchanges

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/c453c137-5971-ba63-8a59-9edbd405c6a9.png)

Queueにそれぞれ入ったメッセージもConsumeされている事がわかります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/913e18e9-2839-4bb7-7562-d96d73370af9.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/ff422ddc-fe5d-bfa1-7c1d-b81d05f60e87.png)

#### Zipkin側 Dapr分散トレースの様子

http://localhost:9411/zipkin/

ZipkinでもDaprによるメッセージ送信先Topicなど確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/4778516a-94a5-e8e5-8874-631dda40a880.png)

#### SubServiceとSubService2側 Daprのログ

以下のように、DaprがAppStatus Topicから、メッセージをACKしている事がわかります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/e4f73d12-6d60-3988-fe50-92e839fd17c5.png)

#### SubServiceとSubService2のログ
SubService側では、Daprから渡された結果を表示していますが、双方とも同じデータを受け取っている事がわかります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/3cfb51f8-5d47-0189-6fc8-771fb8a8adf0.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24782/f9d12f6a-862d-72bd-75ef-5fa821e1c050.png)