基于Scala泛型与Trait实现可扩展消息队列消费者的设计咨询

Got it, let's build this out in Scala with extensibility front of mind—starting with Kafka, but structured so adding other brokers later is a breeze. Here's a step-by-step implementation:

1. Define the Base Record Hierarchy #

First, we'll set up the sealed abstract class for consumer records (this enforces a closed hierarchy for now, making pattern matching safe later) and a Kafka-specific case class to hold broker-specific metadata:

import scala.collection.concurrent.TrieMap

// Sealed abstract class to standardize consumer records across brokers
sealed abstract class NotificationConsumerRecords {
  // Add common fields all record types should have here
  def timestamp: Long
}

// Kafka-specific implementation with broker-specific details
case class KafkaNotificationRecords(
  override val timestamp: Long,
  offset: Long,
  partition: Int,
  messageContent: String
) extends NotificationConsumerRecords

2. Create the Extensible Consumer Trait #

This trait defines the core contract all queue consumers must follow, using a type parameter to support different record implementations:

trait MessagingQueueConsumer[B <: NotificationConsumerRecords] {
  /**
   * Consumes messages from a topic, filtered to specific users
   * @param targetTopic The topic to pull messages from
   * @param targetUsers List of usernames to filter messages for
   * @return Thread-safe map linking usernames to their respective consumer records
   */
  def consume(targetTopic: String, targetUsers: List[String]): TrieMap[String, B]
}

3. Implement the Kafka Consumer #

Now let's build the Kafka-specific consumer that adheres to our trait. I'll include a production-ready skeleton (you'll want to tweak configs and message parsing to match your setup):

import org.apache.kafka.clients.consumer.KafkaConsumer
import java.util.Properties
import scala.jdk.CollectionConverters._

class KafkaQueueConsumer extends MessagingQueueConsumer[KafkaNotificationRecords] {
  // Initialize Kafka consumer with your cluster configs
  private val kafkaConsumer: KafkaConsumer[String, String] = {
    val props = new Properties()
    props.put("bootstrap.servers", "localhost:9092") // Replace with your broker addresses
    props.put("group.id", "notification-service-consumer")
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.put("auto.offset.reset", "earliest")
    new KafkaConsumer[String, String](props)
  }

  override def consume(targetTopic: String, targetUsers: List[String]): TrieMap[String, KafkaNotificationRecords] = {
    kafkaConsumer.subscribe(List(targetTopic).asJava)
    val polledRecords = kafkaConsumer.poll(java.time.Duration.ofMillis(1000))
    val userRecordMap = TrieMap.empty[String, KafkaNotificationRecords]

    // Process each record and filter for target users
    polledRecords.asScala.foreach { record =>
      // Replace this with your actual logic to extract username from the message
      val username = extractUsernameFromMessage(record.value())
      
      if (targetUsers.contains(username)) {
        val kafkaRecord = KafkaNotificationRecords(
          timestamp = record.timestamp(),
          offset = record.offset(),
          partition = record.partition(),
          messageContent = record.value()
        )
        userRecordMap.put(username, kafkaRecord)
      }
    }

    userRecordMap
  }

  // Helper: Parse username from message (customize for your message format, e.g., JSON)
  private def extractUsernameFromMessage(message: String): String = {
    // Dummy example—replace with your actual parsing logic
    message.split("user:")(1).split(",")(0).trim
  }
}

4. Usage Example #

Here's how you'd use this consumer in your application:

object ConsumerDemo extends App {
  val kafkaConsumer = new KafkaQueueConsumer()
  val userNotifications = kafkaConsumer.consume("user-alerts", List("jane_doe", "john_smith"))
  
  userNotifications.foreach { case (user, records) =>
    println(s"New alert for $user: ${records.messageContent} (Offset: ${records.offset})")
  }
}

Extensibility Tips #

• Adding a new broker (like RabbitMQ) is simple: create a new case class extending NotificationConsumerRecords (e.g., RabbitMQNotificationRecords) and implement MessagingQueueConsumer with broker-specific logic.
• The sealed abstract class ensures all record implementations are known at compile time, which makes future pattern matching operations (if needed) type-safe.
• Using TrieMap gives you thread-safe concurrent access out of the box, which is handy if you're processing records across multiple threads.

内容的提问来源于stack exchange，提问作者S.K