f-mqtt 是开源的,支持集群,高可用的 mqtt broker 实现
目前支持 3.1.1 和 3.1 版本的 mqtt 协议,服务质量仅支持 qos 0 和 1
f-mqtt 试图提供一个可自定义的 mqtt broker 框架,
基于该框架你可以自定义属于符合自身要求的 mqtt broker

1.订阅树
    1.作用
        用于构建全局订阅树,有了全局订阅树之后,才可以通过topic找到对应的订阅客户端
        再根据clientId拿到Session信息,这样就可以通过Session中的Channel发送消息
    2.存储方式
        对于mqtt broker来说,主要的工作就是收发消息,所以订阅树以内存的方式存在是最好的
    3.同步方式
        订阅信息的载体是Session信息,同步方式详见Session同步
        基于客户端的subscribe、unsubscribe,对订阅树进行增删改
        具体实现详见,subscription-service中的MemorySubscriptionService
        
2.Session信息
    1.作用
        Session作为客户端载体,承载了该客户端的订阅信息,queue信息,元数据信息等
    2.存储方式
        cleanSession=true,无需进行持久化,故只考虑cleanSession=false的情况
            将Session信息分成两部分进行存储
                1.元数据
                    1.本地存储「由于每个节点都是对等节点,如果存在大量的持久化Session,建议使用第三方存储」
                        Session信息的数据结构属于JSON,存在增删改的场景
                        对于本地存储,需要实现JSON的增量更新较为复杂
                        最终选型,h2-mvstore来实现「支持事务、mvcc等」
                        具体实现详见,session-service中的H2SessionService
                    2.第三方存储
                        通过Hash结构对会话信息进行存储
                        具体实现详见,session-service中的RedisSessionService
                2.Queue队列
                    1.第三方存储
                        由于消息的特殊性,直接通过第三方进行存储
                        在Redis中构建了三个key
                        1.inflight
                            Hash结构,clientId+":"+inflight为key,messageId为field,value为QueueMessage
                        2.queueList,List结构
                            List结构,clientId+":"+queueList为key,messageId为value,为了保证队列的顺序
                        3.clientId+":"+queueMap
                            Hash结构,clientId+":"+queueMap为key,messageId为field,value为QueueMessage,
                            为了对齐inflight,便于在Lua中更好的进行queueMap到inflight的消息转移
                        当broker.session.maxInflightMessages=1时,就能保证QoS1消息的顺序
                        具体实现详见,queue-service中的RedisQueueService
                3.订阅信息
                        
    3.同步方式,最终一致性
        1.基于消息中间件,能够对消息进行持久化,并且可以削峰填谷,
            1.如果是本地存储的话,
                可以通过消费Kafka重新构建最新的订阅信息,
                对于一个全新的消费者组来说,通过消费所有消息来构建Session信息是不符合预期的.
            2.如果是第三方存储的话,
                因为消息中间件属于增量同步,速度很快,
                如果broker定期做全量同步的话,对收发消息影响很大.
                
            最终选型,Kafka,有了Kafka之后为后期切换存储方式也提供遍历,
            仅需把Kafka中的数据ETL到第三方即可
                    
        2.基于RPC请求通知,需要实现broker之间的服务发现,并且还不能解决持久化问题
        
        3.消息顺序
            1.保证消息顺序的意义
                比如同一个主题,客户端取消订阅事件先于订阅事件被消费,会导致一直订阅着某个主题;
                相反订阅事件先于取消订阅事件,会导致订阅丢失.
                
                无论存储、同步方式如何,顺序都逃不掉的
            
            2.实现方式
                1.同一JVM中,需要对Session加锁,保证线程安全,
                    最终对clientId进行加锁
                2.不同JVM之间,
                    1.首先排除通过消息中间件本身来保持顺序的方式,性能太低
                    2.最终采用版本号的方式保证消息顺序
                        1.在处理客户端的connect、disconnect时,会自增该clientId对应的sessionVersion,
                            当消费connect、disconnect消息时,会判断消息的版本号是否低于全局版本号,低于则忽略
                            
                        2.在处理客户端的subscribe、unsubscribe时,会自增该clientId+topic对应的subscriptionVersion
                            当消费subscribe、unsubscribe消息时,会判断消息的版本号是否低于全局版本号,低于则忽略
                    
                        3.版本号的方式会在connect、disconnect、subscribe、unsubscribe时,
                            需要与获取全局的版本号,这里会有损耗,但是这四类消息量相对较少,是可以接收的
                        
                        4.Queue信息
                            在处理客户端的connect时,会自增该clientId对应的queueVersion,「仅此一个地方」
                                随后对queue的操作,添加消息、删除消息、删除queue队列,都会判断当前版本是否低于全局版本号,低于则忽略
                                如果出现乱序,则会导致上个版本号的消息存在遗留的问题,但是这也满足QoS1至少一次的语义,
                                并且通常情况下,断开连接之后不会立即连接上.
                                
                        5.基于Redis实现,利用Lua脚本保证多个操作的原子性
                        
3.retain信息
    1.存储方式
        1.本地存储
            具体实现详见,retain-service中的H2QueueService
            「大量Retain消息,或经常变更的情况下,建议使用第三方存储,并且关闭Retain消息缓存,retain.cache.enable」
        2.第三方存储
            具体实现详见,retain-service中的RedisQueueService,「」
    2.同步方式,最终一致性
        1.基于版本实现

$ mvn -DskipTests clean install -U

. -d 「f-mqtt root directory」

如果有兴趣的同学,期待你的加入