使用 .Net core 3.1 和 Visual Studio 2019 工作模板。
我按照网上找到的示例代码进行操作。后台任务 MessageReadingService 从 Amazon SQS 读取消息。消息被写入 ChannelReader。后台任务 MessageProcessingService，从 ChannelReader 读取消息并对其进行处理。
这对第一条消息按预期工作。在 MessageProcessingService 中设置断点会在第一条消息之后停止，但不会再停止。我假设任务正在退出。
我的代码遵循示例代码的模式（效果很好）。任何的意见都将会有帮助。也许是一种调试技术？谢谢！
public class Program
{
public static void Main(string[] args)
{
CreateHostBuilder(args).Build().Run();
}
public static IHostBuilder CreateHostBuilder(string[] args) =>
Host.CreateDefaultBuilder(args)
.ConfigureServices((hostContext, services) =>
{
services.Configure<AwsServicesConfiguration>(hostContext.Configuration.GetSection("AWS"));
services.AddAWSService<IAmazonSQS>();
services.AddTransient<IMessageProcessor, AwsMessageProcessor>();
services.AddSingleton<ISqsMessageChannel, SqsMessageChannel>();
services.AddSingleton<ISqsMessageDeleter, SqsMessageDeleter>();
services.AddSingleton<ISqsMessageQueue, SqsMessageQueue>();
services.AddHostedService<MessageReadingService>();
services.AddHostedService<MessageProcessingService>();
});
}
}
// Message Reading Service
public class MessageReadingService : BackgroundService
{
private readonly ILogger<MessageReadingService> _logger;
private readonly ISqsMessageChannel _sqsMessageChannel;
private readonly string _queueUrl;
private static AmazonSQSClient sqs;
public long ReceivesAttempted { get; private set; }
public long MessagesReceived { get; private set; }
public MessageReadingService(
ILogger<MessageReadingService> logger,
ISqsMessageQueue sqsMessageQueue,
IOptions<AwsServicesConfiguration> options,
ISqsMessageChannel sqsMessageChannel)
{
_logger = logger;
MComSettingsManager mComSettingsManager = new MComSettingsManager();
BasicAWSCredentials awsCredentials = new BasicAWSCredentials(Access_Key,Secret_Key);
AmazonSQSConfig amazonSQSConfig = new AmazonSQSConfig();
amazonSQSConfig.ServiceURL = ServiceURL;
sqs = new AmazonSQSClient(awsCredentials, amazonSQSConfig);
_sqsMessageChannel = sqsMessageChannel;
_queueUrl = options.Value.QueueUrl;
_logger.LogInformation($"Reading from {_queueUrl}");
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
_logger.LogInformation("Started queue reading service.");
var receiveMessageRequest = new ReceiveMessageRequest
{
AttributeNames = new List<string>() { "Body" },
MaxNumberOfMessages = 10,
QueueUrl = _queueUrl,
WaitTimeSeconds = 5
};
while (!stoppingToken.IsCancellationRequested)
{
ReceivesAttempted++;
var receiveMessageResponse =
await sqs.ReceiveMessageAsync(receiveMessageRequest, stoppingToken);
if (receiveMessageResponse.HttpStatusCode == HttpStatusCode.OK &&
receiveMessageResponse.Messages.Any())
{
MessagesReceived += receiveMessageResponse.Messages.Count;
_logger.LogInformation("Received {MessageCount} messages from the queue.",
receiveMessageResponse.Messages.Count);
await _sqsMessageChannel.WriteMessagesAsync(receiveMessageResponse.Messages, stoppingToken);
}
else if (receiveMessageResponse.HttpStatusCode == HttpStatusCode.OK)
{
_logger.LogInformation("No messages received. Attempting receive again in 1 minute.",
receiveMessageResponse.Messages.Count);
await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
}
else if (receiveMessageResponse.HttpStatusCode != HttpStatusCode.OK)
{
_logger.LogError("Unsuccessful response from AWS SQS.");
}
}
_sqsMessageChannel.TryCompleteWriter();
}
public class MessageProcessingService : BackgroundService
{
private readonly ILogger<MessageProcessingService> _logger;
private readonly ISqsMessageChannel _sqsMessageChannel;
private readonly IServiceProvider _serviceProvider;
private readonly MySqlConnection conn;
public MessageProcessingService(
ILogger<MessageProcessingService> logger,
ISqsMessageChannel sqsMessageChannel,
IServiceProvider serviceProvider
)
{
_logger = logger;
_sqsMessageChannel = sqsMessageChannel;
_serviceProvider = serviceProvider;
conn = new MySqlConnection(GaaSManager.ConnectionString);
conn.Open();
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
await foreach (var message in _sqsMessageChannel.Reader.ReadAllAsync(stoppingToken))
{
_logger.LogInformation("Read message to process from channel.");
using var scope = _serviceProvider.CreateScope();
var messageProcessor = scope.ServiceProvider.GetRequiredService<IMessageProcessor>();
await messageProcessor.ProcessOperationsFromMessageAsync(message, conn, stoppingToken);
}
}
}
}
public interface IMessageProcessor
{
Task ProcessOperationsFromMessageAsync(Message message, MySqlConnection conn,CancellationToken cancellationToken = default);
}
public class AwsMessageProcessor : IMessageProcessor
{
private readonly ISqsMessageDeleter _sqsMessageDeleter;
public AwsMessageProcessor(ISqsMessageDeleter sqsMessageDeleter)
{
_sqsMessageDeleter = sqsMessageDeleter;
}
public async Task ProcessOperationsFromMessageAsync(Message message, MySqlConnection conn, CancellationToken cancellationToken = default)
{
PayloadManager payloadMgr = PayloadManager.Instance;
Class1 object1 = JsonConvert.DeserializeObject<Class1>(message.Body);
if (object1.field1 == VALUE)
{
// store for later processing
Class2 object2 = JsonConvert.DeserializeObject<Class2>(object1.JsonWork);
ProcessObject(object2,conn);
}
else
{
SaveClass3(object1,conn);
}
await _sqsMessageDeleter.DeleteMessageAsync(message);
}
public void ProcessObject(Class2 object2, MySqlConnection conn)
{
Working MySQL Code
}
public void SaveClass3(Class3 object3, MySqlConnection conn)
{
Working MySQL Code
}
}