Java使用easymqtt4j快速开发工业级mqtt企业级应用

Wesley13
• 阅读 667

Java使用easymqtt4j快速开发工业级mqtt企业级应用

easymqtt4j , netty +mqtt +subscriber+ publisher +broker+cluster server for java

easymqtt4j特点:

1、spring integration 集成模式,自由灵活。

2、完全支持mqtt 3.1、3.1.1国际标准协议,支持tcp\websocket等等,可配置。

3、客户端完全支持接入主流broker服务如:Eclipse Paho,Mosquitto,JBoss A-MQ 6.1, Apache ActiveMQ 5.10-SNAPSHOT,Apache Camel 2.13.0,HiveMQ,EMQ,mosquitto,moquette,JMQTT,mqttwk等等。

4、发布、订阅接口简单&统一Gateway。

5、完全支持event事件EventGateway,灵活自由控制。

6、支持handleEvent、connectionLost、 messageArrived、deliveryComplete。

7、支持preSend、postSend、afterSendCompletion。

8、支持preReceive、postReceive、afterReceiveCompletion。

使用方法&步骤:

1、引用jar

2、实现 MqttSubscriberGateway消息队列订阅 接口 

3、实现 MqttEventGateway 事件 接口 

4、MqttPublisherGateway消息发送 接口 ( 使用请参考 MqttScheduleTask 消息定时发送)

 项目开源地址1: https://github.com/zengfr/easymqtt4j

项目开源地址2:https://gitee.com/zengfr/easymqtt4j

测试代码 :   https://github.com/zengfr/easymqtt4j/tree/master/easymqtt4j-test/src/main/java/com/zengfr/easymqtt4j/test

easymqtt4j-client com.zengfr.easymqtt4j ${project.version}

spring.mqtt.host.username=admin spring.mqtt.host.password=password #spring.mqtt.host.uris=ws://api.easylink.io:1983 spring.mqtt.host.uris=tcp://147.14.141.51:1883 # spring.mqtt.subscriber.id=subscriberId123 spring.mqtt.subscriber.topics=topic/#,testtopic/# spring.mqtt.subscriber.completionTimeout=3000 # spring.mqtt.publisher.id=publisherId456 spring.mqtt.publisher.defaulttopic=topic0 spring.mqtt.publisher.completionTimeout=3000

import com.zengfr.easymqtt4j.client.geteway.MqttSubscriberGateway; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.messaging.Message; import org.springframework.stereotype.Component; _/** _ * Created by zengfr on 2020/5/12. */ @Component public class MqttSubscriberGatewayImpl implements MqttSubscriberGateway { static Logger logger = LoggerFactory.getLogger(MqttSubscriberGatewayImpl.class); @Override public boolean handlerMqttMessage(Message<?> msg, String topic, String qos, String id, String timestamp) { logger.info(String.format("收到:%s", msg)); return false; } }

import com.zengfr.easymqtt4j.client.geteway.MqttPublisherGateway; import com.zengfr.easymqtt4j.client.util.MqttUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; _/** _ * Created by zengfr on 2020/5/12. */ @Configuration @EnableScheduling @EnableAsync public class MqttScheduleTask { private static Logger logger = LoggerFactory.getLogger(MqttScheduleTask.class); @Autowired private MqttPublisherGateway publisherGateway; @Scheduled(fixedRate = 1000*1,initialDelay = 1000*5) public void sendMqtt1() { logger.info("发送开始"); String d= MqttUtil.getNowString(); publisherGateway.publish("testtopic/0", 0, "hello 10 " + d); logger.info("发送结束"); }

@Scheduled(fixedRate \= 1000\*3,initialDelay \= 1000\*5)
public void sendMqtt2() throws InterruptedException {
    int count = 11;

for (int i = 0; i < count; i++) { logger.info("发送开始"); String d= MqttUtil.getNowString(); publisherGateway.publish("topic/"+i, i%3, "hello 00 " + d); publisherGateway.publish("testtopic/"+i, i%3, "hello 10 " + d); logger.info("发送结束"); } } }

