Netty采集数据高效写入TDengine

本文主要是介绍Netty采集数据高效写入TDengine，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

在现代数据处理应用中，高效的数据采集与存储至关重要。Netty 是一个高性能的异步事件驱动的网络应用程序框架，非常适合用于构建高效的数据采集服务。本文将介绍如何使用 Netty 搭建一个数据采集服务，并通过优化的方式将数据高效地写入 TDengine 数据库。

设计思路

我们的目标是构建一个高效的数据采集服务，该服务能够接收来自多个客户端的数据，并在数据量达到一定阈值或经过一定时间后批量写入 TDengine 数据库。为了实现这一目标，我们需要解决以下几个关键问题：

数据缓冲：需要一个高效的缓存结构来暂存接收到的数据。
批量处理：当数据量达到一定阈值或经过一定时间后，应将数据批量写入数据库。
并发控制：确保在多线程环境下数据处理的安全性。
配置动态调整：允许配置参数如批量大小和最大等待时间的动态调整。

实现过程

1. 数据缓冲

为了高效地暂存数据，我们使用 ConcurrentLinkedQueue 作为数据缓冲区。这种队列是线程安全的，并且提供了高效的插入和删除操作。

private final ConcurrentLinkedQueue<String> buffer = new ConcurrentLinkedQueue<>();

2. 批量处理

当数据达到一定数量或经过一定时间后，我们将启动一个批量插入操作。为了实现这一点，我们使用了两个主要的组件：

计数器：用于跟踪当前缓存中的数据数量。
定时任务：如果数据没有达到阈值，则设置一个定时任务来处理数据。

private final AtomicInteger counter = new AtomicInteger(0);
private ScheduledFuture<?> scheduledFuture = null;

3. 并发控制

为了确保数据处理的安全性，我们使用 ReentrantLock 来保护批量插入操作。此外，我们还使用 AtomicBoolean 来标识当前是否有线程正在进行批量插入操作。

private final ReentrantLock lock = new ReentrantLock();
private final AtomicBoolean isBatchInserting = new AtomicBoolean(false);

4. 配置动态调整

我们使用 Nacos 配置中心来动态调整批量大小和最大等待时间。这样可以在不重启服务的情况下调整这些参数。

@NacosValue(value = "${batchSize:1000}", autoRefreshed = true)
private volatile int batchSize;  // 阈值

@NacosValue(value = "${maxWaitTime:500}", autoRefreshed = true)
private volatile long maxWaitTime;  // 最大延迟时间（毫秒）

5. 核心方法

`channelRead` 方法

每当从客户端接收到一条数据时，都会调用此方法。在此方法中，我们将数据添加到缓冲区，并更新计数器。如果数据达到阈值，则立即执行批量插入。

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {String data = (String) msg;buffer.add(data);int currentCount = counter.incrementAndGet();
if (currentCount == 1) {scheduleBatchInsert();}
if (currentCount >= batchSize) {batchInsert();}
}

`scheduleBatchInsert` 方法

计数器为1的时候，我们会安排一个定时任务来预备处理数据，以保证即便数据条目没有达到设定的阈值，也会被及时批量写入数据库中。

private void scheduleBatchInsert() {scheduledFuture = scheduler.schedule(this::batchInsert, maxWaitTime, TimeUnit.MILLISECONDS);
}

`batchInsert` 方法

此方法负责实际的批量插入操作。首先，它会检查是否已经有线程正在进行批量插入。如果是，则直接返回。如果不是，则获取锁，并开始处理数据。

private void batchInsert() {if (isBatchInserting.compareAndSet(false, true)) {lock.lock();try {if (counter.get() == 0) {return;}
List<String> dataToInsert = new ArrayList<>();while (!buffer.isEmpty()) {String data = buffer.poll();if (data != null) {dataToInsert.add(data);}}
counter.set(0);
if (scheduledFuture != null && !scheduledFuture.isDone()) {scheduledFuture.cancel(false);}
if (!dataToInsert.isEmpty()) {try {insertIntoTDengine(dataToInsert);} catch (Exception e) {logger.error("Failed to insert data into TDengine", e);}}} finally {lock.unlock();isBatchInserting.set(false);  // 设置标志位为 false}}
}

`insertIntoTDengine` 方法

此方法实现了将数据写入 TDengine 的逻辑。具体实现取决于 TDengine 的 API 或者使用的 ORM 框架。

private void insertIntoTDengine(List<String> dataToInsert) {// 实现使用 MyBatisPlus 写入 TDengine 的逻辑，可以参照https://blog.csdn.net/qq_47741012/article/details/141181396
}

6. 生命周期管理

为了确保服务的健壮性，我们需要处理通道关闭和异常捕获事件。此外，还需要提供关闭服务的方法来释放资源。

// 客户端断开连接
@Override
public void channelInactive(ChannelHandlerContext ctx) {cancelScheduledTask();super.channelInactive(ctx);
}

Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {logger.error("Exception caught in DataCollectionHandler", cause);ctx.close();
}

public void cancelScheduledTask() {if (scheduledFuture != null && !scheduledFuture.isCancelled()) {scheduledFuture.cancel(false);}
}