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

中国建设银行官网站纪念币网站建设服务版权归谁

wzjs 2025/9/10 8:33:38

中国建设银行官网站纪念币,网站建设服务版权归谁,诚信通网站怎么做外链,高性能网站建设在线阅读为什么在开发中用了很多sqlite本地数据库，sqlite是单例的，为了实现不同实例间的同步，解耦数据写入与数据同步，需要一个本地文件队列来实现，所以做了这个。为什么不用kafka这些现成且成熟的队列，因为我的…

为什么

在开发中用了很多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目录。