草庐IT

java嵌入式持久化消息队列SMQ，改造自FQueue

SunEn 2023-04-19 原文

一、说明

　　之前项目中一直使用ConcurrentLinkedQueue做为缓冲队列（主要是单个项目内，单条改批量的场景，多个项目间使用的是rocketmq），虽然用着方便但是是纯内存的，

如果项目发生异常崩溃内存队列中的数据就会全部丢失（只能从日志中恢复）。所以一直想找一个简单高效支持持久化的嵌入式消息队列。中间用过activemq的嵌入模式，

虽然是支持持久化了，但是配置起来很繁琐，用起来也不简单，性能相比来说也不太行。

　　后来偶然发现了FQueue，项目地址：https://github.com/tietang/fqueue 

看了看项目源码，纯java编写，总共没几个类。完全可以改造成我想要的 简单高效支持持久化的嵌入式消息队列。

　　目前SMQ的性能禁用内存队列的情况下大概在50w每秒。

二、改造

　　1、因为是要做成嵌入式的，所以memcached协议相关的代码都删除了。

　　2、预创建文件删除了，还有一些零零碎碎的改动。（好几年了，记不清了）。

　　3、相较于原代码，改动最大的就是锁的部分，FQueue 读和写使用的是同一把锁，

我改成了读和写使用不同的锁，只在文件切换的时候使用同一把锁。性能大概提示了百分之20左右（本来就很快，锦上添花）。

　　4、添加了内存队列，这个主要解决同一个机器创建了大量队列（上千）时，队列消息消费较快，因为使用了内存映射磁盘（每隔10ms就会调用force()同步磁盘），

频繁操作磁盘导致磁盘io过高的问题。默认情况下队列大小超过50时才会写入持久化队列。可以在项目启动时调用SMQ.setting(String dbPath, int logSize, int memoryQueueSize)

进行设置。



三、使用


1、说明


　　目前是集成在我个人的工具类项目中的，已发布到中央仓库。项目地址：https://github.com/shenbururen/sun-utils


该项目强依赖hutool，算是个人对hutool的个性化的扩展。如果不想依赖该项目，只想单纯的使用SMQ，可以将源码中 cn.sanenen.queue包复制出来，单独使用。


2、maven引用


<!-- https://mvnrepository.com/artifact/cn.sanenen/sun-utils -->
<dependency>
    <groupId>cn.sanenen</groupId>
    <artifactId>sun-utils</artifactId>
    <version>2.3.1</version>
</dependency>


3、调用


　　SMQ使用时只有三个方法，向队列放入数据、从队列取出数据、获取队列大小（一般只在监控队列是否积压时使用，判断队列是否有数据，使用获取队列数据是否为null进行判断）。


　　我一般是写一个单独的类，通过静态方法调用。


 


/**
 * 本地持久化内存队列
 */
public class MsgQueue {
    private static final String testDataTopic = "testData";

    /**
     * 向队列放入数据，支持多线程。
     */
    public static void putTestData(TestData msg) {
        SMQ.push(testDataTopic, JSON.toJSONString(msg));
    }

    /**
     * 从队列取出数据，支持多线程。
     */
    public static TestData getTestData() {
        String poll = SMQ.pop(testDataTopic);
        if (StrUtil.isNotBlank(poll)) {
            return JSON.parseObject(poll, TestData.class);
        }
        return null;
    }

    /**
     * 获取队列大小
     */
    public static long getTestDataSize(){
        return SMQ.size(testDataTopic);
    }
}


四、注意事项


　　1、默认会在项目目录下生成一个smq的文件夹用来存放队列数据。同一个smq的文件夹同时只可被一个项目使用。


　　2、SMQ.setting(String dbPath, int logSize, int memoryQueueSize) 


　　　　dbPath文件存储目录，默认是smq（相对路径，也可以使用绝对路径），会在项目目录下创建一个smq的目录。


　　　　logSize数据文件大小，单位是M，最大不可超过2048M，默认是50M。使用过程中最好不要改变该值。


　　　　memoryQueueSize是内存队列大小，默认是50，内存队列数据积压超过memoryQueueSize后才会写入持久化队列。小于0时禁用内存队列。


　　3、最好使用在不是要求百分百消息不丢失的场景。（在项目异常停止、服务器停电关机时，有概率丢失消息。）


　　4、项目使用了addShutdownHook，会在项目关闭时将内存队列消息写入持久化队列。结束项目时最好使用kill -15  不要用 -9。否则有可能造成消息丢失。同步磁盘时间是10ms。


       5、目前已经在生产环境使用两年多。


 
            

            
