92 lines
3.9 KiB
C#
92 lines
3.9 KiB
C#
using NewLife.MQTT;
|
|
using SimpleAdmin.Cache;
|
|
using SimpleAdmin.Core;
|
|
using SimpleAdmin.SqlSugar;
|
|
using SimpleAdmin.System;
|
|
|
|
namespace SimpleAdmin.MessageCenter;
|
|
|
|
public class MessageWorker : BackgroundService
|
|
{
|
|
private readonly ILogger<MessageWorker> _logger;
|
|
private readonly ISimpleCacheService _simpleCacheService;
|
|
private readonly IMqttClientManager _mqttClientManager;
|
|
private readonly MqttClient _mqttClient;
|
|
|
|
public MessageWorker(ILogger<MessageWorker> logger, ISimpleCacheService simpleCacheService, IMqttClientManager mqttClientManager)
|
|
{
|
|
_logger = logger;
|
|
_simpleCacheService = simpleCacheService;
|
|
_mqttClientManager = mqttClientManager;
|
|
_mqttClient = mqttClientManager.GetClient();
|
|
}
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
//获取延迟队列
|
|
var queue = _simpleCacheService.GetDelayQueue<long>(CacheConst.CACHE_NOTIFICATION);
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
//一次拿十条，如果拿一条就用queue.TakeOneAsync(-1);-1是超时时间，默认0秒永远阻塞；负数表示直接返回，不阻塞。
|
|
var data = await queue.TakeOneAsync(-1);
|
|
if (data != 0)
|
|
{
|
|
_logger.LogDebug($"消费者收到消息,消息ID:{data},时间：{DateTime.Now}");
|
|
var db = DbContext.DB.CopyNew();
|
|
//获取消息
|
|
var message = await db.Queryable<SysMessage>().InSingleAsync(data);
|
|
if (message != null)
|
|
{
|
|
message.Status = SysDictConst.MESSAGE_STATUS_ALREADY;
|
|
//获取待发送的消息
|
|
var messageUsers = await db.Queryable<SysMessageUser>()
|
|
.Where(it => it.MessageId == message.Id && it.Status == SysDictConst.MESSAGE_STATUS_READY).ToListAsync();
|
|
var hasError = false;
|
|
//开启事务
|
|
var result = await db.UseTranAsync(async () =>
|
|
{
|
|
messageUsers.ForEach(it =>
|
|
{
|
|
try
|
|
{
|
|
//发送消息
|
|
_mqttClient.PublishAsync(MqttConst.MQTT_TOPIC_PREFIX + it.UserId, new MqttMessage()
|
|
{
|
|
MsgType = MqttConst.MQTT_MESSAGE_NEW,
|
|
Data = new MessageData()
|
|
{
|
|
Subject = message.Subject,
|
|
Content = message.Content
|
|
}
|
|
});
|
|
it.Status = SysDictConst.MESSAGE_STATUS_ALREADY;
|
|
it.UpdateTime = DateTime.Now;
|
|
}
|
|
catch (Exception e)
|
|
{
|
|
hasError = true;
|
|
_logger.LogError($"发送消息失败:{e.Message}");
|
|
}
|
|
});
|
|
await db.Updateable(messageUsers).ExecuteCommandAsync();
|
|
await db.Updateable(message).ExecuteCommandAsync();
|
|
});
|
|
//如果有失败的，重写发到延迟队列
|
|
if (hasError)
|
|
{
|
|
_logger.LogDebug($"有失败的消息，重新发到延迟队列");
|
|
queue.Add(message.Id, 5);
|
|
}
|
|
}
|
|
queue.Acknowledge(data);//告诉队列已经消费了的数据
|
|
_logger.LogDebug("消费成功");
|
|
}
|
|
else
|
|
{
|
|
_logger.LogDebug("消费者从队列中没有拿到数据:" + DateTime.Now);
|
|
await Task.Delay(1000, stoppingToken);
|
|
}
|
|
}
|
|
}
|
|
}
|