機能構成・概要

Functions①：httpトリガー。webページを表示した際に実行される。受け取ったメッセージをQueue Storageに格納する。
Queue Storage：Functions①からのメッセージを一時的に格納。
Functions②：Azure Queue Storageトリガー。Queueにメッセージが格納されたときに実行される。Queueから取り出したメッセージを加工し、Blob Storage内のファイルに追記する。
Blob Storage：ファイルの配置場所。ファイルは現状1つで、Functions②により追記される。

処理概要

Webページ表示
Functions①実行(Functions①のURL実行)
メッセージをQueue Storageに格納
3をトリガーとして、Functions②を実行
メッセージを加工して、Blob Storage内のファイルに追記

やりたいこと

ほぼ同時にQueue Storageに入ったデータを過不足なく、Blob Storage内のファイルに追記したい。

(Functions②が実行中に別のFunctions②を実行しないか、
Queue Storage内のメッセージをすべて、1つのFunctions②でBlob Storage内のファイルに格納するか。)

課題・問題点

現状、Queue Storageに同時に10個メッセージが入った場合、Functions②が10個同時に実行され、処理がバッティングしてしまう。
したがって、Blob Storageのファイルには、10個分のメッセージは追記されず、1個分とか2個分のみ追記されてしまう。

エラー内容

The specified block list is invalid.

コード

C#
1/*Functions②*/
2#r "Newtonsoft.Json"
3
4using System;
5using Newtonsoft.Json;
6using System.Threading;
7
8public static void Run(
9    string queue01,
10    string Blob01In,
11    out string Blob01Out,
12    ILogger log)
13{
14    //JSONデータからC#オブジェクト作成
15    dynamic jsonData = JsonConvert.DeserializeObject(queue01);
16    //C#オブジェクト⇒CSVデータ
17    string csvData = jsonData.clm1 + "," + jsonData.clm2;
18    
19    //Blobの内容にQueueの内容を追記
20    Blob01Out = Blob01In + csvData + System.Environment.NewLine;
21    Thread.Sleep(5000);
22    log.LogInformation("処理終了");
23}

(Sleepは、処理がバッティングしていることをわかりやすくしているために使用しています。ログ出力も同様)

json
1/*Function.json*/
2{
3  "bindings": [
4    {
5      "name": "queue01",
6      "type": "queueTrigger",
7      "direction": "in",
8      "queueName": "queue01",
9      "connection": "AzureWebJobsStorage"
10    },
11    {
12      "type": "blob",
13      "name": "Blob01Out",
14      "path": "test01/test.csv",
15      "connection": "AzureWebJobsStorage",
16      "direction": "out"
17    },
18    {
19      "type": "blob",
20      "name": "Blob01In",
21      "path": "test01/test.csv",
22      "connection": "AzureWebJobsStorage",
23      "direction": "in"
24    }
25  ]
26}

Azure機能にあまり詳しくないため、不備等ございましたら、
補足致しますので、お力添えをお願いいたします。

補足(20181227)

それっぽい情報は見つかりましたが、記事通り、host.jsonに追記しても、何も変化はありません・・・
https://4423lab.wpblog.jp/archives/548

NakamuraYoichi

2018/12/27 23:57

すみません、念のために確認です。アトリビュートやhost.jsonを変更した後、変更を反映するためにFunction Appの「再開」はしていますでしょうか。

HelloWorld2

2019/01/10 09:09 編集

レスポンスが大変遅くなり、申し訳ございません。「再開」はしてみましたが、状況は変わりませんでした。 ■試したこと ①メソッドの1行上に[Singleton]を記載。 ②同様に、[Singleton(Mode = SingletonMode.Listener)]を記載。 ③host.jsonに、「 "queues": {"batchSize": 1,"newBatchThreshold": 0}」を記載。 ④①と③の組み合わせ ⑤②と③の組み合わせ ※いずれも設定後「再開」をしています。 host.jsonの設定は、「Function App の設定」からで合っていますでしょうか？(ストレージ内にもhost.jsonがありますが・・・) 上記のhost.jsonには、{"version":"2.0"}の記述のみ存在していました。

HelloWorld2

2019/01/10 11:51 編集

public static async Task Run()じゃなければならない、みたいな制約とかってないですか(サンプルがAsyncなため不安になりました)

行動規範の内容に同意します

回答3件

自己解決

インスタンス自動増加のため

投稿2019/04/02 05:30

HelloWorld2

総合スコア32

C# なら Singleton が効きますが、微妙な挙動です。

Singleton を指定しても、メソッドの先頭で待機するようになるだけで、同時に呼ばれる点は変わらないようです。すると、呼ばれた時の Blob を元に処理を進めてしまうので、同時に開始された他のメソッドによる更新は反映されない。

Blob の内容を string で受け取るのではなく、一旦 ICloudBlob で受けておいて、メソッドの中で読み込む、という作戦を考えました。

C#
1using Microsoft.Azure.WebJobs;
2using Microsoft.Azure.WebJobs.Host;
3using Microsoft.WindowsAzure.Storage.Blob;
4using System.IO;
5using System.Text;
6
7namespace FunctionApp1
8{
9    public static class Function1
10    {
11        [FunctionName("Function1"), Singleton]
12        public static void Run([QueueTrigger("queue01", Connection = "AzureWebJobsStorage")]string queue01,
13            [Blob("test01/test.csv", Connection = "AzureWebJobsStorage")] ICloudBlob Blob01In,
14            [Blob("test01/test.csv", Connection = "AzureWebJobsStorage")] out string Blob01Out,
15            TraceWriter log)
16        {
17            log.Info($"C# Queue trigger function processed: {queue01}");
18
19            var builder = new StringBuilder();
20            using (var stream = Blob01In.OpenRead())
21            {
22                using (var reader = new StreamReader(stream))
23                {
24                    while (!reader.EndOfStream)
25                    {
26                        builder.AppendLine(reader.ReadLine());
27                    }
28                }
29            }
30            builder.AppendLine(queue01);
31            Blob01Out = builder.ToString();
32
33            System.Threading.Thread.Sleep(5_000);
34        }
35    }
36}