Spring Batch集成Pub/Sub到Cloud SQL时MessageChannel无法解析求助

我基于Spring框架开发应用，需要通过Spring Batch将Google Cloud Pub/Sub的消息加载到Cloud SQL（PostgreSQL）中。搭建测试应用后，在PubSubConfig配置类中接收Pub/Sub消息时，出现MessageChannel符号无法解析的错误，同时不确定当前的Batch与PubSub集成方式是否正确，请求调试并获取Spring新手指导。

错误原因及修复步骤 #

1. 导入类错误（直接导致MessageChannel相关符号无法解析） #

PubSubConfig中存在两处关键导入错误：

• 错误导入了org.aspectj.bridge.MessageHandler，应使用Spring Messaging模块的MessageHandler
• DirectChannel未导入，导致无法实例化消息通道

修复后的PubSubConfig代码：

// PubSubConfig.java (修复后)
package com.PubSub_Pipeline.demo.config;

import com.PubSub_Pipeline.demo.entity.PubSubMessage;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.cloud.spring.pubsub.core.PubSubTemplate;
import com.google.cloud.spring.pubsub.integration.inbound.PubSubInboundChannelAdapter;
import com.google.cloud.spring.pubsub.support.BasicAcknowledgeablePubsubMessage;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;

import java.util.concurrent.CopyOnWriteArrayList;

@Configuration
public class PubSubConfig {

    // 用线程安全的List存储PubSub消息，供Spring Batch读取
    private final CopyOnWriteArrayList<PubSubMessage> messageList = new CopyOnWriteArrayList<>();

    @Bean
    public PubSubInboundChannelAdapter messageChannelAdapter(MessageChannel inputChannel, PubSubTemplate pubSubTemplate) {
        PubSubInboundChannelAdapter adapter =
                new PubSubInboundChannelAdapter(pubSubTemplate, "your-subscription-name");
        adapter.setOutputChannel(inputChannel);
        return adapter;
    }

    @Bean
    public MessageChannel inputChannel() {
        return new DirectChannel();
    }

    @Bean
    public IntegrationFlow messageFlow(MessageChannel inputChannel) {
        return IntegrationFlows.from(inputChannel)
                .handle(messageReceiver())
                .get();
    }

    @Bean
    public MessageHandler messageReceiver() {
        return message -> {
            BasicAcknowledgeablePubsubMessage originalMessage =
                    message.getHeaders().get("gcp_pubsub_acknowledgement",
                            BasicAcknowledgeablePubsubMessage.class);

            try {
                String json = (String) message.getPayload();
                PubSubMessage pubSubMessage = new ObjectMapper().readValue(json, PubSubMessage.class);
                // 将消息加入线程安全列表，供Batch读取
                messageList.add(pubSubMessage);
                originalMessage.ack();
            } catch (Exception e) {
                // 处理解析失败，nack消息让PubSub重新投递
                originalMessage.nack();
                e.printStackTrace();
            }
        };
    }

    // 提供消息列表给Spring Batch的Reader
    @Bean
    public CopyOnWriteArrayList<PubSubMessage> messageList() {
        return messageList;
    }
}

2. Spring Batch配置修复（打通PubSub消息与Batch Reader） #

原SpringBatchConfig中ListItemReader依赖的List<PubSubMessage>未正确注入，且未考虑并发安全问题，修复后代码如下：

// SpringBatchConfig.java (修复后)
package com.PubSub_Pipeline.demo.config;

import com.PubSub_Pipeline.demo.entity.PubSubMessage;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.util.concurrent.CopyOnWriteArrayList;

@Configuration
@EnableBatchProcessing
public class SpringBatchConfig {
    @Bean
    public Job job(JobRepository jobRepository, Step pubSubStep) {
        return new JobBuilder("pubSubToCloudSQLJob", jobRepository)
                .start(pubSubStep)
                .build();
    }

    @Bean
    public Step pubSubStep(JobRepository jobRepository, PlatformTransactionManager transactionManager, ListItemReader<PubSubMessage> pubSubReader,
                           JdbcBatchItemWriter<PubSubMessage> writer) {
        return new StepBuilder("pubSubStep", jobRepository)
                .<PubSubMessage, PubSubMessage>chunk(10, transactionManager)
                .reader(pubSubReader)
                .writer(writer)
                .build();
    }

    @Bean
    public ListItemReader<PubSubMessage> pubSubReader(CopyOnWriteArrayList<PubSubMessage> messageList) {
        // 每次读取后清空列表，避免重复处理
        return new ListItemReader<>(messageList) {
            @Override
            public PubSubMessage read() {
                PubSubMessage message = super.read();
                if (message == null) {
                    messageList.clear();
                }
                return message;
            }
        };
    }

    @Bean
    public JdbcBatchItemWriter<PubSubMessage> writer(DataSource dataSource) {
        // 修正SQL：POJO无email字段，移除对应列
        return new JdbcBatchItemWriterBuilder<PubSubMessage>()
                .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
                .sql("INSERT INTO message (id, name) VALUES (:id, :name)")
                .dataSource(dataSource)
                .build();
    }
}

3. POJO与数据库表匹配修复 #

原PubSubMessage未定义email字段，但Batch Writer的SQL中包含:email参数，会导致绑定错误，已在上述修复中移除SQL里的email列；若业务需要该字段，需在PubSubMessage中添加email属性及对应的getter/setter方法。

Spring新手指导建议 #

• 依赖管理：确保项目依赖中包含Spring Cloud GCP Pub/Sub、Spring Batch、PostgreSQL驱动的正确版本，示例Maven依赖：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-batch</artifactId>
    </dependency>
    <dependency>
        <groupId>com.google.cloud</groupId>
        <artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
</dependencies>

• 并发安全：PubSub消息异步接收，Spring Batch同步处理，必须使用线程安全集合（如CopyOnWriteArrayList）存储消息，避免并发修改异常
• 错误处理：消息解析或处理失败时调用nack()，让PubSub重新投递消息，避免数据丢失
• Batch触发方式：当前配置为启动后自动执行一次Job，若需定时或按消息量触发，可添加@Scheduled定时任务，或在消息达到批量阈值时手动触发Job
• 日志调试：添加Slf4j日志组件，在消息接收、Batch处理前后打印关键日志，便于排查问题

内容的提问来源于stack exchange，提问作者Shreyansh Srivastava