springboot设置定时任务（2）

上次了解到springboot也内置了对应调度任务的支持，当时就觉得很有用，但也一时没有想到在生产中具体应该怎么用。

最近接到一个需求，正好可以用到它。

我需要开发一个接口用来接收其他部门发来的数据，并将数据存入mysql数据库。

这个需求实现不难，关键在于当数据一条一条post过来的时候，一条一条insert到数据库，可能会把数据库搞挂！

那么第一个问题来了：能不能缓存多次post的数据，攒到一定数量批次insert？

答案：可以，我们可以在springboot中实现一个Queue，在service层将接收到的数据加入Queue，另起一个线程作为消费者来消费Queue

消费者可以无限循环消费Queue中的数据，我们用一个List攒够100条就往mysql批量插入一次

这时产生了一个新问题：我们如何知道对方的数据已经发完了？

如果不确定对方什么时候发完，缓存中的数据不足100条的时候就永远没有机会存进mysql，所以这种方式不可行

这个时候我想到了调度的方式，也就是我并不启动一个常驻的线程，而是每隔固定时间启动一下消费者，消费者设定超时时间（比方1s）消费数据，

消费不到就直接break出while循环。

测试了一下，还挺好用，上代码：

/**
*构造一个Queue供生产者、消费者使用
*/
@Configuration
public class QueueConfig {

    @Bean
    public ArrayBlockingQueue<GasAccumulate> queue() {
        return new ArrayBlockingQueue<GasAccumulate>(1000);
    }
}

生产者：

@Service
public class GasServiceImpl implements GasService {

    @Autowired
    private GasAccumulateMapper gasAccumulateMapper;

    @Autowired
    private ArrayBlockingQueue<GasAccumulate> queue;

    @Override
    public void saveGasAccumulate(GasAccumulateParam gasAccumulateParam) {
        
        GasAccumulate gasAccumulate = new GasAccumulate();
       //数据加入队列设定超时时间
        try {
            queue.offer(gasAccumulate, 1, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
           //队列满时对数据进行简单单条insert
            gasAccumulateMapper.insert(gasAccumulate);
        }
    }

消费者：

@Component
public class GasAccumulateConsumer {
    private static Logger logger = LoggerFactory.getLogger(GasAccumulateConsumer.class);

    @Autowired
    private GasAccumulateMapper gasAccumulateMapper;

    @Autowired
    private ArrayBlockingQueue<GasAccumulate> queue;


    /**
     * 如果队列没有数据，1秒后超时退出while循环
     * 这样就不能作为一个常驻线程，而应该处理为一个调度线程
     */
    @Scheduled(fixedDelay = 60000)
    public void run() {
        logger.info("scheduler start at:" + LocalDateTime.now());
        List<GasAccumulate> arrayList = new ArrayList<>(100);
        GasAccumulate gasAccumulate = null;

        while (true) {
            try {
                gasAccumulate = queue.poll(1, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (gasAccumulate != null) {
                arrayList.add(gasAccumulate);
                if (arrayList.size() == 100) {
                    gasAccumulateMapper.batchInsert(arrayList);
                    arrayList.clear();
                }
            } else {
                gasAccumulateMapper.batchInsert(arrayList);
                arrayList.clear();
                break;
            }

        }
        logger.info("scheduler end at:" + LocalDateTime.now());
    }
}

启动类：

//开启调度的注解
@EnableScheduling
@EnableSwagger2
@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

}