Spring Boot R2DBC测试异常：Mono的doOnNext/doOnSuccess未触发

问题场景 #

使用Spring Boot R2DBC操作MariaDB时，测试代码调用ReactiveCrudRepository.save()后，通过block()能成功获取保存后的实体（数据已写入数据库），但doOnNext和doOnSuccess回调完全没有触发，仅打印了后续的"We are here!"。

数据库结构 #

CREATE DATABASE IF NOT EXISTS `bugdemo`;
USE `bugdemo`;

CREATE TABLE IF NOT EXISTS `persons` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
  `firstname` varchar(50) DEFAULT NULL,
  `lastname` varchar(50) DEFAULT NULL,
  `title` varchar(50) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8mb4;

实体类 #

@NoArgsConstructor
@Data
@Table(name = "persons")
public class DataDAO
{
   @Id
   private Long id;
   private String firstname;
   private String lastname;
   private String title;

   public DataDAO(String first,String last,String atitle)
   {
      id = null;
      firstname = first;
      lastname = last;
      title = atitle;
   }
}

仓库接口 #

public interface BugRepository extends ReactiveCrudRepository<DataDAO, Long>
{
}

原测试代码 #

@SpringBootTest
class SavebugApplicationTests 
{
   @Autowired
   private BugRepository theRepo;

    @Test
    void testAddOneRecord() 
    {
       DataDAO dao = new DataDAO("Joe","User","Employee");
   
       Mono<DataDAO> row = theRepo.save(dao);
   
       row.doOnNext(next->{
          System.out.println("Data is saved to ID: "+next.getId());
       })
       .doOnSuccess(saved->{
         System.out.println("Saved data. ID="+saved.getId());
      });
   
       DataDAO check = row.block();

      System.out.println("We are here!");
    }
}

问题原因 #

Reactor（Spring R2DBC基于Reactor实现）的操作符（比如doOnNext、doOnSuccess）采用无副作用的链式调用设计：调用这些操作符后，会返回一个新的Mono/Publisher实例，原实例不会被修改。

当前代码的问题：

1. 对row调用doOnNext和doOnSuccess后，生成了带回调的新Mono，但未保存或订阅这个新实例；
2. 最终调用block()的是原row实例，这个实例没有附加任何回调逻辑，因此回调不会触发；
3. 数据能成功保存是因为block()触发了原row的订阅（执行save操作），但回调属于另一个未被订阅的Mono，自然不会执行。

解决方案 #

需要确保附加了回调的Mono实例被订阅（或block），修改后的测试代码如下：

@SpringBootTest
class SavebugApplicationTests 
{
   @Autowired
   private BugRepository theRepo;

    @Test
    void testAddOneRecord() 
    {
       DataDAO dao = new DataDAO("Joe","User","Employee");
   
       // 对附加了回调的Mono执行block
       DataDAO check = theRepo.save(dao)
               .doOnNext(next -> {
                   System.out.println("Data is saved to ID: " + next.getId());
               })
               .doOnSuccess(saved -> {
                   System.out.println("Saved data. ID=" + saved.getId());
               })
               .block();

      System.out.println("We are here!");
    }
}

也可以先保存带回调的Mono再block：

Mono<DataDAO> rowWithCallbacks = theRepo.save(dao)
        .doOnNext(next -> {
            System.out.println("Data is saved to ID: " + next.getId());
        })
        .doOnSuccess(saved -> {
            System.out.println("Saved data. ID=" + saved.getId());
        });

DataDAO check = rowWithCallbacks.block();

核心要点：响应式流的操作符不会修改原对象，必须使用操作符返回的新对象进行后续的订阅/阻塞操作。

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