Spring线程内非Bean类方法@Transactional失效问题求助

首先得明确：Spring的@Transactional注解是通过AOP代理实现的，只有Spring容器管理的Bean（比如标注了@Component、@Service、@Repository的类）的方法，才会被Spring自动生成代理并处理事务逻辑。

你的LoadingWorker是手动new出来的普通类，完全不在Spring的管理范围内，所以它的run()方法上的@Transactional注解根本不会被Spring识别，自然无法开启事务——这就是你遇到报错的核心原因。

下面提供几种可行的方案，你可以根据自己的代码结构选择：

方案1：将LoadingWorker改为Spring原型Bean #

把LoadingWorker变成Spring管理的Bean，同时设置为原型作用域（每次获取都生成新实例，避免线程安全问题），这样Spring会为它生成代理，事务注解就能生效。

修改LoadingWorker： #

@Component
@Scope("prototype") // 关键：每次获取都是新实例，适合多线程场景
public class LoadingWorker implements Runnable {
    private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

    // 改用Spring自动注入，不再通过构造器传参
    @Autowired
    private ModelMapper modelMapper;
    @Autowired
    private ActionRepository dataRepository;
    @Autowired
    private MessageChannel outboundChannel;

    // 空构造器，让Spring创建实例
    public LoadingWorker() {}

    @Override
    @Transactional(readOnly = true)
    public void run() {
        // 原有逻辑完全不变
        LOG.info("LoadingWorker.run() started.");
        long startTime = System.currentTimeMillis();
        long counter = 0;
        try(Stream<ActionEntity> entityStream = dataRepository.getAll()) {
            counter = entityStream.peek(entity -> {
                ActionDocument ad = modelMapper.map(entity, ActionDocument.class);
                LOG.debug("About to build a message '{}'", ad);
                Message<ActionDocument> message = MessageBuilder.withPayload(ad).build();
                try {
                    outboundChannel.send(message);
                } catch (MessagingException me) {
                    LOG.error("Exception encountered while writing request message to queue: {}", me.getRootCause());
                    LOG.debug("Exception encountered while writing request message to queue", me);
                } catch (Exception e) {
                    LOG.error("Some exception encountered while writing request message to queue", e);
                }
            }).count();
        }
        LOG.info("LoadingWorker.run() finished: {} Documents ({} ms)", counter, System.currentTimeMillis() - startTime);
    }
}

修改LoadingWorkerFactory： #

用ObjectFactory获取原型Bean，确保每次拿到新实例：

@Component
public class LoadingWorkerFactory {
    @Autowired
    private ObjectFactory<LoadingWorker> loadingWorkerObjectFactory;

    public LoadingWorker getLoadingWorker() {
        // 获取新的原型实例
        return loadingWorkerObjectFactory.getObject();
    }
}

方案2：手动用TransactionTemplate管理事务 #

如果不想把LoadingWorker改成Spring Bean，可以用TransactionTemplate手动开启事务，完全控制事务生命周期。

第一步：配置TransactionTemplate（Spring Boot可省略，自动配置） #

@Configuration
public class TransactionConfig {
    @Bean
    public TransactionTemplate transactionTemplate(PlatformTransactionManager transactionManager) {
        TransactionTemplate template = new TransactionTemplate(transactionManager);
        template.setReadOnly(true); // 设置为只读事务，符合你的场景
        return template;
    }
}

第二步：修改LoadingWorker和Factory： #

public class LoadingWorker implements Runnable {
    private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
    private ModelMapper modelMapper;
    private ActionRepository dataRepository;
    private MessageChannel outboundChannel;
    private TransactionTemplate transactionTemplate;

    // 构造器新增TransactionTemplate参数
    public LoadingWorker(ActionRepository repository, MessageChannel channel, ModelMapper modelMapper, TransactionTemplate transactionTemplate) {
        this.modelMapper = modelMapper;
        this.dataRepository = repository;
        this.outboundChannel = channel;
        this.transactionTemplate = transactionTemplate;
    }

