springboot整合camel,kafka

野性酷女 2022-05-09 14:14 328阅读 0赞

camel 就是定义路由

## maven依赖 ##

<dependency>
                <groupId>org.apache.camel</groupId>
                <artifactId>camel-spring-boot</artifactId>
                <version>2.15.1</version>
            </dependency>
            <dependency>
                <groupId>org.apache.camel</groupId>
                <artifactId>camel-core</artifactId>
                <version>2.19.2</version>
            </dependency>
            <dependency>
                <groupId>org.apache.camel</groupId>
                <artifactId>camel-kafka</artifactId>
                <version>2.19.2</version>
            </dependency>
            
            
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
                <version>2.1.9.RELEASE</version>
            </dependency>

## 代码部分 ##

### 配置文件 ###

# camel 配置
    kafka:
      server : localhost
      port : 9092
      topic : part
      # 定义的消费组
      channel : 8
    auto.offset.reset: earliest
    
    spring:
      kafka:
        bootstrap-servers: 127.0.0.1:9092
        producer:
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        consumer:
          # 应该可以消费者类指定
          group-id: 9
          enable-auto-commit: true
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

**有没有多余的配置，自己去测试了，LZ是配置文件都写在一起，博客是单写的，就没有去测试拆分了**

### 路由代码 ###

import org.apache.camel.Exchange;
    import org.apache.camel.Processor;
    import org.apache.camel.builder.RouteBuilder;
    import org.springframework.stereotype.Component;
    
    /** * @author lf * @描述 * @date 2018-09-07 */
    @Component
    public class MainRouter  extends RouteBuilder { 
    
        @Override
        public void configure() throws Exception { 
    
            from("kafka:{ {kafka.topic}}?brokers={ {kafka.server}}:{ {kafka.port}}&groupId={ {kafka.channel}}&autoOffsetReset=earliest")
                    .process(new Processor() { 
                        public void process(Exchange exchange) throws Exception { 
                            System.out.println("Message Body : " + exchange.getIn().getBody());
                        }
                    });
    
        }
    }

官网 [http://camel.apache.org/kafka.html][http_camel.apache.org_kafka.html]

[http_camel.apache.org_kafka.html]: http://camel.apache.org/kafka.html