Rmessage

歡迎閱讀Rmessage文檔

技術堆疊

Rmessage是採用Reactor3,基於reactor-netty專案所建構的即時訊息推送api。

• 什麼是Reactor3?

  Reactor 是一個用於JVM的完全非阻塞的響應式程式設計框架，具備高效率的需求管理（即對「背壓（backpressure）」的控制）能力。它與Java 8 函數式API 直接集成，例如CompletableFuture， Stream， 以及Duration。它提供了非同步序列API Flux（用於[N]個元素）和Mono（用於[0|1]個元素），並完全遵循並實現了「響應式擴展規範」（Reactive Extensions Specification）。

• 使用Reactor好處?

  非常容易建立高吞吐量純異步的程式碼,還有就是能夠無縫整合spring5[webflux]專案。

專案簡介

使用Rmessage你需要外部管理群組用戶關係,以及離線訊息存儲，Rmessage不提供持久化，測試可以使用預設Handler記憶體保留離線訊息。 整個專案採用純粹非同步的程式設計思想去開發，旨在學習reactive programming。

• 號碼雲端位址
• github位址

目前支援的功能

1. 單聊
2. 群聊
3. 離線訊息落地以及拉取
4. 離線線上通知管理
5. 心跳,連結鑑權機制
6. 群組關係管理
7. 支援多端在線
8. 目前支援tcp協議，專案設計時考慮多協議擴展。
9. 支援服務端自訂訊息攔截
10. 訊息QOS機制(還未實現)

快速開始

• 服務端

Rmessage不管理用戶群組之間關係,透過介面暴漏外部，只要實現介面注入即可。

 ServerStart
   . builder ()
   . tcp ()
   . ip ( "127.0.0.1" )
   . port ( 1888 )
   . onReadIdle ( 10000l ) //设置读心跳时间
   . onWriteIdle ( 10000l ) //设置写心跳时间
   . option ( ChannelOption . SO_RCVBUF , 1023 )
   . interceptor ( frame -> frame , frame -> frame ) // 拦截所有message
   . setAfterChannelInit ( channel -> { //  channel设置
      })
   . connect ()
   . cast ( TcpServerSession . class )
   . subscribe ( session ->{
                    session . addGroupHandler ( groupId -> null ). subscribe ();
                    session . addOfflineHandler ( new DefaultOffMessageHandler ()). subscribe ();
                    session . addUserHandler ( new DefaultUserTransportHandler ());
  });

具體介面方法定義請參考介面註釋

1. UserHandler介面是處理使用者登陸校驗等
2. OfflineHandler 是處理離線訊息儲存,以及拉取離線訊息的接口
3. GroupHandler 是取得群組下所有使用者的介面

• 客戶端

 ClientStart
     . builder ()
     . tcp ()
     . ip ( "127.0.0.1" )
     . port ( 1888 )
     . userId ( "21344" )  //设置用户名
     . password ( "12312" ) //设置密码
     . onReadIdle ( 10000l ,()->()-> System . out . println ( "心跳了" )) //设置读心跳,以及设置回调runner
     . setClientType ( ClientType . Ios ) //设置客户端类型
     . setAfterChannelInit ( channel -> {
                    //  channel设置
      })
     . connect ()
     . cast ( TcpClientSession . class )
     . subscribe ( session ->{
      session . sendPoint ( "123" , "测试一下哦" ). subscribe (); //发送单聊消息
                    session . sendGroup ( "group1" , "123" ). subscribe ();  // 发送群聊消息
                    session . accept ( message -> {
    }); // 接受所有消息
  });

協議設計

• 單聊多聊

FixHeader 【1 byte】

client_type	message_type
high 4bit	low 4bit

Topic 【n byte】

from length	to length	from	to
1byte	1byte	n byte	n byte

Body 【n byte】

body length	body
2 byte	n byte

timstamp
8 byte

• Online報文

FixHeader 【1 byte】

client_type	message_type
high 4bit	low 4bit

ConnectionState 【n byte】

user length	password length	user	password
1byte	1byte	n byte	n byte

• Offline報文

FixHeader 【1 byte】

client_type	message_type
high 4bit	low 4bit

ConnectionState 【n byte】

user length	user
1byte	n byte

• Ping報文/Pong報文

FixHeader 【1 byte】

client_type	message_type
high 4bit	low 4bit