option:启用RabbitMQ的发布确认模式，提高投递消息的可靠性
This commit is contained in:
hardstylewyl
parent
22080431b7
commit
25eb439324
@ -20,9 +20,14 @@ public interface IBroadcastingService : ITransientDependency
|
||||
/// </summary>
|
||||
/// <param name="eto"></param>
|
||||
/// <param name="subSystemCode"></param>
|
||||
/// <param name="trackStatusCode"></param>
|
||||
void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode);
|
||||
|
||||
/// <summary>
|
||||
/// 轨迹推送完成广播（推送完成回调）
|
||||
/// </summary>
|
||||
/// <param name="list"></param>
|
||||
void TrackPushCompletedPublicTopic(List<TrackPushRecordDto> list);
|
||||
|
||||
/// <summary>
|
||||
/// 轨迹推送完成广播（推送完成回调）
|
||||
/// </summary>
|
||||
|
||||
81
src/SJZY.Track.Application/Rabbit/RabbitClientBusExt.cs
Normal file
81
src/SJZY.Track.Application/Rabbit/RabbitClientBusExt.cs
Normal file
@ -0,0 +1,81 @@
|
||||
using System.Diagnostics;
|
||||
using BSI.Stack.Rabbit;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using RabbitMQ.Client;
|
||||
using Volo.Abp.DependencyInjection;
|
||||
|
||||
namespace SJZY.Track.Application.Rabbit;
|
||||
|
||||
/// <summary>
|
||||
/// 支持发布确认模式
|
||||
/// </summary>
|
||||
public class RabbitClientBusExt : RabbitClientBus, ISingletonDependency
|
||||
{
|
||||
public RabbitClientBusExt(IServiceProvider sp, ILogger<RabbitClient> logger)
|
||||
: base(sp, logger, sp.GetRequiredService<IOptionsMonitor<RabbitOptions>>().CurrentValue)
|
||||
{
|
||||
}
|
||||
|
||||
public void PublishByConfirmSelect(
|
||||
string exchange,
|
||||
string routingKey,
|
||||
byte[] data,
|
||||
Action<IBasicProperties>? options = null
|
||||
)
|
||||
{
|
||||
using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey);
|
||||
activity?.SetTag("kind", "publish");
|
||||
var key = exchange + ":" + routingKey;
|
||||
Lazy<IModel> orAdd = Channels.GetOrAdd(key, (Func<string, Lazy<IModel>>) (_ => new Lazy<IModel>(CreateModel)));
|
||||
var basicProperties1 = orAdd.Value.CreateBasicProperties();
|
||||
|
||||
// 配置属性
|
||||
if (options != null)
|
||||
options(basicProperties1);
|
||||
var model = orAdd.Value;
|
||||
|
||||
// 启用发布确认
|
||||
model.ConfirmSelect();
|
||||
var readOnlyMemory = (ReadOnlyMemory<byte>) data;
|
||||
model.BasicPublish(exchange, routingKey, basicProperties1, readOnlyMemory);
|
||||
model.WaitForConfirmsOrDie();
|
||||
}
|
||||
|
||||
public void PublishByConfirmSelect(
|
||||
string exchange,
|
||||
string routingKey,
|
||||
IEnumerable<byte[]> datas,
|
||||
Action<IBasicProperties>? options = null
|
||||
)
|
||||
{
|
||||
using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey);
|
||||
activity?.SetTag("kind", "publish");
|
||||
var key = exchange + ":" + routingKey;
|
||||
Lazy<IModel> orAdd = Channels.GetOrAdd(key, (Func<string, Lazy<IModel>>) (_ => new Lazy<IModel>(CreateModel)));
|
||||
var basicProperties1 = orAdd.Value.CreateBasicProperties();
|
||||
|
||||
// 配置属性
|
||||
if (options != null)
|
||||
options(basicProperties1);
|
||||
var model = orAdd.Value;
|
||||
|
||||
// 启用发布确认
|
||||
model.ConfirmSelect();
|
||||
foreach (var data in datas)
|
||||
{
|
||||
var readOnlyMemory = (ReadOnlyMemory<byte>) data;
|
||||
model.BasicPublish(exchange, routingKey, basicProperties1, readOnlyMemory);
|
||||
}
|
||||
model.WaitForConfirmsOrDie();
|
||||
}
|
||||
|
||||
private IConnection GetConnection()
|
||||
{
|
||||
var key = Connections.OrderBy((Func<KeyValuePair<IConnection, int>, int>) (q => q.Value)).Select((Func<KeyValuePair<IConnection, int>, IConnection>) (q => q.Key)).FirstOrDefault();
|
||||
Connections.AddOrUpdate(key, (Func<IConnection, int>) (_ => 1), (Func<IConnection, int, int>) ((_, v) => v + 1));
|
||||
return key;
|
||||
}
|
||||
|
||||
private IModel CreateModel() => GetConnection().CreateModel();
|
||||
}
|
||||
@ -84,7 +84,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
|
||||
|
||||
// 推送广播
|
||||
if (logs.First().IsToMq)
|
||||
logs.ForEach(log => _broadcastingService.TrackPushCompletedPublicTopic(BuildDto(log)));
|
||||
_broadcastingService.TrackPushCompletedPublicTopic(logs.Select(BuildDto).ToList());
|
||||
|
||||
// 写入推送日志
|
||||
await _trackPushRecordEsRepository.AddManyAsync(logs);
|
||||
|
||||
@ -1,9 +1,9 @@
|
||||
using System.Text;
|
||||
using BSI.Stack.Rabbit;
|
||||
using SJZY.Expand.ABP.Core.PublicMethords;
|
||||
using SJZY.Track.Application.Contract.Dtos.TrackPush;
|
||||
using SJZY.Track.Application.Contract.Events;
|
||||
using SJZY.Track.Application.Contract.Services;
|
||||
using SJZY.Track.Application.Rabbit;
|
||||
|
||||
namespace SJZY.Track.Application.Services.Broadcasting;
|
||||
|
||||
@ -12,9 +12,9 @@ namespace SJZY.Track.Application.Services.Broadcasting;
|
||||
/// </summary>
|
||||
public class BroadcastingService : IBroadcastingService
|
||||
{
|
||||
private readonly RabbitClientBus _rabbitClientBus;
|
||||
private readonly RabbitClientBusExt _rabbitClientBus;
|
||||
|
||||
public BroadcastingService(RabbitClientBus rabbitClientBus)
|
||||
public BroadcastingService(RabbitClientBusExt rabbitClientBus)
|
||||
{
|
||||
_rabbitClientBus = rabbitClientBus;
|
||||
}
|
||||
@ -64,15 +64,29 @@ public class BroadcastingService : IBroadcastingService
|
||||
else
|
||||
data.Option = 3;
|
||||
|
||||
_rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}",
|
||||
Encoding.UTF8.GetBytes(data.ToJson()));
|
||||
_rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}",
|
||||
Encoding.UTF8.GetBytes(data.ToJson()), prop => { prop.Persistent = true; });
|
||||
}
|
||||
|
||||
public void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode)
|
||||
{
|
||||
eto.Option = 2;
|
||||
_rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}",
|
||||
Encoding.UTF8.GetBytes(eto.ToJson()));
|
||||
_rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}",
|
||||
Encoding.UTF8.GetBytes(eto.ToJson()), prop => { prop.Persistent = true; });
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 轨迹推送完成广播（批量）
|
||||
/// </summary>
|
||||
/// <param name="list"></param>
|
||||
/// <returns></returns>
|
||||
public void TrackPushCompletedPublicTopic(List<TrackPushRecordDto> list)
|
||||
{
|
||||
var datas = list
|
||||
.Select(TrackPushRecordDtoToEvent)
|
||||
.Select(x => Encoding.UTF8.GetBytes(x.ToJson()));
|
||||
var log = list.First();
|
||||
_rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackPushExchange, $"TrackPush.{log.SubSystemCode}", datas, prop => { prop.Persistent = true; });
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@ -82,7 +96,24 @@ public class BroadcastingService : IBroadcastingService
|
||||
/// <returns></returns>
|
||||
public void TrackPushCompletedPublicTopic(TrackPushRecordDto log)
|
||||
{
|
||||
var data = new TrackPushCompletedEvent
|
||||
var data = TrackPushRecordDtoToEvent(log);
|
||||
_rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackPushExchange, $"TrackPush.{log.SubSystemCode}",
|
||||
Encoding.UTF8.GetBytes(data.ToJson()), prop => { prop.Persistent = true; });
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 轨迹推送失败广播
|
||||
/// </summary>
|
||||
/// <param name="log"></param>
|
||||
public void TrackPushFailedPubilcTopic(TrackPushRecordDto log)
|
||||
{
|
||||
//暂时使用推送完成模型，后续可能会出现有区分度的模型
|
||||
TrackPushCompletedPublicTopic(log);
|
||||
}
|
||||
|
||||
private static TrackPushCompletedEvent TrackPushRecordDtoToEvent(TrackPushRecordDto log)
|
||||
{
|
||||
return new TrackPushCompletedEvent
|
||||
{
|
||||
RequestId = log.RequestId,
|
||||
IsSuccess = log.IsSuccess,
|
||||
@ -101,14 +132,5 @@ public class BroadcastingService : IBroadcastingService
|
||||
TrackCode = log.TrackCode,
|
||||
Extra = log.Extra
|
||||
};
|
||||
|
||||
_rabbitClientBus.Publish(GlobalConst.TrackPushExchange, $"TrackPush.{log.SubSystemCode}",
|
||||
Encoding.UTF8.GetBytes(data.ToJson()));
|
||||
}
|
||||
|
||||
public void TrackPushFailedPubilcTopic(TrackPushRecordDto log)
|
||||
{
|
||||
//暂时使用推送完成模型，后续可能会出现有区分度的模型
|
||||
TrackPushCompletedPublicTopic(log);
|
||||
}
|
||||
}
|
||||
|
||||
@ -139,7 +139,7 @@ public class SjzyTrackApplicationModule : AbpModule
|
||||
Console.WriteLine("-- 当前为VS本地环境，用内存存储和内存队列");
|
||||
}
|
||||
|
||||
context.Services.AddRabbitBoot(options =>
|
||||
Action<RabbitOptions> action = options =>
|
||||
{
|
||||
options.ConnectionString =
|
||||
$"amqp://{configuration["RabbitMQ:UserName"]}:{configuration["RabbitMQ:Password"]}@{configuration["RabbitMQ:HostName"]}:{configuration["RabbitMQ:Port"]}";
|
||||
@ -178,7 +178,10 @@ public class SjzyTrackApplicationModule : AbpModule
|
||||
}
|
||||
}
|
||||
};
|
||||
}, typeof(SjzyTrackApplicationModule).Assembly);
|
||||
};
|
||||
|
||||
context.Services.Configure(action);
|
||||
context.Services.AddRabbitBoot(action, typeof(SjzyTrackApplicationModule).Assembly);
|
||||
|
||||
context.Services.Configure<TrackAppOptions>(configuration.GetSection("TrackApp"));
|
||||
context.Services.Configure<InterfaceRelayParamsConfigOptions>(
|
||||
|
||||
Loading…
Reference in New Issue
Block a user