批处理现在是组织执行的一种非常常见的任务，用于处理大量记录，包括日志/跟踪、事务管理、作业处理统计等任务。当您处理大量数据时，不能保证输入将来自另一个基于java的应用程序或甚至数据库。事实上，它很可能必须使用消息总线和队列进行批处理。这是春天批次来拯救!!
在使用消息传递队列时，Spring Batch提供了有益的阅读器，即:JmsItemReader而且AmqpItemReader．
本文将集中讨论使用Spring batch和Rabbit MQ的批处理和一个消息队列，相对来说不需要花多少功夫。
什么是RabbitMQ?
RabbitMQ是一个轻量级、可靠、可扩展和可移植的消息代理。但是与Java开发人员熟悉的许多消息代理不同，它不是基于JMS的。相反，您的应用程序通过与平台无关的有线级协议与它通信:高级消息队列协议(AMQP)。
在批处理的情况下，这尤其有用——即使是从另一个Java应用程序发送消息，也可以避免通过网络发送pojo带来的麻烦，因为这需要在多个应用程序之间建立一致的模型。也就是说，发送和接收字符串要灵活得多。
先决条件
- RabbitMQ应该安装在这个项目可以访问的地方。
- 使用Spring引导项目Spring工具套件．
依赖关系
在你的POM中，一定要在你的依赖项中包含以下构件:
< >的依赖…<依赖> < groupId > org.springframework。bootspring-boot-starter-batch …<依赖> < groupId > org.springframework。bootspring-boot-starter-amqp …< / >的依赖关系
配置
由于这是Spring Boot，您将需要应用程序的入口点。所以，一定要在src/main/java中包含如下类:
@SpringBootApplication公共类MyBatchApplication{公共静态void main(String[] args) {SpringApplication.run(MyBatchApplication.class, args);}}
我们需要为项目设置一个总体配置
公共类MyBatchAppConfig {}
注意到@EnableBatchProcessing注释。这是必须的，以便让Spring Boot知道在设置Spring Batch时要承担很多繁重的工作。
使用Spring AMQP设置消息队列，它还将提供帮助与队列交互所需的模板。
进口org.springframework.amqp.core.Queue;进口org.springframework.amqp.rabbit.connection.ConnectionFactory;进口org.springframework.amqp.rabbit.core.RabbitTemplate;进口org.springframework.beans.factory.annotation.Autowired;进口org.springframework.context.annotation.Bean;…String queueName = "my-queue";@Autowired private ConnectionFactory;//如果它不存在，将创建一个新的队列;否则，它将使用已有的同名文件//…the second argument means to make the queue 'durable' @Bean public Queue myQueue() { return new Queue(queueName, true); } // this is necessary for operations with Spring AMQP @Bean public RabbitTemplate getMyQueueTemplate() { RabbitTemplate template = new RabbitTemplate(this.rabbitConnectionFactory); template.setQueue(queueName); return template;
配置批处理作业一步，也就是说，假设我们使用的是chunk-based处理而不是微线程。
进口org.springframework.batch.core.Step;进口org.springframework.batch.core.configuration.annotation.StepBuilderFactory;进口org.springframework.batch.item.ItemProcessor;进口org.springframework.batch.item.ItemReader;进口org.springframework.batch.item.ItemWriter;进口org.springframework.batch.item.amqp.AmqpItemReader;…@Autowired private StepBuilderFactory;…@Bean公共步骤getMyJobStep(){返回this.stepBuilderFactory.get("myJobStep") .chunk(1) .reader(this.getMyReader()) .processor(this.getMyProcessor()) .writer(this.getMyWriter()) .build(); } @Bean public ItemReader getMyReader() { return new AmqpItemReader(this.getMyQueueTemplate()); } @Bean public ItemProcessor getMyProcessor() { return new MessageProcessor(); } @Bean public ItemWriter getMyWriter() { return new MessageWriter(); }
的AmqpItemReader，它告诉我们它期望一个String来表示队列上的每个消息，它将前面创建的队列模板作为参数。的AmqpItemReader是一个Spring Batch开箱即用的类——不需要其他工作就可以让它工作!这就是Spring Batch的神奇之处!
结论
我们介绍了Spring Batch项目如何以最小的设置接受来自基于amqp的消息队列的输入。
是的，就是这样!!我们准备出发了。
如果你喜欢这篇文章，觉得它很有知识性，一定要和你的同事分享。也请在评论区告诉我我即将发布的博客的主题。
