moquette改造筆記(一):整合到SpringBoot
Moquette簡介
Mqtt作為物聯網比較流行的協議現在已經被大範圍使用,其中也有很多開源的MQTT BROKEN。Moquette是用java基於netty實現的輕量級的MQTT BROKEN. Moquette基於Netty實現,效能問題至少前期可以不用考慮,在使用過程中還算穩定,沒有出現過較大的問題。github地址:https://github.com/andsel/moq...。
整合到SpringBoot
本文更加註重程式碼實踐,對於配置相關的知識會一筆帶過,不做過多的詳解。假設已經搭建好SpringBoot環境,下載完Moquette。至於怎麼引用Moquette,可以在原專案上修改,也可以達成Jar包新增到lib呼叫,也可以上傳到Maven私服後通過配置pom引用。筆者是上傳到Maven私服,然後通過maven匯入。
- 自定義包裝類,實現io.moquette.server.Server的自動注入
@Slf4j @Service public class MoquetteServer { @Value("${mqtt-server.config-path}") private String configFilePath; @Autowired private IAuthorizator authorizator; /** * Safety相關的攔截器,如果有其它業務,可以再去實現一個攔截器處理其它業務 */ @Autowired @Qualifier("safetyInterceptHandler") private InterceptHandler safetyinterceptHandler; private Server mqttServer; public void startServer() throws IOException { IResourceLoader configFileResourceLoader = new ClasspathResourceLoader(configFilePath); final IConfig config = new ResourceLoaderConfig(configFileResourceLoader); mqttServer = new Server(); /**新增處理Safety相關的攔截器,如果有其它業務,可以再去實現一個攔截器處理其它業務,然後也新增上即可*/ List<InterceptHandler> interceptHandlers = Arrays.asList(safetyinterceptHandler); /** * Authenticator 不顯示設定,Server會預設以password_file建立一個ResourceAuthenticator * 如果需要更靈活的連線驗證方案,可以繼承IAuthenticator介面,自定義實現 */ mqttServer.startServer(config, interceptHandlers, null, null, authorizator); } public void stop() { mqttServer.stopServer(); } }
解釋: (1)新增@Service (2)configFilePath是Moquette需要的moquette.conf的配置檔案路徑,筆者將配置檔案放到了resouces目錄下,在application.xml配置的路徑,因此通過@Value自動注入路徑值。 (3)因為是將配置檔案放在了resources目錄下,所以用Moquette提供的ClasspathResourceLoader載入的配置檔案 (4)IAuthorizator 是許可權驗證介面 (5)InterceptHandler是Moquette訂閱、釋出、建立連線等相關事件的攔截回撥業務處理邏輯介面
2.實現IAuthorizator介面
@Component
public class PermitAllAuthorizator implements IAuthorizator {
@Override
public boolean canWrite(Topic topic, String user, String client) {
/**可以控制某個使用者的client,是否具有釋出某個主題的許可權,目前預設任何Client可以釋出任主題*/
return true;
}
@Override
public boolean canRead(Topic topic, String user, String client) {
/**可以控制某個使用者的client,是否具有接收某個主題的許可權,目前預設任何Client可以接收任何主題*/
return true;
}
}
解釋:實現另一個很簡單的授權介面,允許任何使用者所有的讀寫請求
3.實現InterceptHandler介面
@Slf4j
@Component
public class SafetyInterceptHandler extends AbstractInterceptHandler{
@Override
public String getID() {
return SafetyInterceptHandler.class.getName();
}
@Override
public void onConnect(InterceptConnectMessage msg) {
}
@Override
public void onConnectionLost(InterceptConnectionLostMessage msg) {
}
@Override
public void onPublish(InterceptPublishMessage msg) {
}
@Override
public void onMessageAcknowledged(InterceptAcknowledgedMessage msg) {
}
}
解釋:1.簡單實現InterceptHandler,繼承自AbstractInterceptHandler,並重寫了部分方法。可以根據業務需要實現不同的方法。InterceptHandler介面是Moquette預留給開發者根據不同事件處理業務邏輯的介面。
4.通過SpringBoot啟動Moquette
@SpringBootApplication
public class MqttServiceApplication {
public static void main(String[] args) throws IOException {
SpringApplication application = new SpringApplication(MqttServiceApplication.class);
final ApplicationContext context = application.run(args);
MoquetteServer server = context.getBean(MoquetteServer.class);
server.startServer();
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
server.stop();
log.info("Moquette Server stopped");
}
});
}
}
如果發現MoquetteServer無法啟動,是否是SpringBoot預設的包掃描機制的問題,可以通過@ComponentScan解決。
通過以上操作,就可以在任何想要使用MoquetteServer的地方,通過@Autowired自動注入。
當然在MoquetteServer中筆者只是簡單實現了封裝,並沒有實現其它方法,讀者完全可以根據自己的需要在MoquetteServer中實現自己需要的功能,但前提是你要對Moquette的原始碼熟悉。