我正在使用 Scala 和 Kafka 創建基于主題的發布-訂閱體系結構。我的問題是如何使用 Kafka 主題處理我的應用程式的一對一訊息傳遞部分？這是我的制作人課：
class Producer(topic: String, key: String, brokers: String, message: String) {
val producer = new KafkaProducer[String, String](configuration)
private def configuration: Properties = {
val props = new Properties()
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
props.put(ProducerConfig.ACKS_CONFIG, "all")
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getCanonicalName)
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getCanonicalName)
props
}
def sendMessages(): Unit = {
val record = new ProducerRecord[String, String](topic, key, message)
producer.send(record)
producer.close()
}
}
這是我的消費類：
class Consumer(brokers: String, topic: String, groupId: String) {
val consumer = new KafkaConsumer[String, String](configuration)
consumer.subscribe(util.Arrays.asList(topic))
private def configuration: Properties = {
val props = new Properties()
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer].getCanonicalName)
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer].getCanonicalName)
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
//props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true)
props
}
def receiveMessages(): Unit = {
while (true) {
consumer.poll(Duration.ofSeconds(0)).forEach(record => println(s"Received message: $record"))
}
}
}
我還有一個身份驗證服務，它負責處理與通過 JWT 令牌進行身份驗證相關的所有事情。我對如何為特定用戶創建訊息感到困惑，我考慮過創建一個“訊息”類，但是當談到如何發送這些“特定”用戶訊息以及如何在 kafka 上對這些訊息進行磁區以供以后使用時，我迷失了：
class Message {
def sendMessage(sender_id: String, receiver_id: String, content: String): Unit ={
val newMessage = new Producer(brokers = KAFKA_BROKER,key =sender_id " to " receiver_id, topic = "topic_1", message = content)
newMessage.sendMessages()
}
def loadMessage(): Unit ={
//
}
}
My thought was to specify a custom key for all messages belonging to the same conversation but I couldn't find the right way to retrieve these messages later on as my consumer returns everything contained in that topic no matter what the key is. Meaning, all the users will eventually get all the messages. I know my modeling seems messy but I couldn't find the right way to do it, I'm also kinda confused when it comes to the usage of the group_id in the consumer. Could someone make me what's the right way to achieve what I'm trying to do here please ?
uj5u.com熱心網友回復：
以后找不到正確的方法來檢索這些訊息...消費者回傳該主題中包含的所有內容，無論密鑰是什么
您需要.assign將 Consumer 實體指向特定磁區，而不是 use .subscribe，它會讀取所有 partitions。或者你會為每個對話使用特定的主題。
但是，您需要為每個將存在的對話提供唯一的磁區/主題。在用戶隨機創建/洗掉房間的常規聊天應用程式中，這不適用于 Kafka。
最終，我建議將您的資料寫入 Kafka 以外的其他地方，您實際上可以在“convertsationId”和/或用戶 ID 上查詢和索引，而不是嘗試將這些事件直接從 Kafka 轉發到您的“聊天”應用程式中。
轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/453085.html