import com.zengfr.easymqtt4j.client.geteway.MqttEventGateway; import com.zengfr.easymqtt4j.client.util.MqttMsgUtil; import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.context.ApplicationEvent; import org.springframework.integration.mqtt.event.MqttIntegrationEvent; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.stereotype.Component; _/** _ * Created by zengfr on 2020/5/12. */ @Component public class MqttEventGatewayImpl implements MqttEventGateway { static Logger logger = LoggerFactory.getLogger(MqttEventGatewayImpl.class); @Override public void handleEvent(MqttIntegrationEvent event) {

    _logger_.info("event:               {}", event);

}

@Override

public void handleEvent(ApplicationEvent event) {

    _logger_.info("event:               {}", event);

}

@Override

public void connectionLost(String clientId, Throwable cause) {

    _logger_.info("connectionLost:      {} {}", clientId, cause);

}

@Override

public void messageArrived(String clientId, String topic, MqttMessage message) { logger.info("messageArrived: {} {} {}", clientId, topic, message); }

@Override

public void deliveryComplete(String clientId, IMqttDeliveryToken token) { try { logger.info("deliveryComplete: {} {}", clientId, MqttMsgUtil.tokenToString(token)); } catch (MqttException e) { logger.error("error", e); } }

@Override

public void preSend(String clientId, Message<?> message, MessageChannel channel) {

}

@Override

public void postSend(String clientId, Message<?> message, MessageChannel channel, boolean sent) {

}

@Override

public void afterSendCompletion(String clientId, Message<?> message, MessageChannel channel, boolean sent, Exception ex) {

}

@Override

public boolean preReceive(String clientId, MessageChannel channel) { return false; }

@Override

public Message postReceive(String clientId, Message message, MessageChannel channel) { return null; }

@Override

public void afterReceiveCompletion(String clientId, Message<?> message, MessageChannel channel, Exception ex) {

}

}

Java使用easymqtt4j快速开发工业级mqtt企业级应用

其他 参考 关键字:

MQTT的集成和使用基于MQTT协议在物联网系统中数据交互的重要角色,入门java项目中mqtt初始Java
Java连接MQTT 订阅和发布 Java JAVA开发MQTT总结
所以写的一个demo,在这里记录下来,方便有人使用的时候查阅,不涉及mqtt的具体讲解,只是贴代码和运行过程
java 实现mqtt发送和接收消息 客户端代码 Java
mqtt的特点就是可以用很少的网络和硬件资源来保证高并发量的数据传输,其传输的稳定性也可以手动设置Qos(消息质量)。
mqtt服务器多种多样,常见的有ActiveMqtt    EMQ  等,不过无论是什么服务器,其底层机制都是一样的。
mqtt客户端可以由java、c语言等多种预言实现,java来示例
MQTT简单demo(java)   简单的谈了一些MQTT协议的一些知识, 知识具体的Java实现

具体 配置参考:

https://gitee.com/zengfr/easymqtt4j/tree/master/easymqtt4j-client/src/main/java/com/zengfr/easymqtt4j/client/config

点赞
收藏
评论区
推荐文章
blmius blmius
3年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
待兔 待兔
4个月前
手写Java HashMap源码
HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22
Jacquelyn38 Jacquelyn38
3年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
Wesley13 Wesley13
3年前
Java爬虫之JSoup使用教程
title:Java爬虫之JSoup使用教程date:201812248:00:000800update:201812248:00:000800author:mecover:https://imgblog.csdnimg.cn/20181224144920712(https://www.oschin
Wesley13 Wesley13
3年前
Java日期时间API系列31
  时间戳是指格林威治时间1970年01月01日00时00分00秒起至现在的总毫秒数,是所有时间的基础,其他时间可以通过时间戳转换得到。Java中本来已经有相关获取时间戳的方法,Java8后增加新的类Instant等专用于处理时间戳问题。 1获取时间戳的方法和性能对比1.1获取时间戳方法Java8以前
Wesley13 Wesley13
3年前
Java日期时间API系列36
  十二时辰,古代劳动人民把一昼夜划分成十二个时段,每一个时段叫一个时辰。二十四小时和十二时辰对照表:时辰时间24时制子时深夜11:00凌晨01:0023:0001:00丑时上午01:00上午03:0001:0003:00寅时上午03:00上午0
Wesley13 Wesley13
3年前
00:Java简单了解
浅谈Java之概述Java是SUN(StanfordUniversityNetwork),斯坦福大学网络公司)1995年推出的一门高级编程语言。Java是一种面向Internet的编程语言。随着Java技术在web方面的不断成熟,已经成为Web应用程序的首选开发语言。Java是简单易学,完全面向对象,安全可靠,与平台无关的编程语言。
Stella981 Stella981
3年前
Docker 部署SpringBoot项目不香吗?
  公众号改版后文章乱序推荐,希望你可以点击上方“Java进阶架构师”,点击右上角,将我们设为★“星标”!这样才不会错过每日进阶架构文章呀。  !(http://dingyue.ws.126.net/2020/0920/b00fbfc7j00qgy5xy002kd200qo00hsg00it00cj.jpg)  2
Wesley13 Wesley13
3年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
10个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这