我正在尝试编写一些通用代码，可以处理不同类型的输入消息，应用一些工作流逻辑，并为每条消息调用回调。
在下文中， T 以及 S 可以是 Any ; 他们之间可能没有任何共同点

object GenericTest {

  trait DataTransformer[T, S] {
    def transformData(topic: String, message: T) : S
  }

  case class Operation[T, S](transformer: DataTransformer[T, S])

  class WorkflowFlowhandler[T, S] {

    //Common logic that handle bathces for any type of message: T  
    def handleBatch(messages:  List[T], dataTransformers: List[DataTransformer[T, S]]) = {

      val conditionA = true
      val conditionB = true
      //psuedo code
      messages.foreach { message =>

        dataTransformers.foreach{ transformer =>

          //For each message new instance of PartitionerHandler should be created based on some premises
          if(conditionA && conditionB) {
            val ph = PartitionHandler("All") //PartitionHandler[_ <: GenericTest.MessageA, _ <: String]
            ph.process(message, transformer) //type mismatch; found : message.type (with underlying type T) required: _1
          } else {
            val ph = PartitionHandler("A") //PartitionHandler[_ <: GenericTest.MessageA, _ <: String]
            ph.process(message, transformer)
          }
        }
      }
    }
  }

  trait PartitionHandler[T, S] {
    def process(message: T, transformer: DataTransformer[T, S])
  }

  object PartitionHandler {
    //Factory method
//Following returns `PartitionHandler[_ <: GenericTest.MessageA, _ <: String]` which seems to be a problem
    def apply(premise: String ) = premise match {
      case "All" => new AllPartitionHandler
      case "A" => new TopicAPartitionHandler
    }
  }

  class TopicAPartitionHandler extends PartitionHandler[MessageA, String] {

    override def process(message: MessageA, transformer: DataTransformer[MessageA, String]) {

    }

  }

  class AllPartitionHandler[T, S] extends PartitionHandler[T, S] {

    override def process(message: T, transformer: DataTransformer[T, S]) {

    }
  }

  case class MessageA(id: Int, name: String)

  object Execute {

    def init[T, S](messages:  List[T], dataTransformers: List[DataTransformer[T, S]]) {

      val wfhandler = new WorkflowFlowhandler[T, S]
      wfhandler.handleBatch(messages, dataTransformers)

    }
  }

}

我有吗 The Variance Problem 在这里？我需要吗 Marker Traits 让t和s继承？
我想使用通用的handlebatch方法来处理不同类型的消息。问题是工厂 apply 方法 PartitionHandler 不是创建一个我可以作为参数传递给它的示例 process 方法。
编辑
在看了一些在scala和variance示例中为泛型类创建工厂方法的其他示例之后，我想到了下面的示例，这些示例似乎可以工作，但是看起来过于复杂，特别是使用 marker traits . 有没有一种方法可以做到这一点 marker traits ? (我试着用变化、界外球和任何运气来玩）

class WorkflowFlowhandler[T <: MessageMarkerTrait,S <: ResponseMarkerTrait] {

    //Common logic that handle bathces for any type of message: T  
    def handleBatch[T <: MessageMarkerTrait, S <: ResponseMarkerTrait](messages:  List[T], dataTransformers: List[DataTransformer[T, S]]) = {

      val conditionA = true
      val conditionB = true
      //psuedo code
      messages.foreach { message =>

        dataTransformers.foreach{ transformer =>
          import PartitionHandler._
          //For each message new instance of PartitionerHandler should be created based on some premises
          if(conditionA && conditionB) {

            val ph = PartitionHandler[MessageAny, ResponseAny] 
            ph.process(message, transformer) 
          } else {
            val ph = PartitionHandler[MessageA, ResponseA] 
            ph.process(message, transformer)
          }
        }
      }
    }
  }

  trait PartitionHandler[T <: MessageMarkerTrait, S <: ResponseMarkerTrait] {
    def process[T <: MessageMarkerTrait, S <: ResponseMarkerTrait](message: T, transformer: DataTransformer[T, S]) : S
  }

  trait PartitionHandlerFactory[T <: MessageMarkerTrait, S <: ResponseMarkerTrait] {
    def make(): PartitionHandler[T, S]     
  }

  object PartitionHandler {

    def apply[T, S](premise: String ) = premise match {
      case "All" => new AllPartitionHandler()
      case "A" => new TopicAPartitionHandler()
    }

    def apply[T <: MessageMarkerTrait, S <: ResponseMarkerTrait](implicit ev: PartitionHandlerFactory[T, S]) = ev.make()

    implicit val fooImplFactory: PartitionHandlerFactory[MessageAny, ResponseAny] = new PartitionHandlerFactory[MessageAny, ResponseAny] {
      override def make(): PartitionHandler[MessageAny, ResponseAny] = new AllPartitionHandler()
    }

    implicit val anotherFooImplFactory: PartitionHandlerFactory[MessageA, ResponseA] = new PartitionHandlerFactory[MessageA, ResponseA] {
      override def make(): PartitionHandler[MessageA, ResponseA] = new TopicAPartitionHandler()
    }    
  }  

  class MessageA extends MessageMarkerTrait
  class ResponseA extends ResponseMarkerTrait

  class MessageAny extends MessageMarkerTrait
  class ResponseAny extends ResponseMarkerTrait

  class TopicAPartitionHandler[T, S] extends PartitionHandler[MessageA, ResponseA] {

    override def process[T <: MessageMarkerTrait, S <: ResponseMarkerTrait](message: T, transformer: DataTransformer[T, S]) = {
      transformer.transformData("topicA", message)
    }

  }

  class AllPartitionHandler extends PartitionHandler[MessageAny, ResponseAny] {

    override def process[T <: MessageMarkerTrait, S <: ResponseMarkerTrait](message: T, transformer: DataTransformer[T, S]) = {
      transformer.transformData("all", message)
    }
  }