java与apachecamel和kafka的集成

o8x7eapl  于 2021-06-07  发布在  Kafka
关注(0)|答案(2)|浏览(345)

我试着测试camel与kafka的集成,这里已经解释过了
下面是我的代码

public class KafkaTest {

    public static void main(String args[]) throws Exception {
        CamelContext context = new DefaultCamelContext();

        context.addRoutes(new RouteBuilder() {
            public void configure() {
                from("kafka:test?zkConnect=localhost:2181&metadataBrokerList=localhost:9092")
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        System.out.println(exchange.getIn().getBody());
                    }
                })
                .end();
            }
        });

        context.start();
        while (true) {

        }
    }
}

但是,我得到以下错误

Exception in thread "main" org.apache.camel.FailedToCreateRouteException: Failed to create route route1: Route(route1)[[From[kafka:test?zkConnect=localhost:2181&... because of Failed to resolve endpoint: kafka://test?amp%3BmetadataBrokerList=localhost%3A9092&zkConnect=localhost%3A2181 due to: Failed to resolve endpoint: kafka://test?amp%3BmetadataBrokerList=localhost%3A9092&zkConnect=localhost%3A2181 due to: There are 2 parameters that couldn't be set on the endpoint. Check the uri if the parameters are spelt correctly and that they are properties of the endpoint. 

Unknown parameters=[{metadataBrokerList=localhost:9092, zkConnect=localhost:2181}]

请提出可能遗漏的内容。

uubf1zoe

uubf1zoe1#

使用终结点类?
比如:

public static KafkaEndpoint endpoint(String host, String port, String topic, String offset, String groupId) {
        String endpointUri = "kafka://" + host + ":" + port;
        KafkaEndpoint endpoint = new DefaultCamelContext().getEndpoint(endpointUri, KafkaEndpoint.class);
        endpoint.getConfiguration().setTopic(topic);
        endpoint.getConfiguration().setKeyDeserializer("org.apache.kafka.common.serialization.StringDeserializer");
        endpoint.getConfiguration().setValueDeserializer("org.apache.kafka.common.serialization.StringDeserializer");
        endpoint.getConfiguration().setAutoOffsetReset(offset);
        endpoint.getConfiguration().setGroupId(groupId);
        return endpoint;
    }

PollingConsumer consumer = endpoint.createPollingConsumer();

new RouteBuilder() {
            public void configure() {
                from(endpoint)
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        System.out.println(exchange.getIn().getBody());
                    }
                })
                .end();
            }
        }
yvt65v4c

yvt65v4c2#

您应该使用正式文档中指定的正确参数名称。

from("kafka:localhost:9092?topic=test&zookeeperHost=localhost&zookeeperPort=2181")

在github的wiki中描述的您所引用的版本是由apache提供的,并在此后进行了一些更改。

相关问题