当前位置：首页 > news >正文

用java实现的一个本地文件队列，支持并发消费与顺序消费

news 来源：原创 2025/5/2 19:39:32

为什么

在开发中用了很多sqlite本地数据库，sqlite是单例的，为了实现不同实例间的同步，解耦数据写入与数据同步，需要一个本地文件队列来实现，所以做了这个。

为什么不用kafka这些现成且成熟的队列，因为我的服务器要运行在极小的环境中，比如一部安卓手机、一个树莓派中，kafka太大了。也有现成的本地文件队列，为什么不用，因为它们大多不维护了，并且并不完全覆盖我需要的场景。

源码及jar在码云、gitcode、github都可以获取

fileq: 使用java，实现一个高效的本地文件队列。 (gitee.com)

至简网格 / fileq · GitCode

实现原理

Provider -> Consumers

Provider将消息存入本地文件，多个消费者从文件中顺序读取，每消费一次记录都会记录消费的位置。

Provider写入时，如果超过文件最大容许的大小，则重新创建一个文件；如果文件数量超过上限，就会删除所有Consumers都已经消费过的文件。所以，如果有多个消费者，其中有一个很慢，则会出现即使文件超过上限了，文件仍然会保留。

每个Consumer独立消费，消费位置记录在独立的本地文件中。记录位置的文件每次写入2个int，分别记录文件编号、在该文件中的消费位置。采用顺序写入的方式，重新加载时使用最后一对记录。当该文件体积达到阈值时，重置文件，从头写起，丢掉前面无用的信息。这样的实现是为了利用文件顺序写入性能高的特点。

本地文件在fileq中至关重要，所以，它所使用的路径不可删改。

使用方法

推荐的使用方式见以下代码：

private static final int N = 10;
private static final Logger LOG = LogUtil.getInstance();

public static void main(String[] args) throws Exception {
	CountDownLatch counter = new CountDownLatch(N);
	if(!FQTool.started()) {
		ExecutorService threadPool = Executors.newCachedThreadPool();
		FQTool.start(threadPool);
	}
	String queueDir = FileUtil.addPath(System.getProperty("user.dir"), "test");
	FileQueue.Builder builder = new FileQueue.Builder(queueDir, "queue")
        //根据缓存消息的量，确定文件个数，即使文件数超过这个数字，队列也不会删除未消费的文件
		.maxFileNum(50) 
        //单个文件不宜过大，根据队列的规模确定单个文件的大小，避免频繁产生新文件
		.maxFileSize(16 * 1024 * 1024) 
        //false表示及时落盘，但是性能只有20万每秒，能满足绝大部分场景的性能需求
        //true表示使用缓存方式写文件，性能高，但是宕机时，可能有部分消息没有落盘
		.bufferedPush(false) 
		.bufferedPoll(true) //poll时尽量使用缓存文件
        //消费位置缓存多少次才落盘，此值越大，宕机重启时，重复消费的次数可能越多
        //设置的太小，会导致频繁的写消费位置，性能下降很严重，默认为1024
        .bufferedPos(100); 
	FileQueue fq = FQTool.create(builder);
	
	fq.addConsumer("sequential", true, (msg, reader) -> { //保证消费的顺序
		LOG.debug("Msg1:{}", new String(msg.message(), 0, msg.len()));
        reader.confirm(true); //消费完必须确认，只有确认了，才会继续往下消费
		counter.countDown();
		return true;
	});
	
	fq.addConsumer("concurrent", false, (msg, reader) -> { //不保证消费的顺序
		LOG.debug("Msg2:{}", new String(msg.message(), 0, msg.len()));
        //无需确认，不管此处消费是否成功，都会继续往下消费
		counter.countDown();
		return true;
	});
	
	for(int i = 0; i < 10; i++) {
		fq.push(("test" + i).getBytes());
	}
	counter.await();
	LOG.debug("over");
	System.exit(0);
}

其他测试代码请参考项目中的test目录。