Apache Camel集成InfluxDB 2.x：构建自定义组件教程

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <!-- 建议替换为自定义的父POM，此处沿用Camel组件的父POM作为起点 -->
    <parent>
        <groupId>org.apache.camel</groupId>
        <artifactId>components</artifactId>
        <version>3.19.0</version> <!-- 根据你使用的Camel版本调整 -->
    </parent>

    <groupId>my.group.name</groupId>
    <artifactId>camel-influxdb2</artifactId>
    <version>3.19.0</version> <!-- 与Camel父版本保持一致或自定义 -->
    <packaging>jar</packaging>
    <name>Camel :: InfluxDB 2.x Client</name>
    <description>A Custom Camel Component for InfluxDB 2.x</description>
    <url>https://www.example.com</url>

    <properties>
        <version.influx-java-driver>2.7.0</version.influx-java-driver> <!-- InfluxDB 2.x Java客户端版本 -->
    </properties>

    <dependencies>
        <!-- Camel核心支持库，所有自定义组件都需依赖 -->
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-support</artifactId>
        </dependency>

        <!-- InfluxDB 2.x 官方Java客户端驱动 -->
        <dependency>
            <groupId>com.influxdb</groupId>
            <artifactId>influxdb-client-java</artifactId>
            <version>${version.influx-java-driver}</version>
            <exclusions>
                <!-- 排除可能与项目其他依赖冲突的okhttp3日志拦截器 -->
                <exclusion>
                    <groupId>com.squareup.okhttp3</groupId>
                    <artifactId>logging-interceptor</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <!-- 测试依赖 -->
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-test-junit5</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.mockito</groupId>
            <artifactId>mockito-core</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- 其他测试或工具依赖 -->
    </dependencies>
</project>

package my.group.name.camel.influxdb2;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.write.WriteApi;
import com.influxdb.client.write.Point;
import org.apache.camel.Exchange;
import org.apache.camel.support.DefaultProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Influx2DbProducer extends DefaultProducer {

    private static final Logger LOG = LoggerFactory.getLogger(Influx2DbProducer.class);
    private final Influx2DbEndpoint endpoint;
    private InfluxDBClient influxDBClient;
    private WriteApi writeApi;

    public Influx2DbProducer(Influx2DbEndpoint endpoint) {
        super(endpoint);
        this.endpoint = endpoint;
    }

    @Override
    protected void doStart() throws Exception {
        super.doStart();
        // 初始化 InfluxDB 2.x 客户端
        this.influxDBClient = endpoint.getInfluxDBClient(); // 从Endpoint获取配置好的客户端
        this.writeApi = influxDBClient.getWriteApi();
        LOG.info("Influx2DbProducer started, connected to InfluxDB 2.x at {}", endpoint.getUrl());
    }

    @Override
    public void process(Exchange exchange) throws Exception {
        Object body = exchange.getIn().getBody();

        if (body instanceof Point) {
            // 处理单个Point写入
            Point point = (Point) body;
            writeApi.writePoint(endpoint.getBucket(), endpoint.getOrg(), point);
            LOG.debug("Wrote single point to InfluxDB 2.x bucket: {}", endpoint.getBucket());
        } else if (body instanceof Iterable) {
            // 处理Point列表批量写入
            Iterable<Point> points = (Iterable<Point>) body;
            writeApi.writePoints(endpoint.getBucket(), endpoint.getOrg(), points);
            LOG.debug("Wrote multiple points to InfluxDB 2.x bucket: {}", endpoint.getBucket());
        } else {
            // 处理其他类型的数据，例如可以从Map或JSON构建Point
            // ...
            LOG.warn("Unsupported message body type for InfluxDB 2.x: {}", body.getClass().getName());
        }
    }

    @Override
    protected void doStop() throws Exception {
        if (writeApi != null) {
            writeApi.close();
        }
        if (influxDBClient != null) {
            influxDBClient.close();
        }
        super.doStop();
        LOG.info("Influx2DbProducer stopped.");
    }
}

class=my.group.name.camel.influxdb2.Influx2DbComponent