使用Apache camel将MyObject传递到bean路由的可能性？

Question

我正在使用Apache camel JAVA DSL使用来自Apache kafka的消息。我通过在Kafka上将对象转换为byte[]来编写对象。当我使用它时，我会收到一条带有字节[]的消息。我将其反序列化并获得一个对象。

我检查它是否是MyObject的对象，然后需要使用java DSL. to（）将其传递给bean。我的代码如下：

public class KafkaRouter extends RouteBuilder {
    
    private MessageBean msgBean;
    
    @Override
    public void configure() throws Exception {
        
        from("{{kafka.cons.uri}}").process(new Processor() {
            
            
            
            @Override
            public void process(Exchange exchange) throws Exception {
                Object obj = SerializationUtils.deserialize(exchange.getIn().getBody(byte[].class));                //TODO cast to specific class as returned after deserialization.
                
                if(obj !=null && obj instanceof MessageBean){
                    
                    msgBean = (MessageBean)obj;
                    
                }
                else {
                    
                    throw new PTFException("Invalid Message read in Kafka Consumer");
                }
                
            }
            
            
        }).bean(PTFTransformerService.class,"callTransformerService(msgBean)"); ;
    }

现在的问题是我只想MyObject在相应的调用方法，我不想使用TypeConvertors。我不想得到Exchange/body in方法，我将在进程中处理我的流并抛出异常，如果读取无效消息，并且不将其转发到bean。

另一方面，我的方法是：

private void callTransformerService(MessageBean msgObj){
    // Got my object here ;-)   
        
    }

Answer 1

您可以在处理器中设置Exchange主体，如下所示：

msgBean = (MessageBean)obj;
exchange.getIn().setBody(msgBean, MessageBean.class);
void callTransformerService(@Body MessageBean msgObj) {}

Answer 2

您可能需要编写一个自定义的回退类型转换器，可以将kafka字节[]转换为POJO

• http://camel.apache.org/type-converter.html

然后，您可以在bean中定义pojo类型，Camel将使用回退类型转换器尝试转换为pojo类型。

Answer 3

在函数参数前添加@Body：

import org.apache.camel.Body;

private void callTransformerService(@Body MessageBean msgObj){

}