AWS 文档 AWS SDK示例 GitHub 存储库中还有更多SDK示例
本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。
使用亚马逊的SNS示例 AWS SDK for .NET
以下代码示例向您展示如何在 Amazon 中使用来执行操作和实现常见场景SNS。 AWS SDK for .NET
操作是大型程序的代码摘录,必须在上下文中运行。您可以通过操作了解如何调用单个服务函数,还可以通过函数相关场景和跨服务示例的上下文查看操作。
场景是展示如何通过在同一服务中调用多个函数来完成特定任务任务的代码示例。
每个示例都包含一个指向的链接 GitHub,您可以在其中找到有关如何在上下文中设置和运行代码的说明。
开始使用
以下代码示例展示了如何开始使用 Amazon SNS。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 using Amazon.SimpleNotificationService; using Amazon.SimpleNotificationService.Model; namespace SNSActions; public static class HelloSNS { static async Task Main(string[] args) { var snsClient = new AmazonSimpleNotificationServiceClient(); Console.WriteLine($"Hello Amazon SNS! Following are some of your topics:"); Console.WriteLine(); // You can use await and any of the async methods to get a response. // Let's get a list of topics. var response = await snsClient.ListTopicsAsync( new ListTopicsRequest()); foreach (var topic in response.Topics) { Console.WriteLine($"\tTopic ARN: {topic.TopicArn}"); Console.WriteLine(); } } }
-
有关API详细信息,请参阅 “AWS SDK for .NET API参考 ListTopics” 中的。
-
操作
以下代码示例演示如何使用 CheckIfPhoneNumberIsOptedOut
。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 using System; using System.Threading.Tasks; using Amazon.SimpleNotificationService; using Amazon.SimpleNotificationService.Model; /// <summary> /// This example shows how to use the Amazon Simple Notification Service /// (Amazon SNS) to check whether a phone number has been opted out. /// </summary> public class IsPhoneNumOptedOut { public static async Task Main() { string phoneNumber = "+15551112222"; IAmazonSimpleNotificationService client = new AmazonSimpleNotificationServiceClient(); await CheckIfOptedOutAsync(client, phoneNumber); } /// <summary> /// Checks to see if the supplied phone number has been opted out. /// </summary> /// <param name="client">The initialized Amazon SNS Client object used /// to check if the phone number has been opted out.</param> /// <param name="phoneNumber">A string representing the phone number /// to check.</param> public static async Task CheckIfOptedOutAsync(IAmazonSimpleNotificationService client, string phoneNumber) { var request = new CheckIfPhoneNumberIsOptedOutRequest { PhoneNumber = phoneNumber, }; try { var response = await client.CheckIfPhoneNumberIsOptedOutAsync(request); if (response.HttpStatusCode == System.Net.HttpStatusCode.OK) { string optOutStatus = response.IsOptedOut ? "opted out" : "not opted out."; Console.WriteLine($"The phone number: {phoneNumber} is {optOutStatus}"); } } catch (AuthorizationErrorException ex) { Console.WriteLine($"{ex.Message}"); } } }
-
有关API详细信息,请参阅 “AWS SDK for .NET API参考 CheckIfPhoneNumberIsOptedOut” 中的。
-
以下代码示例演示如何使用 CreateTopic
。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 使用特定的名称创建主题。
using System; using System.Threading.Tasks; using Amazon.SimpleNotificationService; using Amazon.SimpleNotificationService.Model; /// <summary> /// This example shows how to use Amazon Simple Notification Service /// (Amazon SNS) to add a new Amazon SNS topic. /// </summary> public class CreateSNSTopic { public static async Task Main() { string topicName = "ExampleSNSTopic"; IAmazonSimpleNotificationService client = new AmazonSimpleNotificationServiceClient(); var topicArn = await CreateSNSTopicAsync(client, topicName); Console.WriteLine($"New topic ARN: {topicArn}"); } /// <summary> /// Creates a new SNS topic using the supplied topic name. /// </summary> /// <param name="client">The initialized SNS client object used to /// create the new topic.</param> /// <param name="topicName">A string representing the topic name.</param> /// <returns>The Amazon Resource Name (ARN) of the created topic.</returns> public static async Task<string> CreateSNSTopicAsync(IAmazonSimpleNotificationService client, string topicName) { var request = new CreateTopicRequest { Name = topicName, }; var response = await client.CreateTopicAsync(request); return response.TopicArn; } }
创建一个具有名称、特定属FIFO性和重复数据消除属性的新主题。
/// <summary> /// Create a new topic with a name and specific FIFO and de-duplication attributes. /// </summary> /// <param name="topicName">The name for the topic.</param> /// <param name="useFifoTopic">True to use a FIFO topic.</param> /// <param name="useContentBasedDeduplication">True to use content-based de-duplication.</param> /// <returns>The ARN of the new topic.</returns> public async Task<string> CreateTopicWithName(string topicName, bool useFifoTopic, bool useContentBasedDeduplication) { var createTopicRequest = new CreateTopicRequest() { Name = topicName, }; if (useFifoTopic) { // Update the name if it is not correct for a FIFO topic. if (!topicName.EndsWith(".fifo")) { createTopicRequest.Name = topicName + ".fifo"; } // Add the attributes from the method parameters. createTopicRequest.Attributes = new Dictionary<string, string> { { "FifoTopic", "true" } }; if (useContentBasedDeduplication) { createTopicRequest.Attributes.Add("ContentBasedDeduplication", "true"); } } var createResponse = await _amazonSNSClient.CreateTopicAsync(createTopicRequest); return createResponse.TopicArn; }
-
有关API详细信息,请参阅 “AWS SDK for .NET API参考 CreateTopic” 中的。
-
以下代码示例演示如何使用 DeleteTopic
。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 按主题删除主题ARN。
/// <summary> /// Delete a topic by its topic ARN. /// </summary> /// <param name="topicArn">The ARN of the topic.</param> /// <returns>True if successful.</returns> public async Task<bool> DeleteTopicByArn(string topicArn) { var deleteResponse = await _amazonSNSClient.DeleteTopicAsync( new DeleteTopicRequest() { TopicArn = topicArn }); return deleteResponse.HttpStatusCode == HttpStatusCode.OK; }
-
有关API详细信息,请参阅 “AWS SDK for .NET API参考 DeleteTopic” 中的。
-
以下代码示例演示如何使用 GetTopicAttributes
。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 using System; using System.Collections.Generic; using System.Threading.Tasks; using Amazon.SimpleNotificationService; /// <summary> /// This example shows how to retrieve the attributes of an Amazon Simple /// Notification Service (Amazon SNS) topic. /// </summary> public class GetTopicAttributes { public static async Task Main() { string topicArn = "arn:aws:sns:us-west-2:000000000000:ExampleSNSTopic"; IAmazonSimpleNotificationService client = new AmazonSimpleNotificationServiceClient(); var attributes = await GetTopicAttributesAsync(client, topicArn); DisplayTopicAttributes(attributes); } /// <summary> /// Given the ARN of the Amazon SNS topic, this method retrieves the topic /// attributes. /// </summary> /// <param name="client">The initialized Amazon SNS client object used /// to retrieve the attributes for the Amazon SNS topic.</param> /// <param name="topicArn">The ARN of the topic for which to retrieve /// the attributes.</param> /// <returns>A Dictionary of topic attributes.</returns> public static async Task<Dictionary<string, string>> GetTopicAttributesAsync( IAmazonSimpleNotificationService client, string topicArn) { var response = await client.GetTopicAttributesAsync(topicArn); return response.Attributes; } /// <summary> /// This method displays the attributes for an Amazon SNS topic. /// </summary> /// <param name="topicAttributes">A Dictionary containing the /// attributes for an Amazon SNS topic.</param> public static void DisplayTopicAttributes(Dictionary<string, string> topicAttributes) { foreach (KeyValuePair<string, string> entry in topicAttributes) { Console.WriteLine($"{entry.Key}: {entry.Value}\n"); } } }
-
有关API详细信息,请参阅 “AWS SDK for .NET API参考 GetTopicAttributes” 中的。
-
以下代码示例演示如何使用 ListSubscriptions
。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 using System; using System.Collections.Generic; using System.Threading.Tasks; using Amazon.SimpleNotificationService; using Amazon.SimpleNotificationService.Model; /// <summary> /// This example will retrieve a list of the existing Amazon Simple /// Notification Service (Amazon SNS) subscriptions. /// </summary> public class ListSubscriptions { public static async Task Main() { IAmazonSimpleNotificationService client = new AmazonSimpleNotificationServiceClient(); Console.WriteLine("Enter a topic ARN to list subscriptions for a specific topic, " + "or press Enter to list subscriptions for all topics."); var topicArn = Console.ReadLine(); Console.WriteLine(); var subscriptions = await GetSubscriptionsListAsync(client, topicArn); DisplaySubscriptionList(subscriptions); } /// <summary> /// Gets a list of the existing Amazon SNS subscriptions, optionally by specifying a topic ARN. /// </summary> /// <param name="client">The initialized Amazon SNS client object used /// to obtain the list of subscriptions.</param> /// <param name="topicArn">The optional ARN of a specific topic. Defaults to null.</param> /// <returns>A list containing information about each subscription.</returns> public static async Task<List<Subscription>> GetSubscriptionsListAsync(IAmazonSimpleNotificationService client, string topicArn = null) { var results = new List<Subscription>(); if (!string.IsNullOrEmpty(topicArn)) { var paginateByTopic = client.Paginators.ListSubscriptionsByTopic( new ListSubscriptionsByTopicRequest() { TopicArn = topicArn, }); // Get the entire list using the paginator. await foreach (var subscription in paginateByTopic.Subscriptions) { results.Add(subscription); } } else { var paginateAllSubscriptions = client.Paginators.ListSubscriptions(new ListSubscriptionsRequest()); // Get the entire list using the paginator. await foreach (var subscription in paginateAllSubscriptions.Subscriptions) { results.Add(subscription); } } return results; } /// <summary> /// Display a list of Amazon SNS subscription information. /// </summary> /// <param name="subscriptionList">A list containing details for existing /// Amazon SNS subscriptions.</param> public static void DisplaySubscriptionList(List<Subscription> subscriptionList) { foreach (var subscription in subscriptionList) { Console.WriteLine($"Owner: {subscription.Owner}"); Console.WriteLine($"Subscription ARN: {subscription.SubscriptionArn}"); Console.WriteLine($"Topic ARN: {subscription.TopicArn}"); Console.WriteLine($"Endpoint: {subscription.Endpoint}"); Console.WriteLine($"Protocol: {subscription.Protocol}"); Console.WriteLine(); } } }
-
有关API详细信息,请参阅 “AWS SDK for .NET API参考 ListSubscriptions” 中的。
-
以下代码示例演示如何使用 ListTopics
。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 using System; using System.Collections.Generic; using System.Threading.Tasks; using Amazon.SimpleNotificationService; using Amazon.SimpleNotificationService.Model; /// <summary> /// Lists the Amazon Simple Notification Service (Amazon SNS) /// topics for the current account. /// </summary> public class ListSNSTopics { public static async Task Main() { IAmazonSimpleNotificationService client = new AmazonSimpleNotificationServiceClient(); await GetTopicListAsync(client); } /// <summary> /// Retrieves the list of Amazon SNS topics in groups of up to 100 /// topics. /// </summary> /// <param name="client">The initialized Amazon SNS client object used /// to retrieve the list of topics.</param> public static async Task GetTopicListAsync(IAmazonSimpleNotificationService client) { // If there are more than 100 Amazon SNS topics, the call to // ListTopicsAsync will return a value to pass to the // method to retrieve the next 100 (or less) topics. string nextToken = string.Empty; do { var response = await client.ListTopicsAsync(nextToken); DisplayTopicsList(response.Topics); nextToken = response.NextToken; } while (!string.IsNullOrEmpty(nextToken)); } /// <summary> /// Displays the list of Amazon SNS Topic ARNs. /// </summary> /// <param name="topicList">The list of Topic ARNs.</param> public static void DisplayTopicsList(List<Topic> topicList) { foreach (var topic in topicList) { Console.WriteLine($"{topic.TopicArn}"); } } }
-
有关API详细信息,请参阅 “AWS SDK for .NET API参考 ListTopics” 中的。
-
以下代码示例演示如何使用 Publish
。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 向主题发布消息。
using System; using System.Threading.Tasks; using Amazon.SimpleNotificationService; using Amazon.SimpleNotificationService.Model; /// <summary> /// This example publishes a message to an Amazon Simple Notification /// Service (Amazon SNS) topic. /// </summary> public class PublishToSNSTopic { public static async Task Main() { string topicArn = "arn:aws:sns:us-east-2:000000000000:ExampleSNSTopic"; string messageText = "This is an example message to publish to the ExampleSNSTopic."; IAmazonSimpleNotificationService client = new AmazonSimpleNotificationServiceClient(); await PublishToTopicAsync(client, topicArn, messageText); } /// <summary> /// Publishes a message to an Amazon SNS topic. /// </summary> /// <param name="client">The initialized client object used to publish /// to the Amazon SNS topic.</param> /// <param name="topicArn">The ARN of the topic.</param> /// <param name="messageText">The text of the message.</param> public static async Task PublishToTopicAsync( IAmazonSimpleNotificationService client, string topicArn, string messageText) { var request = new PublishRequest { TopicArn = topicArn, Message = messageText, }; var response = await client.PublishAsync(request); Console.WriteLine($"Successfully published message ID: {response.MessageId}"); } }
使用组、复制和属性选项向主题发布消息。
/// <summary> /// Publish messages using user settings. /// </summary> /// <returns>Async task.</returns> public static async Task PublishMessages() { Console.WriteLine("Now we can publish messages."); var keepSendingMessages = true; string? deduplicationId = null; string? toneAttribute = null; while (keepSendingMessages) { Console.WriteLine(); var message = GetUserResponse("Enter a message to publish.", "This is a sample message"); if (_useFifoTopic) { Console.WriteLine("Because you are using a FIFO topic, you must set a message group ID." + "\r\nAll messages within the same group will be received in the order " + "they were published."); Console.WriteLine(); var messageGroupId = GetUserResponse("Enter a message group ID for this message:", "1"); if (!_useContentBasedDeduplication) { Console.WriteLine("Because you are not using content-based deduplication, " + "you must enter a deduplication ID."); Console.WriteLine("Enter a deduplication ID for this message."); deduplicationId = GetUserResponse("Enter a deduplication ID for this message.", "1"); } if (GetYesNoResponse("Add an attribute to this message?")) { Console.WriteLine("Enter a number for an attribute."); for (int i = 0; i < _tones.Length; i++) { Console.WriteLine($"\t{i + 1}. {_tones[i]}"); } var selection = GetUserResponse("", "1"); int.TryParse(selection, out var selectionNumber); if (selectionNumber > 0 && selectionNumber < _tones.Length) { toneAttribute = _tones[selectionNumber - 1]; } } var messageID = await SnsWrapper.PublishToTopicWithAttribute( _topicArn, message, "tone", toneAttribute, deduplicationId, messageGroupId); Console.WriteLine($"Message published with id {messageID}."); } keepSendingMessages = GetYesNoResponse("Send another message?", false); } }
将用户的选择应用于发布操作。
/// <summary> /// Publish a message to a topic with an attribute and optional deduplication and group IDs. /// </summary> /// <param name="topicArn">The ARN of the topic.</param> /// <param name="message">The message to publish.</param> /// <param name="attributeName">The optional attribute for the message.</param> /// <param name="attributeValue">The optional attribute value for the message.</param> /// <param name="deduplicationId">The optional deduplication ID for the message.</param> /// <param name="groupId">The optional group ID for the message.</param> /// <returns>The ID of the message published.</returns> public async Task<string> PublishToTopicWithAttribute( string topicArn, string message, string? attributeName = null, string? attributeValue = null, string? deduplicationId = null, string? groupId = null) { var publishRequest = new PublishRequest() { TopicArn = topicArn, Message = message, MessageDeduplicationId = deduplicationId, MessageGroupId = groupId }; if (attributeValue != null) { // Add the string attribute if it exists. publishRequest.MessageAttributes = new Dictionary<string, MessageAttributeValue> { { attributeName!, new MessageAttributeValue() { StringValue = attributeValue, DataType = "String"} } }; } var publishResponse = await _amazonSNSClient.PublishAsync(publishRequest); return publishResponse.MessageId; }
-
有关API详细信息,请参阅在AWS SDK for .NET API参考中发布。
-
以下代码示例演示如何使用 Subscribe
。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 通过电子邮件地址订阅主题。
/// <summary> /// Creates a new subscription to a topic. /// </summary> /// <param name="client">The initialized Amazon SNS client object, used /// to create an Amazon SNS subscription.</param> /// <param name="topicArn">The ARN of the topic to subscribe to.</param> /// <returns>A SubscribeResponse object which includes the subscription /// ARN for the new subscription.</returns> public static async Task<SubscribeResponse> TopicSubscribeAsync( IAmazonSimpleNotificationService client, string topicArn) { SubscribeRequest request = new SubscribeRequest() { TopicArn = topicArn, ReturnSubscriptionArn = true, Protocol = "email", Endpoint = "recipient@example.com", }; var response = await client.SubscribeAsync(request); return response; }
使用可选筛选条件为队列订阅主题。
/// <summary> /// Subscribe a queue to a topic with optional filters. /// </summary> /// <param name="topicArn">The ARN of the topic.</param> /// <param name="useFifoTopic">The optional filtering policy for the subscription.</param> /// <param name="queueArn">The ARN of the queue.</param> /// <returns>The ARN of the new subscription.</returns> public async Task<string> SubscribeTopicWithFilter(string topicArn, string? filterPolicy, string queueArn) { var subscribeRequest = new SubscribeRequest() { TopicArn = topicArn, Protocol = "sqs", Endpoint = queueArn }; if (!string.IsNullOrEmpty(filterPolicy)) { subscribeRequest.Attributes = new Dictionary<string, string> { { "FilterPolicy", filterPolicy } }; } var subscribeResponse = await _amazonSNSClient.SubscribeAsync(subscribeRequest); return subscribeResponse.SubscriptionArn; }
-
有关API详细信息,请参阅AWS SDK for .NET API参考资料中的订阅。
-
以下代码示例演示如何使用 Unsubscribe
。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 通过订阅取消订阅ARN某个主题。
/// <summary> /// Unsubscribe from a topic by a subscription ARN. /// </summary> /// <param name="subscriptionArn">The ARN of the subscription.</param> /// <returns>True if successful.</returns> public async Task<bool> UnsubscribeByArn(string subscriptionArn) { var unsubscribeResponse = await _amazonSNSClient.UnsubscribeAsync( new UnsubscribeRequest() { SubscriptionArn = subscriptionArn }); return unsubscribeResponse.HttpStatusCode == HttpStatusCode.OK; }
-
有关API详细信息,请参阅 “AWS SDK for .NET API参考资料” 中的 “取消订阅”。
-
场景
以下代码示例展示了如何使用 Amazon 发布SMS消息SNS。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 namespace SNSMessageExample { using System; using System.Threading.Tasks; using Amazon; using Amazon.SimpleNotificationService; using Amazon.SimpleNotificationService.Model; public class SNSMessage { private AmazonSimpleNotificationServiceClient snsClient; /// <summary> /// Initializes a new instance of the <see cref="SNSMessage"/> class. /// Constructs a new SNSMessage object initializing the Amazon Simple /// Notification Service (Amazon SNS) client using the supplied /// Region endpoint. /// </summary> /// <param name="regionEndpoint">The Amazon Region endpoint to use in /// sending test messages with this object.</param> public SNSMessage(RegionEndpoint regionEndpoint) { snsClient = new AmazonSimpleNotificationServiceClient(regionEndpoint); } /// <summary> /// Sends the SMS message passed in the text parameter to the phone number /// in phoneNum. /// </summary> /// <param name="phoneNum">The ten-digit phone number to which the text /// message will be sent.</param> /// <param name="text">The text of the message to send.</param> /// <returns>Async task.</returns> public async Task SendTextMessageAsync(string phoneNum, string text) { if (string.IsNullOrEmpty(phoneNum) || string.IsNullOrEmpty(text)) { return; } // Now actually send the message. var request = new PublishRequest { Message = text, PhoneNumber = phoneNum, }; try { var response = await snsClient.PublishAsync(request); } catch (Exception ex) { Console.WriteLine($"Error sending message: {ex}"); } } } }
-
有关API详细信息,请参阅在AWS SDK for .NET API参考中发布。
-
以下代码示例展示了如何:
创建主题(FIFO或非-FIFO)。
针对主题订阅多个队列,并提供应用筛选条件的选项。
将消息发布到主题。
轮询队列中是否有收到的消息。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在 AWS 代码示例存储库
中查找完整实例,了解如何进行设置和运行。 在命令提示符中运行交互式场景。
/// <summary> /// Console application to run a workflow scenario for topics and queues. /// </summary> public static class TopicsAndQueues { private static bool _useFifoTopic = false; private static bool _useContentBasedDeduplication = false; private static string _topicName = null!; private static string _topicArn = null!; private static readonly int _queueCount = 2; private static readonly string[] _queueUrls = new string[_queueCount]; private static readonly string[] _subscriptionArns = new string[_queueCount]; private static readonly string[] _tones = { "cheerful", "funny", "serious", "sincere" }; public static SNSWrapper SnsWrapper { get; set; } = null!; public static SQSWrapper SqsWrapper { get; set; } = null!; public static bool UseConsole { get; set; } = true; static async Task Main(string[] args) { // Set up dependency injection for Amazon EventBridge. using var host = Host.CreateDefaultBuilder(args) .ConfigureLogging(logging => logging.AddFilter("System", LogLevel.Debug) .AddFilter<DebugLoggerProvider>("Microsoft", LogLevel.Information) .AddFilter<ConsoleLoggerProvider>("Microsoft", LogLevel.Trace)) .ConfigureServices((_, services) => services.AddAWSService<IAmazonSQS>() .AddAWSService<IAmazonSimpleNotificationService>() .AddTransient<SNSWrapper>() .AddTransient<SQSWrapper>() ) .Build(); ServicesSetup(host); PrintDescription(); await RunScenario(); } /// <summary> /// Populate the services for use within the console application. /// </summary> /// <param name="host">The services host.</param> private static void ServicesSetup(IHost host) { SnsWrapper = host.Services.GetRequiredService<SNSWrapper>(); SqsWrapper = host.Services.GetRequiredService<SQSWrapper>(); } /// <summary> /// Run the scenario for working with topics and queues. /// </summary> /// <returns>True if successful.</returns> public static async Task<bool> RunScenario() { try { await SetupTopic(); await SetupQueues(); await PublishMessages(); foreach (var queueUrl in _queueUrls) { var messages = await PollForMessages(queueUrl); if (messages.Any()) { await DeleteMessages(queueUrl, messages); } } await CleanupResources(); Console.WriteLine("Messaging with topics and queues workflow is complete."); return true; } catch (Exception ex) { Console.WriteLine(new string('-', 80)); Console.WriteLine($"There was a problem running the scenario: {ex.Message}"); await CleanupResources(); Console.WriteLine(new string('-', 80)); return false; } } /// <summary> /// Print a description for the tasks in the workflow. /// </summary> /// <returns>Async task.</returns> private static void PrintDescription() { Console.WriteLine(new string('-', 80)); Console.WriteLine($"Welcome to messaging with topics and queues."); Console.WriteLine(new string('-', 80)); Console.WriteLine($"In this workflow, you will create an SNS topic and subscribe {_queueCount} SQS queues to the topic." + $"\r\nYou can select from several options for configuring the topic and the subscriptions for the 2 queues." + $"\r\nYou can then post to the topic and see the results in the queues.\r\n"); Console.WriteLine(new string('-', 80)); } /// <summary> /// Set up the SNS topic to be used with the queues. /// </summary> /// <returns>Async task.</returns> private static async Task<string> SetupTopic() { Console.WriteLine(new string('-', 80)); Console.WriteLine($"SNS topics can be configured as FIFO (First-In-First-Out)." + $"\r\nFIFO topics deliver messages in order and support deduplication and message filtering." + $"\r\nYou can then post to the topic and see the results in the queues.\r\n"); _useFifoTopic = GetYesNoResponse("Would you like to work with FIFO topics?"); if (_useFifoTopic) { Console.WriteLine(new string('-', 80)); _topicName = GetUserResponse("Enter a name for your SNS topic: ", "example-topic"); Console.WriteLine( "Because you have selected a FIFO topic, '.fifo' must be appended to the topic name.\r\n"); Console.WriteLine(new string('-', 80)); Console.WriteLine($"Because you have chosen a FIFO topic, deduplication is supported." + $"\r\nDeduplication IDs are either set in the message or automatically generated " + $"\r\nfrom content using a hash function.\r\n" + $"\r\nIf a message is successfully published to an SNS FIFO topic, any message " + $"\r\npublished and determined to have the same deduplication ID, " + $"\r\nwithin the five-minute deduplication interval, is accepted but not delivered.\r\n" + $"\r\nFor more information about deduplication, " + $"\r\nsee https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html."); _useContentBasedDeduplication = GetYesNoResponse("Use content-based deduplication instead of entering a deduplication ID?"); Console.WriteLine(new string('-', 80)); } _topicArn = await SnsWrapper.CreateTopicWithName(_topicName, _useFifoTopic, _useContentBasedDeduplication); Console.WriteLine($"Your new topic with the name {_topicName}" + $"\r\nand Amazon Resource Name (ARN) {_topicArn}" + $"\r\nhas been created.\r\n"); Console.WriteLine(new string('-', 80)); return _topicArn; } /// <summary> /// Set up the queues. /// </summary> /// <returns>Async task.</returns> private static async Task SetupQueues() { Console.WriteLine(new string('-', 80)); Console.WriteLine($"Now you will create {_queueCount} Amazon Simple Queue Service (Amazon SQS) queues to subscribe to the topic."); // Repeat this section for each queue. for (int i = 0; i < _queueCount; i++) { var queueName = GetUserResponse("Enter a name for an Amazon SQS queue: ", $"example-queue-{i}"); if (_useFifoTopic) { // Only explain this once. if (i == 0) { Console.WriteLine( "Because you have selected a FIFO topic, '.fifo' must be appended to the queue name."); } var queueUrl = await SqsWrapper.CreateQueueWithName(queueName, _useFifoTopic); _queueUrls[i] = queueUrl; Console.WriteLine($"Your new queue with the name {queueName}" + $"\r\nand queue URL {queueUrl}" + $"\r\nhas been created.\r\n"); if (i == 0) { Console.WriteLine( $"The queue URL is used to retrieve the queue ARN,\r\n" + $"which is used to create a subscription."); Console.WriteLine(new string('-', 80)); } var queueArn = await SqsWrapper.GetQueueArnByUrl(queueUrl); if (i == 0) { Console.WriteLine( $"An AWS Identity and Access Management (IAM) policy must be attached to an SQS queue, enabling it to receive\r\n" + $"messages from an SNS topic"); } await SqsWrapper.SetQueuePolicyForTopic(queueArn, _topicArn, queueUrl); await SetupFilters(i, queueArn, queueName); } } Console.WriteLine(new string('-', 80)); } /// <summary> /// Set up filters with user options for a queue. /// </summary> /// <param name="queueCount">The number of this queue.</param> /// <param name="queueArn">The ARN of the queue.</param> /// <param name="queueName">The name of the queue.</param> /// <returns>Async Task.</returns> public static async Task SetupFilters(int queueCount, string queueArn, string queueName) { if (_useFifoTopic) { Console.WriteLine(new string('-', 80)); // Only explain this once. if (queueCount == 0) { Console.WriteLine( "Subscriptions to a FIFO topic can have filters." + "If you add a filter to this subscription, then only the filtered messages " + "will be received in the queue."); Console.WriteLine( "For information about message filtering, " + "see https://docs.aws.amazon.com/sns/latest/dg/sns-message-filtering.html"); Console.WriteLine( "For this example, you can filter messages by a" + "TONE attribute."); } var useFilter = GetYesNoResponse($"Filter messages for {queueName}'s subscription to the topic?"); string? filterPolicy = null; if (useFilter) { filterPolicy = CreateFilterPolicy(); } var subscriptionArn = await SnsWrapper.SubscribeTopicWithFilter(_topicArn, filterPolicy, queueArn); _subscriptionArns[queueCount] = subscriptionArn; Console.WriteLine( $"The queue {queueName} has been subscribed to the topic {_topicName} " + $"with the subscription ARN {subscriptionArn}"); Console.WriteLine(new string('-', 80)); } } /// <summary> /// Use user input to create a filter policy for a subscription. /// </summary> /// <returns>The serialized filter policy.</returns> public static string CreateFilterPolicy() { Console.WriteLine(new string('-', 80)); Console.WriteLine( $"You can filter messages by one or more of the following" + $"TONE attributes."); List<string> filterSelections = new List<string>(); var selectionNumber = 0; do { Console.WriteLine( $"Enter a number to add a TONE filter, or enter 0 to stop adding filters."); for (int i = 0; i < _tones.Length; i++) { Console.WriteLine($"\t{i + 1}. {_tones[i]}"); } var selection = GetUserResponse("", filterSelections.Any() ? "0" : "1"); int.TryParse(selection, out selectionNumber); if (selectionNumber > 0 && !filterSelections.Contains(_tones[selectionNumber - 1])) { filterSelections.Add(_tones[selectionNumber - 1]); } } while (selectionNumber != 0); var filters = new Dictionary<string, List<string>> { { "tone", filterSelections } }; string filterPolicy = JsonSerializer.Serialize(filters); return filterPolicy; } /// <summary> /// Publish messages using user settings. /// </summary> /// <returns>Async task.</returns> public static async Task PublishMessages() { Console.WriteLine("Now we can publish messages."); var keepSendingMessages = true; string? deduplicationId = null; string? toneAttribute = null; while (keepSendingMessages) { Console.WriteLine(); var message = GetUserResponse("Enter a message to publish.", "This is a sample message"); if (_useFifoTopic) { Console.WriteLine("Because you are using a FIFO topic, you must set a message group ID." + "\r\nAll messages within the same group will be received in the order " + "they were published."); Console.WriteLine(); var messageGroupId = GetUserResponse("Enter a message group ID for this message:", "1"); if (!_useContentBasedDeduplication) { Console.WriteLine("Because you are not using content-based deduplication, " + "you must enter a deduplication ID."); Console.WriteLine("Enter a deduplication ID for this message."); deduplicationId = GetUserResponse("Enter a deduplication ID for this message.", "1"); } if (GetYesNoResponse("Add an attribute to this message?")) { Console.WriteLine("Enter a number for an attribute."); for (int i = 0; i < _tones.Length; i++) { Console.WriteLine($"\t{i + 1}. {_tones[i]}"); } var selection = GetUserResponse("", "1"); int.TryParse(selection, out var selectionNumber); if (selectionNumber > 0 && selectionNumber < _tones.Length) { toneAttribute = _tones[selectionNumber - 1]; } } var messageID = await SnsWrapper.PublishToTopicWithAttribute( _topicArn, message, "tone", toneAttribute, deduplicationId, messageGroupId); Console.WriteLine($"Message published with id {messageID}."); } keepSendingMessages = GetYesNoResponse("Send another message?", false); } } /// <summary> /// Poll for the published messages to see the results of the user's choices. /// </summary> /// <returns>Async task.</returns> public static async Task<List<Message>> PollForMessages(string queueUrl) { Console.WriteLine(new string('-', 80)); Console.WriteLine($"Now the SQS queue at {queueUrl} will be polled to retrieve the messages." + "\r\nPress any key to continue."); if (UseConsole) { Console.ReadLine(); } var moreMessages = true; var messages = new List<Message>(); while (moreMessages) { var newMessages = await SqsWrapper.ReceiveMessagesByUrl(queueUrl, 10); moreMessages = newMessages.Any(); if (moreMessages) { messages.AddRange(newMessages); } } Console.WriteLine($"{messages.Count} message(s) were received by the queue at {queueUrl}."); foreach (var message in messages) { Console.WriteLine("\tMessage:" + $"\n\t{message.Body}"); } Console.WriteLine(new string('-', 80)); return messages; } /// <summary> /// Delete the message using handles in a batch. /// </summary> /// <returns>Async task.</returns> public static async Task DeleteMessages(string queueUrl, List<Message> messages) { Console.WriteLine(new string('-', 80)); Console.WriteLine("Now we can delete the messages in this queue in a batch."); await SqsWrapper.DeleteMessageBatchByUrl(queueUrl, messages); Console.WriteLine(new string('-', 80)); } /// <summary> /// Clean up the resources from the scenario. /// </summary> /// <returns>Async task.</returns> private static async Task CleanupResources() { Console.WriteLine(new string('-', 80)); Console.WriteLine($"Clean up resources."); try { foreach (var queueUrl in _queueUrls) { if (!string.IsNullOrEmpty(queueUrl)) { var deleteQueue = GetYesNoResponse($"Delete queue with url {queueUrl}?"); if (deleteQueue) { await SqsWrapper.DeleteQueueByUrl(queueUrl); } } } foreach (var subscriptionArn in _subscriptionArns) { if (!string.IsNullOrEmpty(subscriptionArn)) { await SnsWrapper.UnsubscribeByArn(subscriptionArn); } } var deleteTopic = GetYesNoResponse($"Delete topic {_topicName}?"); if (deleteTopic) { await SnsWrapper.DeleteTopicByArn(_topicArn); } } catch (Exception ex) { Console.WriteLine($"Unable to clean up resources. Here's why: {ex.Message}."); } Console.WriteLine(new string('-', 80)); } /// <summary> /// Helper method to get a yes or no response from the user. /// </summary> /// <param name="question">The question string to print on the console.</param> /// <param name="defaultAnswer">Optional default answer to use.</param> /// <returns>True if the user responds with a yes.</returns> private static bool GetYesNoResponse(string question, bool defaultAnswer = true) { if (UseConsole) { Console.WriteLine(question); var ynResponse = Console.ReadLine(); var response = ynResponse != null && ynResponse.Equals("y", StringComparison.InvariantCultureIgnoreCase); return response; } // If not using the console, use the default. return defaultAnswer; } /// <summary> /// Helper method to get a string response from the user through the console. /// </summary> /// <param name="question">The question string to print on the console.</param> /// <param name="defaultAnswer">Optional default answer to use.</param> /// <returns>True if the user responds with a yes.</returns> private static string GetUserResponse(string question, string defaultAnswer) { if (UseConsole) { var response = ""; while (string.IsNullOrEmpty(response)) { Console.WriteLine(question); response = Console.ReadLine(); } return response; } // If not using the console, use the default. return defaultAnswer; } }
创建一个封装 Amazon SQS 操作的类。
/// <summary> /// Wrapper for Amazon Simple Queue Service (SQS) operations. /// </summary> public class SQSWrapper { private readonly IAmazonSQS _amazonSQSClient; /// <summary> /// Constructor for the Amazon SQS wrapper. /// </summary> /// <param name="amazonSQS">The injected Amazon SQS client.</param> public SQSWrapper(IAmazonSQS amazonSQS) { _amazonSQSClient = amazonSQS; } /// <summary> /// Create a queue with a specific name. /// </summary> /// <param name="queueName">The name for the queue.</param> /// <param name="useFifoQueue">True to use a FIFO queue.</param> /// <returns>The url for the queue.</returns> public async Task<string> CreateQueueWithName(string queueName, bool useFifoQueue) { int maxMessage = 256 * 1024; var queueAttributes = new Dictionary<string, string> { { QueueAttributeName.MaximumMessageSize, maxMessage.ToString() } }; var createQueueRequest = new CreateQueueRequest() { QueueName = queueName, Attributes = queueAttributes }; if (useFifoQueue) { // Update the name if it is not correct for a FIFO queue. if (!queueName.EndsWith(".fifo")) { createQueueRequest.QueueName = queueName + ".fifo"; } // Add an attribute for a FIFO queue. createQueueRequest.Attributes.Add( QueueAttributeName.FifoQueue, "true"); } var createResponse = await _amazonSQSClient.CreateQueueAsync( new CreateQueueRequest() { QueueName = queueName }); return createResponse.QueueUrl; } /// <summary> /// Get the ARN for a queue from its URL. /// </summary> /// <param name="queueUrl">The URL of the queue.</param> /// <returns>The ARN of the queue.</returns> public async Task<string> GetQueueArnByUrl(string queueUrl) { var getAttributesRequest = new GetQueueAttributesRequest() { QueueUrl = queueUrl, AttributeNames = new List<string>() { QueueAttributeName.QueueArn } }; var getAttributesResponse = await _amazonSQSClient.GetQueueAttributesAsync( getAttributesRequest); return getAttributesResponse.QueueARN; } /// <summary> /// Set the policy attribute of a queue for a topic. /// </summary> /// <param name="queueArn">The ARN of the queue.</param> /// <param name="topicArn">The ARN of the topic.</param> /// <param name="queueUrl">The url for the queue.</param> /// <returns>True if successful.</returns> public async Task<bool> SetQueuePolicyForTopic(string queueArn, string topicArn, string queueUrl) { var queuePolicy = "{" + "\"Version\": \"2012-10-17\"," + "\"Statement\": [{" + "\"Effect\": \"Allow\"," + "\"Principal\": {" + $"\"Service\": " + "\"sns.amazonaws.com\"" + "}," + "\"Action\": \"sqs:SendMessage\"," + $"\"Resource\": \"{queueArn}\"," + "\"Condition\": {" + "\"ArnEquals\": {" + $"\"aws:SourceArn\": \"{topicArn}\"" + "}" + "}" + "}]" + "}"; var attributesResponse = await _amazonSQSClient.SetQueueAttributesAsync( new SetQueueAttributesRequest() { QueueUrl = queueUrl, Attributes = new Dictionary<string, string>() { { "Policy", queuePolicy } } }); return attributesResponse.HttpStatusCode == HttpStatusCode.OK; } /// <summary> /// Receive messages from a queue by its URL. /// </summary> /// <param name="queueUrl">The url of the queue.</param> /// <returns>The list of messages.</returns> public async Task<List<Message>> ReceiveMessagesByUrl(string queueUrl, int maxMessages) { // Setting WaitTimeSeconds to non-zero enables long polling. // For information about long polling, see // https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-short-and-long-polling.html var messageResponse = await _amazonSQSClient.ReceiveMessageAsync( new ReceiveMessageRequest() { QueueUrl = queueUrl, MaxNumberOfMessages = maxMessages, WaitTimeSeconds = 1 }); return messageResponse.Messages; } /// <summary> /// Delete a batch of messages from a queue by its url. /// </summary> /// <param name="queueUrl">The url of the queue.</param> /// <returns>True if successful.</returns> public async Task<bool> DeleteMessageBatchByUrl(string queueUrl, List<Message> messages) { var deleteRequest = new DeleteMessageBatchRequest() { QueueUrl = queueUrl, Entries = new List<DeleteMessageBatchRequestEntry>() }; foreach (var message in messages) { deleteRequest.Entries.Add(new DeleteMessageBatchRequestEntry() { ReceiptHandle = message.ReceiptHandle, Id = message.MessageId }); } var deleteResponse = await _amazonSQSClient.DeleteMessageBatchAsync(deleteRequest); return deleteResponse.Failed.Any(); } /// <summary> /// Delete a queue by its URL. /// </summary> /// <param name="queueUrl">The url of the queue.</param> /// <returns>True if successful.</returns> public async Task<bool> DeleteQueueByUrl(string queueUrl) { var deleteResponse = await _amazonSQSClient.DeleteQueueAsync( new DeleteQueueRequest() { QueueUrl = queueUrl }); return deleteResponse.HttpStatusCode == HttpStatusCode.OK; } }
创建一个封装 Amazon SNS 操作的类。
/// <summary> /// Wrapper for Amazon Simple Notification Service (SNS) operations. /// </summary> public class SNSWrapper { private readonly IAmazonSimpleNotificationService _amazonSNSClient; /// <summary> /// Constructor for the Amazon SNS wrapper. /// </summary> /// <param name="amazonSQS">The injected Amazon SNS client.</param> public SNSWrapper(IAmazonSimpleNotificationService amazonSNS) { _amazonSNSClient = amazonSNS; } /// <summary> /// Create a new topic with a name and specific FIFO and de-duplication attributes. /// </summary> /// <param name="topicName">The name for the topic.</param> /// <param name="useFifoTopic">True to use a FIFO topic.</param> /// <param name="useContentBasedDeduplication">True to use content-based de-duplication.</param> /// <returns>The ARN of the new topic.</returns> public async Task<string> CreateTopicWithName(string topicName, bool useFifoTopic, bool useContentBasedDeduplication) { var createTopicRequest = new CreateTopicRequest() { Name = topicName, }; if (useFifoTopic) { // Update the name if it is not correct for a FIFO topic. if (!topicName.EndsWith(".fifo")) { createTopicRequest.Name = topicName + ".fifo"; } // Add the attributes from the method parameters. createTopicRequest.Attributes = new Dictionary<string, string> { { "FifoTopic", "true" } }; if (useContentBasedDeduplication) { createTopicRequest.Attributes.Add("ContentBasedDeduplication", "true"); } } var createResponse = await _amazonSNSClient.CreateTopicAsync(createTopicRequest); return createResponse.TopicArn; } /// <summary> /// Subscribe a queue to a topic with optional filters. /// </summary> /// <param name="topicArn">The ARN of the topic.</param> /// <param name="useFifoTopic">The optional filtering policy for the subscription.</param> /// <param name="queueArn">The ARN of the queue.</param> /// <returns>The ARN of the new subscription.</returns> public async Task<string> SubscribeTopicWithFilter(string topicArn, string? filterPolicy, string queueArn) { var subscribeRequest = new SubscribeRequest() { TopicArn = topicArn, Protocol = "sqs", Endpoint = queueArn }; if (!string.IsNullOrEmpty(filterPolicy)) { subscribeRequest.Attributes = new Dictionary<string, string> { { "FilterPolicy", filterPolicy } }; } var subscribeResponse = await _amazonSNSClient.SubscribeAsync(subscribeRequest); return subscribeResponse.SubscriptionArn; } /// <summary> /// Publish a message to a topic with an attribute and optional deduplication and group IDs. /// </summary> /// <param name="topicArn">The ARN of the topic.</param> /// <param name="message">The message to publish.</param> /// <param name="attributeName">The optional attribute for the message.</param> /// <param name="attributeValue">The optional attribute value for the message.</param> /// <param name="deduplicationId">The optional deduplication ID for the message.</param> /// <param name="groupId">The optional group ID for the message.</param> /// <returns>The ID of the message published.</returns> public async Task<string> PublishToTopicWithAttribute( string topicArn, string message, string? attributeName = null, string? attributeValue = null, string? deduplicationId = null, string? groupId = null) { var publishRequest = new PublishRequest() { TopicArn = topicArn, Message = message, MessageDeduplicationId = deduplicationId, MessageGroupId = groupId }; if (attributeValue != null) { // Add the string attribute if it exists. publishRequest.MessageAttributes = new Dictionary<string, MessageAttributeValue> { { attributeName!, new MessageAttributeValue() { StringValue = attributeValue, DataType = "String"} } }; } var publishResponse = await _amazonSNSClient.PublishAsync(publishRequest); return publishResponse.MessageId; } /// <summary> /// Unsubscribe from a topic by a subscription ARN. /// </summary> /// <param name="subscriptionArn">The ARN of the subscription.</param> /// <returns>True if successful.</returns> public async Task<bool> UnsubscribeByArn(string subscriptionArn) { var unsubscribeResponse = await _amazonSNSClient.UnsubscribeAsync( new UnsubscribeRequest() { SubscriptionArn = subscriptionArn }); return unsubscribeResponse.HttpStatusCode == HttpStatusCode.OK; } /// <summary> /// Delete a topic by its topic ARN. /// </summary> /// <param name="topicArn">The ARN of the topic.</param> /// <returns>True if successful.</returns> public async Task<bool> DeleteTopicByArn(string topicArn) { var deleteResponse = await _amazonSNSClient.DeleteTopicAsync( new DeleteTopicRequest() { TopicArn = topicArn }); return deleteResponse.HttpStatusCode == HttpStatusCode.OK; } }
-
有关API详细信息,请参阅 “参AWS SDK for .NET API考” 中的以下主题。
-
无服务器示例
以下代码示例说明如何实现一个 Lambda 函数,该函数接收通过接收来自主题的消息而触发的事件。SNS该函数从事件参数检索消息并记录每条消息的内容。
- AWS SDK for .NET
-
注意
还有更多相关信息 GitHub。在无服务器示例
存储库中查找完整示例,并了解如何进行设置和运行。 使用 Lambda 使用一个SNS事件。 NET。
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 using Amazon.Lambda.Core; using Amazon.Lambda.SNSEvents; // Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class. [assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))] namespace SnsIntegration; public class Function { public async Task FunctionHandler(SNSEvent evnt, ILambdaContext context) { foreach (var record in evnt.Records) { await ProcessRecordAsync(record, context); } context.Logger.LogInformation("done"); } private async Task ProcessRecordAsync(SNSEvent.SNSRecord record, ILambdaContext context) { try { context.Logger.LogInformation($"Processed record {record.Sns.Message}"); // TODO: Do interesting work based on the new message await Task.CompletedTask; } catch (Exception e) { //You can use Dead Letter Queue to handle failures. By configuring a Lambda DLQ. context.Logger.LogError($"An error occurred"); throw; } } }