メッセージ キューとメッセージ タイプの作成と処理
メッセージ プロセッサは、イベントを表すメッセージを処理するフレームワークです。 次のプロパティがあります:
- メッセージは正しい順序で処理されます。 (依存メッセージは、次に示す順序で処理されます。)
- スケーラブルです。 (独立したメッセージは並列で処理できます。)
- 必要なシステム リソースを使用します。
- メッセージ内に不備が発生した場合、システム リソースが消耗しないようにします。
- 信頼性が高い。
- 追跡可能です。
たとえば、このフレームワークを使用して、外部システムとのカスタム統合を作成および管理し、他のカスタム機能を処理することができます。 Microsoft Dynamics 365 Supply Chain Management には、たとえば、定義済のメッセージ タイプおよびメッセージ キューを使用するいくつかの既定の機能が含まれています。 このような機能には、サード パーティ製造実施システム (MES) 統合、延期転記、コンテナ梱包明細の転記 があります。 倉庫管理のみモード 、メッセージ プロセッサ フレームワークを使用して 出荷注文を管理できます。
この記事では、Visual Studio を使用して独自のカスタム メッセージ キューとメッセージ タイプを設計する方法と、メッセージ プロセッサのメッセージ ページを使ってすべてのメッセージ タイプ (定義済みのメッセージ・タイプを含む) の処理を監視し制御する方法について説明します。
メッセージ プロセッサのメッセージ ページ
メッセージ プロセッサ メッセージ ページを使用して、着信メッセージの一覧の表示、メッセージ ログの表示、メッセージの手動処理、および問題のトラブルシューティングを行います。
メッセージ プロセッサのメッセージ ページを開きます
メッセージ プロセッサが処理したメッセージのリストを表示するには、システム管理 > メッセージ プロセッサ > メッセージ プロセッサ メッセージ にアクセスします。
「メッセージ プロセッサ メッセージ」 ページのグリッド列およびフィルタ
メッセージ プロセッサ メッセージ ページの上部にあるフィールドを使用して、検索する特定のメッセージを見つけることができます。 これらのフィルターのほとんどは、メッセージ グリッド列のヘッダーと一致します。 次のフィルターと列のヘッダーが使用可能です。
[メッセージ ] : メッセージのタイプ。
メッセージ キュー : メッセージが処理されるキューの名前。 次のキューが用意されています。
- 製造上の実行3 - このキューには、"製造実行システム統合" または "製造 実行システム統合" の一部として作成されたメッセージ 含まれます。 これらのメッセージは、 製造実行システム統合 ページにも一覧表示されます (メッセージ プロセッサ メッセージ ページなど)。 詳細情報については、サード パーティの製造実行システムと統合する を参照してください。
- 生産 : このキューには、仕訳帳に転記する前に、 Makeの完成品を物理的に利用できる一部として作成されたメッセージ 含まれます。 これらのメッセージは、生産指示の延期 ページにも一覧表示されます (メッセージ プロセッサ メッセージ ページなど)。 詳細情報については、 完成品を仕訳帳に転記する前に現物在庫にする を参照してください。
- [ 倉庫] : このキューには、前回の出荷コンテナが梱包プロセスの一環として閉じ込まれると販売梱包明細を転記するなど、倉庫管理用に作成されたメッセージが保持されます の梱包プロセス。 (このメッセージには、コンテナの梱包明細を実行する のメッセージタイプが含まれます。)
- [出荷注文 ] : このキューには、 管理モードのみをサポートするメッセージが保持。
- ソース システム製品 : このキューには、製品マスタ データのソースを するメッセージが。
- [外部倉庫出荷注文の更新 ] : このキューには、他 共有倉庫処理をサポートするメッセージ。
- Dynamics 365 Salesの統合 : このキューには、Dynamics 365 Salesと統合するメッセージが保持Dynamics 365 Sales。 この機能と、このキューに追加されるメッセージの詳細については、Dynamics 365 Sales を使用して、見積から現金に対して追加された効率性を活用する を参照してください。
- <カスタム キュー> : 他のタイプのキューをサポートするためにシステムがカスタマイズされている場合、キューもここに一覧表示されます。 カスタム キューを追加する方法の詳細については、新しいキューの実装 を参照してください。
[メッセージ ] : メッセージの状態。 次の状態があります:
- [キュー ] : メッセージをメッセージ プロセッサで処理できます。
- [処理 ] : メッセージがメッセージ プロセッサによって正常に処理されました。
- [キャンセル ] : ユーザーによってメッセージがキャンセルされました。
- [失敗 ] : メッセージを処理に失敗しました。
メッセージの : このフィルタでは、メッセージのコンテンツを全文検索します。 (グリッドにメッセージ コンテンツは表示されません。) このフィルターでは、ほとんどの特殊記号 (「-」など) を空白として処理し、すべての空白文字をブール OR 演算子として処理します。 たとえば、USMF-123456 に相当する特定の
journalid値を検索すると、システムは 「USMF」 または 「123456」 のいずれかを含むすべてのメッセージを検索することになります。この場合、結果のリストは長くなる可能性があります。 したがって、この場合は 123456 とだけ入力した方が、より具体的な結果が返されます。
メッセージ ログ、メッセージの内容、および詳細の表示
グリッドで選択し、各処理イベントが表示されているメッセージ グリッドの下にある ログ または コンテンツの管理 タブを選択することで、メッセージに関する詳細情報を検索することができます。
メッセージのコンテンツ タブ のテキストは、メッセージ タイプ の値 によって異なります。 したがって、テキストの長さは異なります。 典型的なメッセージ内容のテキストは、括弧 ({) で始まり、閉じ括弧 (}) で終わります。 入力の間には、フィールド名 (例: journalId) が続き、その間にコロンと値が続きます (例: US COLON-123456)。
ログ タブのツールバーには次のボタンが含まれます。
- [ログ ] 選択をクリックすると、処理結果が表示されます。 この機能は、メッセージの 処理結果 の値が 失敗 で、処理が失敗した理由を調べる場合に特に役立ちます。
- バンドル : 複数のメッセージ処理工程を同じバッチ プロセスの一部として実行できます。 このボタンを選択して、この詳細データを表示します。 たとえば、システムが指定された順序で特定のメッセージを処理することが必要もなる依存関係が存在するかどうかを確認できます。
メッセージを手動で処理、キャンセル、再送信する
現在のメッセージの状態に応じて、手動で処理したりキャンセルしたりすることができます。 グリッドでメッセージを選択し、アクション ウィンドウで 処理 または キャンセル を選択します。
以前にキャンセルされたメッセージ を再キューする場合は、グリッドでメッセージを選択し、アクション ウィンドウで キュー を選択します。 メッセージはシステムによって通常どおり処理されます。
メッセージ プロセッサーのバッチジョブを使用して、メッセージ処理をスケジュールする
メッセージ キューを処理するには、メッセージ キューを実行するバッチ ジョブを設定する必要があります。 通常は、各キューを処理する固定の定期的なスケジュールを設定します。 ただし、どのキューも要求時に実行できます。 必要なバッチジョブを作成してスケジュールするには、以下の手順に従います。
- システム管理 > メッセージ プロセッサ > メッセージ プロセッサに移動します。
- メッセージ プロセッサ ダイアログ ボックスの メッセージ キュー フィールドで、処理するメッセージに関連付けられているメッセージ キューを選択します。 選択するキューは、メッセージを生成した機能またはシステムによって異なる場合があります。
- バックグラウンドで実行 クイックタブで、Supply Chain Management の 他のタイプのジョブ と同様に、必要に応じてバッチとスケジューリング オプションを設定します。
- OK を選択して設定を適用し、設定に基づいてジョブをスケジュールします。
メッセージ プロセッサ キューの設定
各メッセージ プロセッサ キュー専用のプロセッサ タスクの数を構成できます。 構成されていないキューには、必要に合って上書きできる既定値が使用されます。 次の手順に従って、1 つ以上のキューをカスタマイズします。
システム管理 > メッセージ プロセッサ > メッセージ キューの設定 に移動します。
次の手順のいずれかを実行します。
- 既存のキューを編集するには、アクションペインで 編集 を選択し、グリッドで対象のキューを選択します。
- 新しい構成を追加するには、アクションペインで 追加 を選択し、グリッドに新しい行を追加します。 次に、新しい行の メッセージ キュー フィールドで、構成するキューの名前を選択します。
新しい行または選択された行に対して、プロセッサタスク数 フィールドを、指定されたキューに専用されるべきプロセッサタスク数に設定します。 最大値は 8 です。 最小値は、システムに対して構成されるバッチ スレッドの最小数によって異なります (通常は 2)。
アクション ウィンドウで、保存を選択します。
失敗した処理結果に対する警告を配信するビジネス イベントの設定
この記事で前述したように、メッセージの状態 フィールドの 失敗 (場合によっては キャンセル済み) 値でフィルタリングすることで、失敗したメッセージを照会することができます。 さらに、処理の失敗を警告する ビジネス イベント を設定することができます。 この設定を完了するには、ビジネス イベント カタログ ページで メッセージ プロセッサ メッセージ処理 ビジネス イベントを有効化します (システム管理 > 設定 > ビジネス イベント > ビジネス イベント カタログ)。 アクティブ化プロセスの一環として、イベントが特定の法人に固有であるか、すべての法人に適用するかを指定するように求められます。 また、エンドポイント名を指定して事前に定義する必要があります。
ノート
ビジネス イベントの発生時 が (HTTPS ではなく) Microsoft Power Automate に設定されている場合、エンドポイント名は、Microsoft Power Automate の設定に基づいて Supply Chain Management で自動的に作成されます。
Power Automate の例
この例では、ビジネス イベントの発生時を Microsoft Power Automateに設定して、アクション センター (旧 Infolog) メッセージと、特定の失敗メッセージの メッセージ プロセッサ メッセージ ページを開くハイパーリンクを含む電子メール通知を送信しています。 必要に応じて、異なるチャネルを使用して並列で通知を送信し、イベント データに基づく受信者を制御する追加ロジックを追加することができます。
Power Automate では、次の図に示されているように、JSON 解析 および 電子メール送信 のステップに続くフロー トリガー ビジネス イベントの発生時 財務と運用アプリ (Dynamics 365) に対する新しい自動化クラウド フローを作成します。
ビジネス イベントの発生時 ステップでは、インスタンス フィールドと カテゴリ フィールドで値を検索または入力し、ビジネス イベント フィールドを 処理されたメッセージ プロセッサのメッセージ に設定します。
JSON の解析 のステップで、スキーマ フィールドで、拡張フィールドを定義するスキーマを入力します。 Supply Chain Management の ビジネス イベント カタログ ページにある スキーマのダウンロード オプションを使用するか、または例にあるスキーマ テキストを貼り付けて開始することができます。
{ "properties": { "BusinessEventId": { "type": "string" }, "ControlNumber": { "type": "integer" }, "EventId": { "type": "string" }, "EventTime": { "type": "string" }, "MajorVersion": { "type": "integer" }, "MessageContent": { "type": "string" }, "MessageDestinationCompanyId": { "type": "string" }, "MessageDestinationOperationalSiteId": { "type": "string" }, "MessageDestinationWarehouseId": { "type": "string" }, "MessageDestinationWorkloadName": { "type": "string" }, "MessageInfolog": { "type": "string" }, "MessageProcessingResult": { "type": "string" }, "MessageProcessingResultLabel": { "type": "string" }, "MessageProcessorMessagePageUrl": { "type": "string" }, "MessageQueue": { "type": "string" }, "MessageQueueLabel": { "type": "string" }, "MessageSourceCompanyId": { "type": "string" }, "MessageSourceOperationalSiteId": { "type": "string" }, "MessageSourceWarehouseId": { "type": "string" }, "MessageSourceWorkloadName": { "type": "string" }, "MessageState": { "type": "string" }, "MessageStateLabel": { "type": "string" }, "MessageType": { "type": "string" }, "MessageTypeLabel": { "type": "string" }, "MinorVersion": { "type": "integer" } }, "type": "object" }
電子メールの送信 ステップにおいて、個別のフィールドを選択するか、または電子メール本文の例を 本文 フィールドに貼り付けて開始することができます。
Message queue: @{body('Parse_JSON')?['MessageQueue']} Message queue label: @{body('Parse_JSON')?['MessageQueueLabel']} Message type: @{body('Parse_JSON')?['MessageQueueType']} Message type label: @{body('Parse_JSON')?['MessageQueueTypeLabel']} Message content: @{body('Parse_JSON')?['MessageContent']} Message state: @{body('Parse_JSON')?['MessageState']} Message state label: @{body('Parse_JSON')?['MessageStateLabel']} Message processing result: @{body('Parse_JSON')?['MessageProcessingResult']} Message processing result label: @{body('Parse_JSON')?['MessageProcessingResultLabel']} Message infolog: @{body('Parse_JSON')?['MessageInfolog']} Message source company ID: @{body('Parse_JSON')?['MessageSourceCompanyId']} Message source workload name: @{body('Parse_JSON')?['MessageSourceWorkloadName']} Message source site ID: @{body('Parse_JSON')?['MessageSourceOperationalSiteId']} Message source warehouse ID: @{body('Parse_JSON')?['MessageSourceWarehouseId']} Message destination company ID: @{body('Parse_JSON')?['MessageDestinationCompanyId']} Message destination workload name: @{body('Parse_JSON')?['MessageDestinationWorkloadName']} Message destination site ID: @{body('Parse_JSON')?['MessageDestinationOperationalSiteId']} Message destination warehouse ID: @{body('Parse_JSON')?['MessageDestinationWarehouseId']} Message processor message page URL: @{body('Parse_JSON')?['MessageProcessorMessagePageUrl']}
ビジネス イベントを保存すると自動的に有効化され、Supply Chain Management の一部として使用可能になります。
スケジューラ
メッセージ プロセッサには、1 つのスケジューラがあります。 SysMessageKeyDateTimeSequenceProcessorScheduler クラスは、キー、日付、および時刻に基づく依存関係を持つメッセージをスケジュールして、メッセージが正しい順序で処理されます。 処理する必要があるメッセージは、SysMessageProcessorTaskBundle テーブルと SysMessageProcessorTaskBundleMessage テーブルに格納されます。 依存メッセージは同じバンドルに含む必要があります。
SysMessageKeyDateTimeSequenceProcessorScheduler クラスは、サードパーティの MES 統合 機能によって使用され、生産オーダーに関連するメッセージを保護し、受信または作成された順番に処理されます。 この依存関係は、キー (製造オーダー番号) と時刻で定義されます。
たとえば、システムは次のメッセージを受信または作成します:
- 製造オーダー 1 - メッセージの開始
- 製造オーダー 2 - メッセージの開始
- 製造オーダー 2 - 終了メッセージとして報告
- 製造オーダー 1 - 終了メッセージとして報告
この場合、SysMessageKeyDateTimeSequenceProcessorScheduler クラスは以下の順序でメッセージの処理をスケジューリングします:
- 製造オーダー 1 - メッセージの開始
- 製造オーダー 1 - 終了メッセージとして報告
- 製造オーダー 2 - メッセージの開始
- 製造オーダー 2 - 終了メッセージとして報告
各完了レポート メッセージは、関連する開始メッセージが正常に処理されたかどうかに依存します。
ボトルネックが発生する場合は、複数のタスクの処理用にレコードを取得するときにボトルネックが発生する可能性があります。また、メッセージ間に依存関係がある可能性があるから、システムはメッセージをバンドルします。 たとえば、バンドル サイズが 2 の場合、次のバンドルが作成されます:
- バンドル 1 - 製造オーダー 1 - メッセージの開始
- バンドル 1 - 製造オーダー 1 - メッセージの開始
- バンドル 2 - 製造オーダー 2 - メッセージの開始
- バンドル 2 - 製造オーダー 2 - メッセージの開始
スケーラビリティについては、新しいキューを作成するときに、バンドルを処理する必要があるタスクの数を構成できます。 (詳細については、新しいキューを実装する を参照してください。 )。構成が 2 つのメッセージ プロセッサ タスクを使う場合、2 つのバンドルは並列に処理できます。
実装の例
このセクションでは、メッセージ プロセッサで使用できる新しいメッセージ キューおよびメッセージ タイプの開発方法を示します。
新しいキューの実装
この例では、新規キューを追加する方法も示します。 キューが作成された後、キューにメッセージを送信できます。 これらのメッセージは、前述の設定に従ってメッセージ プロセッサによって処理されます。
Visual Studio で、
SysMessageQueue列挙型の拡張子を作成します。新しいキューの新しい列挙値を追加します。
列挙値の機能クラスを設定します。
次の例に示すように、
SysMessageQueueMetadataを継承したクラスを作成します。[SysMessageQueueFactory(SysMessageQueue::MyQueue)] public final class MyQueue extends SysMessageQueueMetadata { /// <summary> /// Gets the default number of message processor tasks. /// </summary> /// <returns>The default number of message processor tasks.</returns> public SysMessageProcessorsTasks getDefaultMessageProcessorsTasks() { return 1; } /// <summary> /// Get the maximum number of messages in a bundle. /// </summary> /// <returns>The maximum number of messages in a bundle.</returns> public SysMessageProcessorMaxBundleSize getMaxBundleSize() { return 10; } /// <summary> /// Determines if messages should be queued when sending or processing messages. /// </summary> /// <returns>true if messages should be queued when sending a message; otherwise false.</returns> public boolean queueMessageOnSend() { return true; } /// <summary> /// Determines if dependent messages should be committed per created transaction ID. /// </summary> /// <returns>true if dependent messages should be committed per created transaction ID.; otherwise false.</returns> public boolean commitPerCreatedTransactionID() { return false; } }ユーザー インターフェイス (UI) でメッセージ プロセッサを有効にするために、
SysMessageProcessorToggleクラスの拡張機能を作成します。[ExtensionOf(classStr(SysMessageProcessorToggle))] final class MySysMessageProcessorToggle_Extension { /// <summary> /// Determines if the toggle is enabled. /// </summary> /// <returns> /// true if the toggle is enabled; otherwise, false. /// </returns> internal boolean isEnabled() { return next isEnabled() || MyFeature::instance().isEnabled(); } }
新しいメッセージ タイプの実装
この例では、メッセージ プロセッサで使用するプロセスのタイプごとにメッセージ タイプを作成する方法を示します。 メッセージ タイプによって、そのタイプのメッセージによって実行できるプロセスが指定されます。 キュー内の各メッセージの目的を識別するために、メッセージのタイプ名が メッセージ プロセッサ メッセージ ページに表示されます。
以下の手順に従って、新しいメッセージ タイプを作成してください。
Visual Studio で、
SysMessageType列挙型の拡張子を作成します。新しいメッセージ タイプの新しい列挙値を追加します。
列挙値の機能クラスを設定します。
次の例に示すように、
SysMessageTypeMetadataを継承したクラスを作成します。[SysMessageTypeFactory(SysMessageType::MyMessage)] public final class MyMessage extends SysMessageTypeMetadata { /// <summary> /// Validates if it is allowed to send a message on a given queue. /// </summary> /// <param name = "_messageQueue">A queue to send message to.</param> /// <param name = "_messageTarget">A message target.</param> /// <param name = "_sourceMessageTarget">A message source.</param> /// <returns>true if it is allowed to send a message; false, otherwise.</returns> public boolean canSendMessage(SysMessageQueue _messageQueue, SysMessageTargetRecId _messageTarget, SysMessageTargetRecId _sourceMessageTarget) { if (_messageQueue == SysMessageQueue::MyQueue) { return true; } return false; } /// <summary> /// Returns a list of allowed message state transitions. /// </summary> /// <returns>A list of allowed message state transitions.</returns> public Enumerator getAllowedMessageStateTransitions() { // Allow the following state transitions: // SysMessageState::None -> SysMessageState::Queued // SysMessageState::None -> SysMessageState::Processed // SysMessageState::Queued -> SysMessageState::Processed // SysMessageState::Failed -> SysMessageState::Queued // SysMessageState::Failed -> SysMessageState::Cancelled // SysMessageState::Failed -> SysMessageState::Processed return SysMessageStateTransition::getProcessedCancelledStateTransitions(); } /// <summary> /// Returns dependencies related to a message. /// </summary> /// <param name = "_message">The message to add dependencies to.</param> /// <returns>A list of dependencies related to a message.</returns> public List getDependencies(SysMessage _message) { List dependencies = new List(Types::Record); // Add a dependency to the message if you need to process messages in order. // For example, process a start message before an end message. // The message will not be processed before the dependent message is processed or cancelled. // The default scheduler, SysMessageKeyDateTimeSequenceProcessorScheduler only supports one dependency per message. // Messages with different keys may be processed in parallel. SysMessageDependencyKey dependencyKey = strFmt('Mykey:%1', this.getContract(_message).parmMyKey()); dependencies.addEnd(SysMessageKeyDateTimeSequenceDependency::create(dependencyKey)); return dependencies; } /// <summary> /// Execute the logic associated with the given message state transition. /// </summary> /// <param name = "_message">The message.</param> /// <param name = "_messageStateTransition">The state transition to execute.</param> public void processMessageStateTransition(SysMessage _message, SysMessageStateTransition _messageStateTransition) { if (_messageStateTransition.isEqual(SysMessageStateTransition::newFromStates(SysMessageState::Queued, SysMessageState::Processed))) { this.processMessage(_message); } else if (_messageStateTransition.isEqual(SysMessageStateTransition::newFromStates(SysMessageState::Failed, SysMessageState::Processed))) { this.cancelMessage(_message); } } private void processMessage(SysMessage _message) { // Add logic to process the message. } private void cancelMessage(SysMessage _message) { // Add logic to cancel the message if needed. } private ClassId contractClass() { return classNum(MyContract); } protected MyContract getContract(SysMessage _message, ClassId _objectTypeId = this.contractClass()) { var contract = FormJsonSerializer::deserializeObject(_objectTypeId, _message.Content) as MyContract; return contract; } }