29

SQSがそのために構築されていないことは知っていますが、いくつかの基準を満たすキュー内のメッセージを見つけることは可能ですか?

ループ内でメッセージをプルし、メッセージ本文で特定のパターンを検索し (デシリアライズすることさえせずに)、必要なメッセージをフィルタリングすることができます。しかし、その後、無限ループに陥る可能性があります。最初に読み取ったメッセージは、キューの最後に到達するまでにキューに戻されます...

メッセージの可視性を拡張することは可能ですが、キュー全体をスキャンするのにかかる時間を正確に知るにはどうすればよいですか? また、可視性をどのくらいの期間拡張する必要がありますか? そこに文字どおり 1 万通のメッセージがあるとしたら?

ここに回避策はありますか？いくつかのメッセージのキューをスキャンし、それらを削除する必要があります...

4

11 に答える 11

27

短い答え：いいえ。

キューは、タスクなどのために設計されています。マシンは、キューから新しいタスク (つまり、メッセージ) を取得し、タスクを実行してから、タスクを削除します。

メッセージを検索してフィルタリングしようとしている場合、その仕事に間違ったツールを使用しているのではないかと思わずにはいられません…</p>

于 2012-10-05T08:33:20.370 に答える
6

短い答えでも長い答えでも「いいえ」ではないと思います。

「はい」である2つの反対の解決策があります。

  1. キューをトラバースし、訪問リストを維持する
  2. エンタープライズ統合パターン (メッセージ ルーティング) を使用して、基準に基づいてメッセージをダウンストリームに分割する

1. キューをトラバースし、訪問済みリストを維持する

Nメッセージが追加または削除されていない、メッセージのあるキューの場合を考えてみましょう。追加情報がない場合 (たとえば、条件に一致するメッセージの数がわかっている場合)、すべてのメッセージをトラバースする必要がありNます。

Nここでの重要なポイントは、すべてのメッセージをいつトラバースしたかを知ることです。ここにはいくつかの問題があります。

  1. 正確に知るには、キューに追加されたメッセージを追跡する必要があります
  2. おおよそを知るApproximateNumberOfMessagesには、キューの属性を取得できます
  3. または、ループ内でメッセージを受信し、訪問済みリストを維持して、最終的にはキューが分割されている各サーバーからメッセージをサンプリングして排出すると想定することもできます。

message_id訪問済みリストを維持するには、メッセージを受信して​​一致基準を評価するときに、訪問済みのすべてのメッセージを保存できます。

メッセージ ID はほぼ一意です。このスレッドを参照してください

https://forums.aws.amazon.com/message.jspa?messageID=76119

(3) を使用した場合、キューを使い果たすのに何回反復する必要があるかはわかりません。ただし、これを無期限に実行した場合、SQS シャード サーバーに対する加重ランダム分散によってすべての確率がゼロではない限り、キューを使い果たすことが保証されます。

2. エンタープライズ統合パターン (メッセージ ルーティング) を使用して、基準に基づいてメッセージをダウンストリームに分割する

メッセージング アーキテクチャを制御できる場合は、基準に基づいてさまざまな受信者にメッセージをディスパッチする「フロントエンド」メッセージ プロセッサとして Message Router を使用できます。

具体的には、Content-Based Router を使用します。

http://www.enterpriseintegrationpatterns.com/patterns/messaging/ContentBasedRouter.html

于 2017-04-04T06:39:29.623 に答える
3

同様の要件があり、この「ハンズオン」チュートリアルで説明されているアーキテクチャに行き着きました: Filter Messages Published to Topics .

基本的に、イベント/メッセージを SQS キューに発行する代わりに、それらを SNS トピックに発行すると、各コンシューマーはトピックにサブスクライブされる独自の SQS キューを持ちます。その後、SNS サブスクリプション フィルターを使用して、関連するメッセージのみが各コンシューマーのキューに入れられるようにすることができます。

