88 lines
3.3 KiB
C#
88 lines
3.3 KiB
C#
|
using System.Text;
|
|||
|
|
using Common;
|
|||
|
|
using CommonEntity;
|
|||
|
|
using CommonEntity.RCUEntity;
|
|||
|
|
using CommonTools;
|
|||
|
|
using Confluent.Kafka;
|
|||
|
|
using IOT_JieKou;
|
|||
|
|
using NLog;
|
|||
|
|
using NPOI.OpenXml4Net.OPC;
|
|||
|
|
|
|||
|
|
namespace BLWLogServer.Services
|
|||
|
|
{
|
|||
|
|
public class KafkaConsume : BackgroundService
|
|||
|
|
{
|
|||
|
|
public IConfiguration Configuration { get; set; }
|
|||
|
|
public IClusterClient clusterClient { get; set; }
|
|||
|
|
public KafkaConsume(IConfiguration configuration,IClusterClient ccc)
|
|||
|
|
{
|
|||
|
|
Configuration = configuration;
|
|||
|
|
this.clusterClient = ccc;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
public static Logger logger = NLog.LogManager.GetCurrentClassLogger();
|
|||
|
|
|
|||
|
|
protected async override Task ExecuteAsync(CancellationToken stoppingToken)
|
|||
|
|
{
|
|||
|
|
await Task.Factory.StartNew(async () =>
|
|||
|
|
{
|
|||
|
|
var consumers = new List<Task>();
|
|||
|
|
for (int i = 0; i < 3; i++)
|
|||
|
|
{
|
|||
|
|
consumers.Add(Task.Run(() => StartConsumer(stoppingToken)));
|
|||
|
|
}
|
|||
|
|
await Task.WhenAll(consumers);
|
|||
|
|
}, TaskCreationOptions.LongRunning);
|
|||
|
|
}
|
|||
|
|
private async Task StartConsumer(CancellationToken stoppingToken)
|
|||
|
|
{
|
|||
|
|
string? ipport = Configuration["Kafka:EndPoint"];
|
|||
|
|
string? user = Configuration["Kafka:UserName"];
|
|||
|
|
string? pwd = Configuration["Kafka:PassWord"];
|
|||
|
|
string? mongodbconnectstr = Configuration["Mongodb:Connectstr"];
|
|||
|
|
while (!stoppingToken.IsCancellationRequested)
|
|||
|
|
{
|
|||
|
|
var config = new ConsumerConfig
|
|||
|
|
{
|
|||
|
|
GroupId = "blwlogserver-consumer-group",
|
|||
|
|
AutoOffsetReset = AutoOffsetReset.Earliest,
|
|||
|
|
BootstrapServers = ipport,
|
|||
|
|
SecurityProtocol = SecurityProtocol.SaslPlaintext,
|
|||
|
|
SaslMechanism = SaslMechanism.Plain,
|
|||
|
|
SaslUsername = user,
|
|||
|
|
SaslPassword = pwd
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
var c = new ConsumerBuilder<string, byte[]>(config).Build();
|
|||
|
|
c.Subscribe(KafkaKey.New_RCU_Data_Topic);
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
while (true)
|
|||
|
|
{
|
|||
|
|
var cr = c.Consume(stoppingToken);
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
var k = cr.Message.Key;
|
|||
|
|
var v = cr.Message.Value;
|
|||
|
|
if (k.Equals(KafkaKey.UDPckageAllUDPDataKey))
|
|||
|
|
{
|
|||
|
|
RCU_UDPData_With_String UDPPPP = MyMessagePacker.FastDeserialize<RCU_UDPData_With_String>(v);
|
|||
|
|
string? endpoint1= UDPPPP.endpoint;
|
|||
|
|
var cl1= clusterClient.GetGrain<IDataTran>(endpoint1,GrainPrefix.RCU_Grain_Prefix);
|
|||
|
|
await cl1.TongXin(UDPPPP);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
catch (Exception ex)
|
|||
|
|
{
|
|||
|
|
logger.Error("Ex出错：" + ex.Message);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
catch (Exception ex)
|
|||
|
|
{
|
|||
|
|
logger.Error("未知错误" + ex.Message);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|