持久FQueuespancolorstyleJava
        

        

            
有关java嵌入式持久化消息队列SMQ，改造自FQueue的更多相关文章

            
    
                
  1. java - 等价于 Java 中的 Ruby Hash - 2
    				我真的很习惯使用Ruby编写以下代码:my_hash={}my_hash['test']=1Java中对应的数据结构是什么？															最佳答案														HashMapmap=newHashMap();map.put("test",1);我假设？																								关于java-等价于Java中的RubyHash，我们在StackOverflow上找到一个类似的问题：															https://stackoverflow.com/questions/22737685/																		
  2. ruby-on-rails - Rails 模型——非持久类成员或属性？ - 2
    				对于Rails模型，是否可以/建议让一个类的成员不持久保存到数据库中？我想将用户最后选择的类型存储在session变量中。由于我无法从我的模型中设置session变量，我想将值存储在一个“虚拟”类成员中，该成员只是将值传递回Controller。你能有这样的类(class)成员吗？															最佳答案														将非持久属性添加到Rails模型就像任何其他Ruby类一样:classUser扩展解释:在Ruby中，所有实例变量都是私有(private)的，不需要在赋值前定义。attr_accessor创建一个setter和getter方法:classUs
  3. java - 从 JRuby 调用 Java 类的问题 - 2
    				我正在尝试使用boilerpipe来自JRuby。我看过guide从JRuby调用Java，并成功地将它与另一个Java包一起使用，但无法弄清楚为什么同样的东西不能用于boilerpipe。我正在尝试基本上从JRuby中执行与此Java等效的操作:URLurl=newURL("http://www.example.com/some-location/index.html");Stringtext=ArticleExtractor.INSTANCE.getText(url);在JRuby中试过这个:require'java'url=java.net.URL.new("http://www
  4. java - 我的模型类或其他类中应该有逻辑吗 - 2
    				我只想对我一直在思考的这个问题有其他意见，例如我有classuser_controller和classuserclassUserattr_accessor:name,:usernameendclassUserController//dosomethingaboutanythingaboutusersend问题是我的User类中是否应该有逻辑user=User.newuser.do_something(user1)oritshouldbeuser_controller=UserController.newuser_controller.do_something(user1,user2)我
  5. java - 什么相当于 ruby​​ 的 rack 或 python 的 Java wsgi？ - 2
    				什么是ruby​​的rack或python的Java的wsgi？还有一个路由库。															最佳答案														来自Python标准PEP333:Bycontrast,althoughJavahasjustasmanywebapplicationframeworksavailable,Java's"servlet"APImakesitpossibleforapplicationswrittenwithanyJavawebapplicationframeworktoruninanywebserverthatsupportstheservletAPI.ht
  6. Observability：从零开始创建 Java 微服务并监控它 （二） - 2
    这篇文章是继上一篇文章“Observability：从零开始创建Java微服务并监控它（一）”的续篇。在上一篇文章中，我们讲述了如何创建一个Javaweb应用，并使用Filebeat来收集应用所生成的日志。在今天的文章中，我来详述如何收集应用的指标，使用APM来监控应用并监督web服务的在线情况。源码可以在地址 https://github.com/liu-xiao-guo/java_observability 进行下载。摄入指标指标被视为可以随时更改的时间点值。当前请求的数量可以改变任何毫秒。你可能有1000个请求的峰值，然后一切都回到一个请求。这也意味着这些指标可能不准确，你还想提取最小/
  7. 【Java 面试合集】HashMap中为什么引入红黑树，而不是AVL树呢 - 2
    HashMap中为什么引入红黑树，而不是AVL树呢1.概述开始学习这个知识点之前我们需要知道，在JDK1.8以及之前，针对HashMap有什么不同。JDK1.7的时候，HashMap的底层实现是数组+链表JDK1.8的时候，HashMap的底层实现是数组+链表+红黑树我们要思考一个问题，为什么要从链表转为红黑树呢。首先先让我们了解下链表有什么不好？？？2.链表上述的截图其实就是链表的结构，我们来看下链表的增删改查的时间复杂度增：因为链表不是线性结构，所以每次添加的时候，只需要移动一个节点，所以可以理解为复杂度是N(1)删：算法时间复杂度跟增保持一致查：既然是非线性结构，所以查询某一个节点的时候
  8. 【Java入门】使用Java实现文件夹的遍历 - 2
    遍历文件夹我们通常是使用递归进行操作，这种方式比较简单，也比较容易理解。本文为大家介绍另一种不使用递归的方式，由于没有使用递归，只用到了循环和集合，所以效率更高一些！一、使用递归遍历文件夹整体思路1、使用File封装初始目录，2、打印这个目录3、获取这个目录下所有的子文件和子目录的数组。4、遍历这个数组，取出每个File对象4-1、如果File是否是一个文件，打印4-2、否则就是一个目录，递归调用代码实现publicclassSearchFile{publicstaticvoidmain(String[]args){//初始目录Filedir=newFile("d:/Dev");Datebeg
  9. java - 为什么 ruby​​ modulo 与 java/other lang 不同？ - 2
    				我基本上来自Java背景并且努力理解Ruby中的模运算。(5%3)(-5%3)(5%-3)(-5%-3)Java中的上述操作产生，2个-22个-2但在Ruby中，相同的表达式会产生21个-1-2.Ruby在逻辑上有多擅长这个？模块操作在Ruby中是如何实现的？如果将同一个操作定义为一个web服务，两个服务如何匹配逻辑。															最佳答案														在Java中，模运算的结果与被除数的符号相同。在Ruby中，它与除数的符号相同。remainder()在Ruby中与被除数的符号相同。您可能还想引用modulooperation.													
  10. java - Ruby 相当于 Java 的 Collections.unmodifiableList 和 Collections.unmodifiableMap - 2
    				Java的Collections.unmodifiableList和Collections.unmodifiableMap在Ruby标准API中是否有等价物？															最佳答案														使用freeze应用程序接口(interface):Preventsfurthermodificationstoobj.ARuntimeErrorwillberaisedifmodificationisattempted.Thereisnowaytounfreezeafrozenobject.SeealsoObject#frozen?.Thismethodretur
    11.             

        

        

            
随机推荐