Kafka的主题叫“webmessages”,分为两部分。
我们有两个用户组在同一台服务器上,但在iis的不同站点上。
其中一个使用者无法接收消息。另一个错过了大部分信息。
当我在本地计算机上编写simple consumer时,我也错过了一些消息。知道怎么回事吗?
以下是生产商代码:
_producerConfig = new ProducerConfig {
BootstrapServers = _addressWithPort,
Acks = Acks.All
};
using (var p = new ProducerBuilder<string, string>(_producerConfig).Build())
{
p.ProduceAsync(_topicName, new Message<string, string>
{
Key = ldtoKafkaMessage.Key,
Value = ldtoKafkaMessage.Message
}).ContinueWith(task =>
{
if (task.IsFaulted)
{
TraceController.TraceError(Common.Enums.TraceEventCategories.X, "Key|Message", ldtoKafkaMessage.Key + " " + ldtoKafkaMessage.Message + " " + task.Exception.Message + " " + task.Exception.InnerException+ " " + task.Exception.StackTrace);
}
else
{
TraceController.TraceInformation(Common.Enums.TraceEventCategories.X, "Key|Message|Result", ldtoKafkaMessage.Key + " " + ldtoKafkaMessage.Message + " " + "Success");
}
});
}
所以我一定要给制作人发信息。
这是消费代码。
_consumerConfig = new ConsumerConfig
{
BootstrapServers = _addressWithPort,
AutoOffsetReset = AutoOffsetReset.Earliest,
GroupId = consumerGroupId
};
using (var c = new ConsumerBuilder<string, string>(_consumerConfig).Build())
{
c.Subscribe(_topicName);
CancellationTokenSource cts = new CancellationTokenSource();
try
{
while (!cts.IsCancellationRequested)
{
try
{
var cr = c.Consume(cts.Token);
LdtoKafkaMessage.Key = cr.Key;
LdtoKafkaMessage.Message = cr.Value;
this.OnMessageChanged();
}
catch (ConsumeException e)
{
Console.WriteLine($"Error occured: {e.Error.Reason}");
}
}
}
catch (OperationCanceledException)
{
// Ensure the consumer leaves the group cleanly and final offsets are committed.
c.Close();
}
}
1条答案
按热度按时间ds97pgxw1#
您的某些消息可能无法生成,因为您处理了producer,而不是等待它完成。您可以确保通过使用
await
关键字打开ProduceAsync
方法或调用Flush()
在处理生产商之前。