これにより、追加のインフラストラクチャ オーバーヘッドが発生しますが、ソリューションとしてはうまく機能しました。

于 2020-04-06T17:00:11.313 に答える
0

古いトピックですが、役立つかもしれませんメッセージスレッドの小さなリストにGroupIdでFIFOを使用できます。

于 2020-06-04T07:33:25.330 に答える
-1

いくつかの例でこれを理解してみましょう。10 個のメッセージを作成して送信します

// Send a message
for (int i = 0; i < 10; i++) {
    System.out.println("Sending a message to MyQueue.\n");
    Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
    // extra code

    String sdate;
    Format formatter;
    Date date = new Date();

    // 2012-12-01
    formatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ");
    sdate = formatter.format(date);
    System.out.println(sdate);

    messageAttributes.put("Datestamp"+i, new MessageAttributeValue().withDataType("String").withStringValue(sdate));

    Map<String, MessageAttributeValue> messageAttributes1 = new HashMap<>();
    messageAttributes1.put("attributeName", new MessageAttributeValue().withDataType("String").withStringValue(sdate));
    SendMessageRequest request = new SendMessageRequest();
    request.withMessageBody("A test message body."+sdate);
    request.withQueueUrl(myQueueUrl);
    request.withMessageAttributes(messageAttributes);
    sqs.sendMessage(request);
}

datetimestamp1 から datetimestamp10 までの 10 個のメッセージがあっても
、属性によるフィルタリングは機能しません

いくつかの myTag 属性でフィルターを試してみましょう

ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest(myQueueUrl);

//ReceiveMessageRequest receiveRequest = new ReceiveMessageRequest(queueUrl);

receiveMessageRequest.withMaxNumberOfMessages(10);
receiveMessageRequest.withMessageAttributeNames("myTag");
List<Message> messages = sqs.receiveMessage(receiveMessageRequest).getMessages();

10 個のメッセージが返され、myTag 値は null です

message.getMessageAttributes().get("日付スタンプ") は null です message.getMessageAttributes().get("myTag") は null です

したがって、そのキーが見つからないかのようにメッセージ属性でフィルタリングすることはできません。メッセージ属性がないか、すべてのメッセージ属性が同じです。

長い答えはNOOOOOです

于 2016-05-24T03:45:29.083 に答える
-6

これは実際にはすべて真実ではありませんが、

実際、メッセージ属性のトリックを使用して、キュー内のメッセージを「ちょっと」フィルタリングできます。

各メッセージには、メッセージの作成中に追加できる属性を含めることができます (属性ごとに名前、タイプ、値の 3 つを指定する必要があります)。

後で、新しい ReceiveMessageRequest オブジェクトを作成するときに、「withMessageAttributeNames」を使用して属性を指定できます。実際には、その特定の属性を含むメッセージに対してキューがフィルター処理されます。

例えば：

String queueUrl = sqs.getQueueUrl("myQueue").getQueueUrl();

ReceiveMessageRequest receiveRequest = new ReceiveMessageRequest(queueUrl);

receiveRequest.withMaxNumberOfMessages(10);
receiveRequest.withMessageAttributeNames("myTag");

Queue が 5 つのメッセージを保持していて、"myTag" 属性を持つメッセージが 1 つだけだった場合、その特定のメッセージだけが返されます。

これはReceiveMessageRequest APIで言及されていないので、私にとって圧倒的でした

したがって、基本的には、各メッセージに一意の属性を与えるだけです (属性の制限に注意してください:The message attribute name can contain the following characters: A-Z, a-z, 0-9, underscore (_), hyphen (-), and period (.). The name must not start or end with a period, and it should not have successive periods. The name is case sensitive and must be unique among all attribute names for the message. The name can be up to 256 characters long. The name cannot start with "AWS." or "Amazon." (or any variations in casing)

于 2016-04-02T21:24:53.210 に答える