    @Override
    public void run() {
        LOG.info("LoadingWorker.run() started.");
        long startTime = System.currentTimeMillis();
        // 用transactionTemplate执行事务逻辑
        Long counter = transactionTemplate.execute(status -> {
            long count = 0;
            try(Stream<ActionEntity> entityStream = dataRepository.getAll()) {
                count = entityStream.peek(entity -> {
                    // 原有处理逻辑不变
                    ActionDocument ad = modelMapper.map(entity, ActionDocument.class);
                    LOG.debug("About to build a message '{}'", ad);
                    Message<ActionDocument> message = MessageBuilder.withPayload(ad).build();
                    try {
                        outboundChannel.send(message);
                    } catch (MessagingException me) {
                        LOG.error("Exception encountered while writing request message to queue: {}", me.getRootCause());
                        LOG.debug("Exception encountered while writing request message to queue", me);
                    } catch (Exception e) {
                        LOG.error("Some exception encountered while writing request message to queue", e);
                    }
                }).count();
            }
            return count;
        });
        LOG.info("LoadingWorker.run() finished: {} Documents ({} ms)", counter, System.currentTimeMillis() - startTime);
    }
}

@Component
public class LoadingWorkerFactory {
    @Autowired
    ModelMapper modelMapper;
    @Autowired
    private ActionRepository actionRepository;
    @Autowired
    private MessageChannel actionBulkOutboundChannel;
    @Autowired
    private TransactionTemplate transactionTemplate;

    public LoadingWorker getLoadingWorker() {
        return new LoadingWorker(actionRepository, actionBulkOutboundChannel, modelMapper, transactionTemplate);
    }
}

方案3：将事务逻辑抽离到Spring Service #

把数据流处理的核心逻辑移到一个Spring管理的Service类中，让LoadingWorker只负责线程执行，调用Service的事务方法。这种方式更符合Spring的分层架构思想。

第一步：创建LoadingService： #

@Service
public class LoadingService {
    private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

    @Autowired
    private ActionRepository actionRepository;
    @Autowired
    private ModelMapper modelMapper;
    @Autowired
    private MessageChannel actionBulkOutboundChannel;

    @Transactional(readOnly = true)
    public long processAllData() {
        long counter = 0;
        try(Stream<ActionEntity> entityStream = actionRepository.getAll()) {
            counter = entityStream.peek(entity -> {
                ActionDocument ad = modelMapper.map(entity, ActionDocument.class);
                LOG.debug("About to build a message '{}'", ad);
                Message<ActionDocument> message = MessageBuilder.withPayload(ad).build();
                try {
                    actionBulkOutboundChannel.send(message);
                } catch (MessagingException me) {
                    LOG.error("Exception encountered while writing request message to queue: {}", me.getRootCause());
                    LOG.debug("Exception encountered while writing request message to queue", me);
                } catch (Exception e) {
                    LOG.error("Some exception encountered while writing request message to queue", e);
                }
            }).count();
        }
        return counter;
    }
}

第二步：修改LoadingWorker和Factory： #

public class LoadingWorker implements Runnable {
    private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
    private LoadingService loadingService;

    public LoadingWorker(LoadingService loadingService) {
        this.loadingService = loadingService;
    }

    @Override
    public void run() {
        LOG.info("LoadingWorker.run() started.");
        long startTime = System.currentTimeMillis();
        long counter = loadingService.processAllData();
        LOG.info("LoadingWorker.run() finished: {} Documents ({} ms)", counter, System.currentTimeMillis() - startTime);
    }
}

@Component
public class LoadingWorkerFactory {
    @Autowired
    private LoadingService loadingService;

    public LoadingWorker getLoadingWorker() {
        return new LoadingWorker(loadingService);
    }
}

• 如果想最小改动原有代码结构，优先选方案1；
• 如果不想让LoadingWorker依赖Spring容器，选方案2；
• 如果希望代码更符合Spring分层设计，逻辑更清晰，选方案3。

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