2024-08-08 10:40:36
在漫长的折腾生涯中，玩过太多东西，但是大浪淘沙，最终剩下来常用的只有那么几款。 在这个系列中，准备介绍一些非常有意思的开源产品。 n8n 是一个开源免费的自动化工作流平台，它提供了 200多个不同的节点来自动化工作流程，通过编排，可以实现跨不同服务的自动化流程。 相比市面上已有的 Zapier 等其他商业化自动化工具，n8n稍显稚嫩。但是免费开源以及强大的拓展性和自定义能力使得n8n在这个领域也是独树一帜。 部署 n8n 分为两个版本： 云服务版本： 官方提供的云服务版本，到官网上注册账号使用，需要付费 自托管版本： n8n项目本身有开源版本，可以自己编译源码部署，也可以通过官方提供的Docker镜像，使用 Docker 部署 推荐使用 Docker 进行部署，简单快捷省心 docker volume create n8n_data docker run -it --rm --name n8n -p 5678:5678 -v n8n_data:/home/node/.n8n docker.n8n.io/n8nio/n8n 在默认情况下，n8n会使用 SQLite 数据库进行数据存储，我们可以通过配置环境变量，把存储切换到 PostgresDB docker volume create n8n_data docker run -it --rm \ --name n8n \ -p 5678:5678 \ -e DB_TYPE=postgresdb \ -e DB_POSTGRESDB_DATABASE=<POSTGRES_DATABASE> \ -e DB_POSTGRESDB_HOST=<POSTGRES_HOST> \ -e DB_POSTGRESDB_PORT=<POSTGRES_PORT> \ -e DB_POSTGRESDB_USER=<POSTGRES_USER> \ -e DB_POSTGRESDB_SCHEMA=<POSTGRES_SCHEMA> \ -e DB_POSTGRESDB_PASSWORD=<POSTGRES_PASSWORD> \ -v n8n_data:/home/node/.
2024-05-14 09:58:26
在我的日常工作中，有很大精力投入到数据采集上。我需要从 syslog 采集大量数据，通常的流程是，将每条数据进行校验之后解析为对象进行一系列的处理与分析。这会产生大量对象，在 Java 中，大量对象必然意味着大量堆内存和频繁的 GC。为提高对象利用率，降低 GC 压力，我们基于对象池技术进行了一些优化手段。 一、为什么需要对象池 在数据采集系统中，每秒钟可能处理成千上万条日志记录，每条记录都需要转换为对象。频繁的对象创建和销毁会导致较高的性能开销，尤其是增加垃圾回收（GC）的频率，从而影响系统的整体性能。对象池通过复用对象减少创建和销毁的次数，提升性能和资源利用率。 二、对象池的原理 在 Java 中，说到池，我们通常会想到连接池、线程池。实际上，所有的池都是为了解决同一个问题：降低资源重复创建和销毁的频率。 对象池的工作机制与线程池和连接池相似。对象池通过维护一定数量的对象，当需要使用时从池中取出，使用完毕后再归还池中，避免了频繁的对象创建和销毁，显著减少了 GC 的负担。基本原理如下： 预创建对象：在初始化时，预先创建一组对象或线程，放入池中备用。 获取和归还：需要时从池中取出，使用完毕后归还池中。 复用机制：通过复用已有的对象或线程，避免频繁创建和销毁，提升系统性能。 三、自定义对象池的核心实现 以下是一个自定义对象池在数据采集场景中的实战示例代码： import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; public class ObjectPool<T> { private BlockingQueue<T> pool; private int maxPoolSize; private ObjectFactory<T> factory; public ObjectPool(int maxPoolSize, ObjectFactory<T> factory) { this.maxPoolSize = maxPoolSize; this.factory = factory; this.pool = new LinkedBlockingQueue<>(maxPoolSize); initializePool(); } private void initializePool() { for (int i = 0; i < maxPoolSize; i++) { pool.
2024-05-13 17:11:53
金三银四，最近开放简历做了一些面试。在一次面试中，就 Kafka 消息的有序性进行了一番讨论，这里贴一下相关思考。 首先贴结论： 在kafka中，多 partition 的情况下，kafka本身是无法保证消息的有序性的。但是可以通过逻辑控制保证消息的有序性。 为什么无序？ 在Apache Kafka中，一个主题（Topic）可以被分为多个分区（Partitions），这种设计是为了实现水平扩展和提高吞吐量。每个分区都是一个有序的、不可变的消息序列，新的消息不断追加到序列的末尾。 然而，当一个主题包含多个分区时，Kafka的架构确实决定了它无法全局保证消息的有序性。主要原因如下： 生产者的分区策略：生产者可以根据消息的键（Key）或者自定义的分区策略来决定将消息发送到哪个分区。如果不同的消息使用了不同的键或者被发送到不同的分区，那么这些消息之间的顺序就无法得到保证 分区间的并行性：Kafka允许消费者并行地从多个分区中读取消息。由于不同分区的消息可以被不同的消费者实例同时处理，因此这些消息的到达和处理顺序在全局范围内是无法保证的 分区间的独立性：每个分区都是独立的，它们之间没有直接的顺序关联。生产者可以将消息发送到任意一个分区，而消费者也可以独立地从每个分区中消费消息。这种独立性意味着，即使在一个分区内部消息是有序的，但在不同分区之间的消息顺序是无法控制的 可以有序吗？如何保证？ Kafka 每个分区都是一个有序的、不可变的消息序列，新的消息不断追加到序列的末尾。消费者按照消息在分区中的顺序来消费消息。因此，要保证消息的顺序处理，关键在于确保同一业务逻辑的消息发送到同一个分区。 可以通过以下方式来处理有序性需求： 单分区主题： 最简单的方法是为每个需要保证顺序的逻辑创建一个单独的Kafka主题，并设置该主题只有一个分区。这样，所有消息都会按照发送顺序被消费。但是，这种方法牺牲了Kafka的水平扩展能力。 使用相同的键： Kafka允许生产者为每条消息指定一个键（Key）。当消息被发送到Kafka时，Kafka会根据消息键的哈希值来决定将消息发送到哪个分区。因此，如果所有需要保证顺序的消息都使用相同的键，那么这些消息就会被发送到同一个分区，从而保证了顺序。 自定义分区策略： 如果默认的哈希分区策略不能满足需求，可以自定义分区策略。通过实现Partitioner接口，可以控制消息发送到哪个分区。例如，可以根据业务逻辑将属于同一顺序逻辑的消息发送到特定的分区。 消费者端顺序处理： 即使生产者保证了消息的顺序，消费者端也需要正确处理以维持顺序。消费者应该确保在处理完一条消息后，再拉取下一条消息，避免并发处理导致顺序混乱 注意事项 当使用多个消费者实例消费同一个分区时，无法保证消息的顺序处理 在保证顺序的同时，也要考虑系统的吞吐量和可用性，避免过度限制Kafka的性能
2024-04-25 10:17:04
昨天，有人跟我聊起项目中对程序的优化，有一个特别有意思的话题《如何榨干一台机器的CPU》 现在的市面上，多核CPU是主流，有了多核的加持，可以更加有效的发挥硬件的能力，基于Java程序，我们究竟该如何更加有效的应用多核的能力？我个人经验来讲，主要考虑一下几个方面： 并行执行任务 减少共享数据的写操作 采用合适的方式处理竞争资源 减少数据拷贝次数 合适的GC 接下来详细说明。 1. 并行执行任务 合理利用多线程执行任务，可以有效的发挥CPU的多核性能。由于超线程技术的存在，实际上CPU可以并行执行的线程数量通常是物理核心数量的2倍。 我们都知道，在计算机中，进程是操作系统资源（内存、显卡、磁盘）分配的最小单位。线程是CPU执行调度的最小单位。 因此，实现并行计算的方式大体上有三种：多进程、多线程、多进程+多线程。具体采用哪种方式，就需要实际情况实际分析了。整体指导方针是：如果多线程可以解决，就不要尝试引入多进程。因为每个进程之间是独立的，多进程任务难免会涉及到进程之间通信，而进程之间的协调与通信通常会比较复杂。容易为程序引入额外的复杂度，得不偿失。 2. 减少共享数据的写操作 深入到线程中，每个线程都有自己的内存空间，在这个内存中，线程可以随意进行读写。因此多线程任务中，提高效率的优化手段之一就是： 尽量避免多个线程共同操作共享资源，如果条件允许，尽量采用以空间换时间的方式，将数据复制多份保存在每个线程单独的内存空间中。 如果必须存在共享内存的操作，我们的措施通常是，尽量减少共享数据的写操作，在共享内存中，多个线程的读操作是不存在资源的竞争的。一旦涉及到写共享内存，通常会使用 volatile 关键字保证内存数据对多个线程的可见性，这种情况下就不可避免的要涉及到插入内存屏障指令，用来保证处理器对指令的执行顺序不会打乱。相比不存在内存屏障的操作，性能会有所下降。 因此，需要尽量减少多个线程对共享内存的写操作。具体的方案是： 通过业务逻辑控制，在程序设计之初，排除掉共享数据的方案 在每个线程内部创建单独的对象，互不影响 使用 ThreadLocal 生成线程的本地对象副本 3. 采用合适的方式处理竞争资源 多线程任务中，涉及到资源竞争的部分，通常都需要采用对应的措施来保证资源的一致性。常见的解决方案有两种： 对资源加线程锁 采用乐观策略实现无锁操作（CAS） 线程锁的使用： 使用线程锁来保证资源的一致性是由来已久的一种非常简单便捷的方法。这种操作可以粗暴的控制多个线程对资源的访问，所以在处理多线程资源竞争关系的时候，我们通常会优先想到加锁的方式。 为了提高执行性能，通常会采用轻量级锁来代替重量级锁，在 Java 1.5 中 synchronize 是一个重量级锁，是相对低效率的；相比之下使用 Lock 对象的性能更高一些。但是这种情况到了 Java 1.6 发生了很大的变化，由于官方对 synchronize 引入了适应自旋、锁消除、轻量级锁、偏向锁等优化手段， synchronize 与 Lock 在性能上不存在什么差距。所以如果你使用高于 Java 1.6 的版本，请放心大胆的使用 synchronize 。 无锁操作（CAS）： 对于传统的加锁操作，我们通常认为是悲观策略。相对于悲观策略，我们还有一个乐观策略可以选择。乐观策略认为不会存在资源不一致的情况，假如出现了，就再试一次。 实际上在 Java 中，一些锁的实现也利用了 CAS，体现在 Java 中的应用如下： 应用领域 示例 java.
2024-04-12 14:25:41
在我的工作中，数据处理占据了比较大的权重。在数据处理的过程中，有一项比较繁琐的工作，就是对日志中的每个字段进行单独校验和处理，校验的内容大概有以下几类： 字段数量校验 字段为空判断 字段内容校验 特殊字段校验、信息补充 增加标签字段 在这类场景中，最常规的方法就是编写冗长的 if-else 代码段进行按部就班的校验，这显然不是最佳方案。 责任链模式可以有效地解决上述繁琐的工作。通过将不同的校验和处理逻辑分配给不同的处理者，形成一条责任链，数据依次通过各个处理者进行处理。这样做的好处是： 避免编写大量的重复 if-else 代码 将每个校验逻辑进行隔离，区分责任边界 逻辑清晰，代码简洁 以下是一个简单的责任链模式代码示例： public class ChainHandler { private List<Handler> handlers; public ChainHandler() { this.handlers = new ArrayList<>(); } public void addHandler(Handler handler) { handlers.add(handler); } public boolean handle(LogEntry logEntry) { for (Handler handler : handlers) { handler.handle(logEntry); if (!handler.isComplete()) { return false; // 如果处理程序未完成，则整个处理链失败 } } return true; // 如果所有处理程序均完成，则整个处理链成功 } } // 定义责任链接口 public interface Handler { boolean isComplete(); void handle(LogEntry logEntry); } // 字段数量校验 public class FieldCountHandler implements Handler { @Override public boolean isComplete() { // 检查字段数量是否满足要求 return true; } @Override public void handle(LogEntry logEntry) { // 进行字段数量的校验和处理 } } 在上述代码中，ChainHandler 类管理责任链中的处理者。FieldCountHandler 是处理字段数量的实现。
2024-02-02 10:04:42
在电商等需要在线支付的应用中，通常需要设置订单自动取消的功能。本文将介绍几种在 Spring Boot 中实现订单 30 分钟自动取消的方案，包括定时任务、延迟队列和 Redis 过期事件。 方案一：定时任务 定时任务是一种简单且常用的实现订单自动取消的方案。在 Spring Boot 中，可以使用注解@Scheduled来定义定时任务，任务会按照指定的时间间隔执行。在这个方案中，我们可以定义一个定时任务，每隔 30 分钟检查一次未支付的订单，如果订单生成时间超过 30 分钟，则自动取消该订单。 代码示例： import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; @EnableScheduling @Component public class OrderCancelSchedule { @Autowired private OrderService orderService; @Scheduled(cron = "0 0/1 * * *?") public void cancelUnpaidOrders() { List<Order> unpaidOrders = orderService.getUnpaidOrders(); unpaidOrders.forEach(order -> { if (order.getCreationTime().plusMinutes(30).isBefore(LocalDateTime.now())) { orderService.cancelOrder(order.getId()); } }); } } 在上面的代码中，我们定义了一个名为OrderCancelSchedule的组件，并使用@EnableScheduling注解启用定时任务功能。在组件中，我们定义了一个名为cancelUnpaidOrders的方法，并使用@Scheduled注解来指定该方法作为定时任务执行。cron表达式"0 0/1 * * *?"表示任务每隔 1 分钟执行一次。 方案二：延迟队列 延迟队列是一种将任务延迟执行的机制，入队的元素在一定的延迟时间之后才能出队。在这个方案中，我们可以将订单的 ID 放入延迟队列中，并设置延迟时间为 30 分钟。当延迟时间到期时，从队列中取出订单 ID，并执行取消订单的操作。
