带有Java DSL的Spring Integration MongoDB适配器

1引言

这篇文章解释了如何使用Spring Integration从MongoDB数据库中保存和检索实体。为了实现这一点，我们将使用Java DSL配置扩展来配置入站和出站MongoDB通道适配器。例如，我们将构建一个应用程序，使您可以将订单写入MongoDB存储，然后检索它们进行处理。

应用程序流程可以分为两部分：

新订单将发送到消息传递系统，在该系统中它们将被转换为实际产品，然后存储到MongoDB。

另一方面，另一个组件正在连续轮询数据库并处理它找到的任何新产品。

可以在我的Spring Integration存储库中找到源代码。

2 MessagingGateway –进入消息传递系统

我们的应用程序对消息传递系统一无所知。实际上，它只会创建新订单并将其发送到接口（OrderService）：

@SpringBootApplication
@EnableIntegration
public class MongodbBasicApplication {public static void main(String[] args) {ConfigurableApplicationContext context = SpringApplication.run(MongodbBasicApplication.class, args);new MongodbBasicApplication().start(context);}public void start(ConfigurableApplicationContext context) {resetDatabase(context);Order order1 = new Order("1", true);Order order2 = new Order("2", false);Order order3 = new Order("3", true);InfrastructureConfiguration.OrderService orderService = context.getBean(InfrastructureConfiguration.OrderService.class);orderService.order(order1);orderService.order(order2);orderService.order(order3);}private void resetDatabase(ConfigurableApplicationContext context) {ProductRepository productRepository = context.getBean(ProductRepository.class);productRepository.deleteAll();}
}

首先看一下配置，我们可以看到OrderService实际上是一个消息传递网关。

@Configuration
@ComponentScan("xpadro.spring.integration.endpoint")
@IntegrationComponentScan("xpadro.spring.integration.mongodb")
public class InfrastructureConfiguration {@MessagingGatewaypublic interface OrderService {@Gateway(requestChannel = "sendOrder.input")void order(Order order);}...
}

发送到order方法的任何订单都将通过“ sendOrder.input”直接通道作为Message <Order>引入消息系统。

3第一部分-处理订单

Spring Integration消息流的第一部分由以下组件组成：

flow_firstpart

我们使用lambda创建一个IntegrationFlow定义，该定义将DirectChannel注册为其输入通道。输入通道的名称解析为'beanName + .input'。因此，该名称就是我们在网关中指定的名称：“ sendOrder.input”

@Bean
@Autowired
public IntegrationFlow sendOrder(MongoDbFactory mongo) {return f -> f.transform(Transformers.converter(orderToProductConverter())).handle(mongoOutboundAdapter(mongo));
}

流程在收到新订单时要做的第一件事是使用变压器将订单转换为产品。要注册一个变压器，我们可以使用DSL API提供的Transformers工厂。在这里，我们有不同的可能性。我选择的是使用PayloadTypeConvertingTransformer ，它将有效负载转换为对象的委托给转换器。

public class OrderToProductConverter implements Converter<Order, Product> {@Overridepublic Product convert(Order order) {return new Product(order.getId(), order.isPremium());}
}

订单流程的下一步是将新创建的产品存储到数据库中。在这里，我们使用MongoDB出站适配器：

@Bean
@Autowired
public MessageHandler mongoOutboundAdapter(MongoDbFactory mongo) {MongoDbStoringMessageHandler mongoHandler = new MongoDbStoringMessageHandler(mongo);mongoHandler.setCollectionNameExpression(new LiteralExpression("product"));return mongoHandler;
}

如果您想知道消息处理程序在内部实际上在做什么，它将使用mongoTemplate保存该实体：

@Override
protected void handleMessageInternal(Message<?> message) throws Exception {String collectionName = this.collectionNameExpression.getValue(this.evaluationContext, message, String.class);Object payload = message.getPayload();this.mongoTemplate.save(payload, collectionName);
}

4第二部分–加工产品

在第二部分中，我们还有另一个用于处理产品的集成流程：

flow_secondpart

为了检索以前创建的产品，我们定义了一个入站通道适配器，它将继续轮询MongoDB数据库：

@Bean
@Autowired
public IntegrationFlow processProduct(MongoDbFactory mongo) {return IntegrationFlows.from(mongoMessageSource(mongo), c -> c.poller(Pollers.fixedDelay(3, TimeUnit.SECONDS))).route(Product::isPremium, this::routeProducts).handle(mongoOutboundAdapter(mongo)).get();
}

MongoDB入站通道适配器是负责从数据库轮询产品的适配器。我们在构造函数中指定查询。在这种情况下，我们每次都会轮询一种未加工的产品：

@Bean
@Autowired
public MessageSource<Object> mongoMessageSource(MongoDbFactory mongo) {MongoDbMessageSource messageSource = new MongoDbMessageSource(mongo, new LiteralExpression("{'processed' : false}"));messageSource.setExpectSingleResult(true);messageSource.setEntityClass(Product.class);messageSource.setCollectionNameExpression(new LiteralExpression("product"));return messageSource;
}

路由器定义显示了如何根据“溢价”字段将产品发送到其他服务激活器方法：

private RouterSpec<Boolean, MethodInvokingRouter> routeProducts(RouterSpec<Boolean, MethodInvokingRouter> mapping) {return mapping.subFlowMapping(true, sf -> sf.handle(productProcessor(), "fastProcess")).subFlowMapping(false, sf -> sf.handle(productProcessor(), "process"));
}

作为服务激活器，我们有一个简单的bean，它记录一条消息并将产品设置为已处理。然后，它将返回产品，以便流程中的下一个端点可以处理它。

public class ProductProcessor {public Product process(Product product) {return doProcess(product, String.format("Processing product %s", product.getId()));}public Product fastProcess(Product product) {return doProcess(product, String.format("Fast processing product %s", product.getId()));}private Product doProcess(Product product, String message) {System.out.println(message);product.setProcessed(true);return product;}
}

将产品设置为已处理的原因是因为下一步是更新其在数据库中的状态，以便不再对其进行轮询。我们通过将流再次重定向到mongoDb出站通道适配器来保存它。