Groovy如何处理可选参数？Kafka消费者功能扩展咨询

原Kafka消费者类代码 #

package kafka

@Grab(group='org.apache.kafka', module='kafka-clients', version='2.8.0')
@Grab(group='org.slf4j', module='slf4j-simple', version='2.0.0')

import com.dell.techops.TechopsKafkaProducer

import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import java.time.Duration
import groovy.util.logging.Slf4j


class KConsumer {
  def consumer

  KConsumer(String NexusServers, String topic, String groupId) {
    def props = new Properties()
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())

    consumer = new KafkaConsumer<String, String>(props)
    consumer.subscribe([topic])
  }
  
  List<String> consumeMessages() {
    def messages = []
    while (true) {
      def records = consumer.poll(100)
      if (!records.isEmpty()) {
        records.each { record ->
          messages.add(record.value())
        }
        return messages
      }
    }
  }
 void close() {
   consumer.close()
  }
}  

现有使用方式 #

def servers = "localhost:9092"
def topic = "my-topic"
def groupId = "my-group"

def consumer = new KConsumer(servers, topic, groupId)
messages = consumer.consumeMessages()
consumer.close()

需求说明 #

目前功能正常，希望添加一个可选参数key，当传入该参数时，仅返回对应key的消息，对应逻辑如下：

def messages = []
def records = consumer.poll(100)
for (ConsumerRecord<String, String> record : records) {
    if (record.key() == key) {
        messages.add(record.value())
    }
}

问题解答 #

1. 是否必须将类默认设置为接收4个参数？ #

不需要。Groovy提供了多种灵活处理可选参数的方式，完全可以在兼容原有3个参数调用方式的前提下，支持传入可选的key参数，无需强制修改构造函数的参数数量。

2. Groovy中是否有类似Python的*args的方式处理可选参数？ #

有，Groovy支持多种可选参数实现方式：

• 默认参数值：给参数设置默认值（比如null），调用时可传可不传
• 命名参数：通过Map接收可选参数，调用时用键值对传入，扩展性更强
• 可变参数：用...定义可变参数列表，类似Python的*args，不过针对单个可选参数，默认参数或命名参数的可读性更好

3. 是否需要新增函数来实现该功能，还是可以复用现有的consumeMessages函数？ #

完全可以复用现有的consumeMessages函数，只需要给它添加可选参数即可，这样能保持API的简洁性，避免冗余函数。

优化后的代码示例 #

方式一：给consumeMessages添加默认参数 #

修改consumeMessages方法，添加可选的key参数并设默认值为null，在逻辑中判断是否需要过滤：

List<String> consumeMessages(String key = null) {
    def messages = []
    while (true) {
        def records = consumer.poll(100)
        if (!records.isEmpty()) {
            records.each { record ->
                // key为null时返回所有消息，否则只返回匹配key的消息
                if (key == null || record.key() == key) {
                    messages.add(record.value())
                }
            }
            return messages
        }
    }
}

调用方式：

// 原有调用方式不变，返回所有消息
messages = consumer.consumeMessages()

// 传入key参数，仅返回对应key的消息
messages = consumer.consumeMessages("target-key")

方式二：使用命名参数（适合后续扩展更多参数） #

如果之后可能添加更多可选参数，用命名参数更清晰：

List<String> consumeMessages(Map opts = [:]) {
    def targetKey = opts.key
    def messages = []
    while (true) {
        def records = consumer.poll(100)
        if (!records.isEmpty()) {
            records.each { record ->
                if (targetKey == null || record.key() == targetKey) {
                    messages.add(record.value())
                }
            }
            return messages
        }
    }
}

调用方式：

// 原有调用方式不变
messages = consumer.consumeMessages()

// 传入命名参数
messages = consumer.consumeMessages(key: "target-key")

另外注意：原构造函数中参数名NexusServers和实际使用的bootstrapServers不匹配，建议修正为一致，避免运行时错误：

KConsumer(String bootstrapServers, String topic, String groupId) {
    def props = new Properties()
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
    // 其他代码不变
}

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