Implementing Simple Flow Orchestration with Nacos, Docker, and Spring Boot

Recently I have been developing micro‑services for data‑processing, creating an independent service for each business task to simplify expansion and flow orchestration.

After trying Spring Cloud Data Flow and finding it too heavyweight, I designed a lightweight flow‑orchestration approach based on our current stack.

The goal is to make micro‑services plug‑and‑play, allowing their input and output topics to be changed without downtime.

Nacos Installation and Basic Usage

For quick local setup, use Docker:

docker pull nacos/nacos-server

docker run --env MODE=standalone --name nacos -d -p 8848:8848 nacos/nacos-server

Then access the console at http://<i>host</i>:8848/nacos with username nacos and password nacos.

Prepare Three Spring Boot Services with Nacos and Kafka

<parent>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-parent</artifactId>
  <version>2.1.0.RELEASE</version>
</parent>

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>

<dependency>
  <groupId>com.alibaba.boot</groupId>
  <artifactId>nacos-config-spring-boot-starter</artifactId>
  <version>0.2.1</version>
</dependency>

Configuration file (application.yml):

spring:
  kafka:
    bootstrap-servers: kafka-server:9092
    producer:
      acks: all
    consumer:
      group-id: node1-group
      enable-auto-commit: false
nacos:
  config:
    server-addr: nacos-server:8848

Business Logic Explanation

node1 listens to the upstream service’s output topic.

node2 consumes node1’s output topic, and node3 consumes node2’s output.

Removing a node only requires changing the downstream sink topic, achieving plug‑and‑play.

If a node fails, the flow can be redirected to a temporary storage service to avoid Kafka backlog.

Nacos Configuration

Create configuration items in Nacos for each service’s input and sink topics, using the service name as groupId and the item name as dataId.

@Configuration
@NacosPropertySource(dataId = "input", groupId = "node1-server", autoRefreshed = true)
@NacosPropertySource(dataId = "sink", groupId = "node1-server", autoRefreshed = true)
public class NacosConfig {
    @NacosValue(value = "${input:}", autoRefreshed = true)
    private String input;
    @NacosValue(value = "${sink:}", autoRefreshed = true)
    private String sink;
    public String getInput() { return input; }
    public String getSink() { return sink; }
}

Listening to Configuration Changes

@Component
public class ConsumerManager {
    @Value("${spring.kafka.bootstrap-servers}")
    private String servers;
    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private boolean enableAutoCommit;
    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;
    @Autowired
    private NacosConfig nacosConfig;
    @Autowired
    private KafkaTemplate kafkaTemplate;
    private String topic;
    private ExecutorService executorService;

    @NacosConfigListener(dataId = "node1-server", groupId = "input")
    public void inputListener(String input) {
        String inputTopic = nacosConfig.getInput();
        if (topic != null) {
            executorService.shutdownNow();
        }
        topic = inputTopic;
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat(topic + "-pool-%d").build();
        executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(2), threadFactory);
        executorService.execute(() -> consumer(topic));
    }

    public void consumer(String topic) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", servers);
        properties.put("enable.auto.commit", enableAutoCommit);
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("group.id", groupId);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList(topic));
        try {
            while (!Thread.currentThread().isInterrupted()) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                for (ConsumerRecord<String, String> record : records) {
                    String message = record.value();
                    String handleMessage = handle(message); // business logic omitted
                    kafkaTemplate.send(nacosConfig.getSink(), handleMessage);
                }
            }
            consumer.commitAsync();
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        } finally {
            try { consumer.commitSync(); } finally { consumer.close(); }
        }
    }

    private String handle(String message) { /* ... */ return message; }
}

Summary

The flow‑orchestration concept focuses on making data‑flow direction adjustable; by leveraging Nacos for dynamic configuration and Kafka for messaging, we can build a plug‑and‑play micro‑service pipeline that can be re‑wired at runtime without service downtime.