Я новичок в Kafka и каким-то образом могу запускать kafka Avro для потребителей и производителей.Продюсер производит сообщение, и я успешно получаю его в потребителе.Вот мой фрагмент кода производителя:
static async void AvroProducer()
{
string bootstrapServers = "localhost:9092";
string schemaRegistryUrl = "Production163:8081";
string topicName = "player";
string groupName = "avro-generic-example-group";
var s = (RecordSchema)RecordSchema.Parse(
@"{
""namespace"": ""Confluent.Kafka.Examples.AvroSpecific"",
""type"": ""record"",
""name"": ""User"",
""fields"": [
{""name"": ""name"", ""type"": ""string""},
{""name"": ""favorite_number"", ""type"": [""int"", ""null""]},
{""name"": ""favorite_color"", ""type"": [""string"", ""null""]}
]
}"
);
using (var schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { SchemaRegistryUrl = schemaRegistryUrl }))
using (var producer =
new ProducerBuilder<string, GenericRecord>(new ProducerConfig { BootstrapServers = bootstrapServers })
.SetKeySerializer(new AsyncAvroSerializer<string>(schemaRegistry))
.SetValueSerializer(new AsyncAvroSerializer<GenericRecord>(schemaRegistry))
.Build())
{
Console.WriteLine($"{producer.Name} producing on {topicName}. Enter user names, q to exit.");
int i = 0;
string text;
while ((text = Console.ReadLine()) != "q")
{
var record = new GenericRecord(s);
record.Add("name", text);
record.Add("favorite_number", i++);
record.Add("favorite_color", "blue");
producer.ProduceAsync(topicName, new Message<string, GenericRecord> { Key = text, Value = record })
.ContinueWith(task => task.IsFaulted
? $"error producing message: {task.Exception.Message}"
: $"produced to: {task.Result.TopicPartitionOffset}");
}
}
Console.ReadLine();
}
Как вы можете видеть в приведенном выше коде, я использую схему записи, но я пробую эту схему:
//this is the new schema try
var s = (RecordSchema)RecordSchema.Parse(
@"{
""type"": ""record"",
""name"": ""TestingMsg"",
""doc"": ""Sample"",
""fields"": [
{
""name"": ""key"",
""type"": ""string""
},
{
""name"": ""Time"",
""type"": ""long""
},
{
""name"": ""sourceSeconds"",
""type"": ""long""
},
{
""name"": ""serverT"",
""type"": ""long""
},
{
""name"": ""statusCode"",
""type"": ""int""
}
]
}"
);
Новая, котораяЯ пытаюсь использовать, но это не работает, так как я не получаю сообщения от потребителя.А вот и потребитель:
void KafkaReader(CancellationToken cancellationToken)
{
Debug.Log("kafka reader started. . .");
// Set up your Kafka connection here.
while (_keepThreadRunning)
{
using (CachedSchemaRegistryClient schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { SchemaRegistryUrl = schemaRegistryUrl }))
using (IConsumer<string, GenericRecord> consumer = new ConsumerBuilder<string, GenericRecord>(new ConsumerConfig { BootstrapServers = bootstrapServers, GroupId = groupName })
//using (IConsumer<string, GenericRecord> consumer = new ConsumerBuilder<string, GenericRecord>(new ConsumerConfig { BootstrapServers = bootstrapServers})
.SetKeyDeserializer(new AsyncAvroDeserializer<string>(schemaRegistry).AsSyncOverAsync())
.SetValueDeserializer(new AsyncAvroDeserializer<GenericRecord>(schemaRegistry).AsSyncOverAsync())
.SetErrorHandler((_, e) => Debug.Log($"Error: {e.Reason}"))
.Build())
{
Debug.Log("subscribe" );
consumer.Subscribe(topicName);
while (true)
{
ConsumeResult<string, GenericRecord> consumeResult = consumer.Consume(cancellationToken);//TimeSpan.FromMilliseconds(50000)//new TimeSpan(0,0,1)
_stringsReceived.Enqueue(consumeResult.Value.ToString());
if (consumeResult != null)
{
Debug.Log($"Key: {consumeResult.Message.Key}\nValue: {consumeResult.Value}");
}
else
{
Debug.Log("consumer Result is null");
}
//yield return new WaitForSeconds(1);
}
}
}
GetComponent<KafkaServerConfigUI>().KafkaDisconnected();
// Disconnect and clean up your connection here.
}
Помните, я просто запускаю реестр Apache Kafka по умолчанию с помощью командного файла.
D:\ApachKafka\confluent\confluent-5.2.1\bin\windows\schema-registry-start.bat D:\ApachKafka\confluent\confluent-5.2.1\etc\schema-registry\schema-registry.properties
Что я делаю не так?Нужно ли регистрировать схему где-нибудь?