java 实现mqtt发送和接收消息 客户端代码

时间：2025-02-15 11:50:33

mqtt的特点就是可以用很少的网络和硬件资源来保证高并发量的数据传输，其传输的稳定性也可以手动设置Qos（消息质量）。

mqtt服务器多种多样，常见的有ActiveMqtt    EMQ  等，不过无论是什么服务器，其底层机制都是一样的。mqtt客户端可以由java、c语言等多种预言实现，我接下来就以java来示例。

1、mqtt客户端发送消息

mqtt发送消息首先要建立与服务器连接，接下来指定主题topic、消息质量qos、消息体，然后通过发送消息方法将消息发送至服务器，假如其他客户端订阅这个主题就可以接收到这个消息

发送消息客户端：

package ;

import .;
import .;
import .;
import .;
import .;
import .;
import .;
import .;
import .;



public class MyMqttClient  {

	public static MqttClient mqttClient = null;
	private static MemoryPersistence memoryPersistence = null;
	private static MqttConnectOptions mqttConnectOptions = null;
	

	static {
		init("123");
	}
	

	public static  void init(String clientId) {
		//初始化连接设置对象
		mqttConnectOptions = new MqttConnectOptions();
		//初始化MqttClient
		if(null != mqttConnectOptions) {
//			true可以安全地使用内存持久性作为客户端断开连接时清除的所有状态
			(true);
//			设置连接超时
			(30);
//			设置持久化方式
			memoryPersistence = new MemoryPersistence();
			if(null != memoryPersistence && null != clientId) {
				try {
					mqttClient = new MqttClient("tcp://127.0.0.1:1883", clientId,memoryPersistence);
				} catch (MqttException e) {
					// TODO Auto-generated catch block
					();
				}
			}else {
				
			}
		}else {
			("mqttConnectOptions对象为空");
		}
		
		(());
		//设置连接和回调
		if(null != mqttClient) {
			if(!()) {
			
//			创建回调函数对象
				MqttReceriveCallback mqttReceriveCallback = new MqttReceriveCallback();
//			客户端添加回调函数
				(mqttReceriveCallback);
//			创建连接
				try {
					("创建连接");
					(mqttConnectOptions);
				} catch (MqttException e) {
					// TODO Auto-generated catch block
					();
				}
			
			}
		}else {
			("mqttClient为空");
		}
		(());
	}
	
//	关闭连接
	public void closeConnect() {
		//关闭存储方式
		if(null != memoryPersistence) {
			try {
				();
			} catch (MqttPersistenceException e) {
				// TODO Auto-generated catch block
				();
			}
		}else {
			("memoryPersistence is null");
		}
		
//		关闭连接
		if(null != mqttClient) {
			if(()) {
				try {
					();
					();
				} catch (MqttException e) {
					// TODO Auto-generated catch block
					();
				}
			}else {
				("mqttClient is not connect");
			}
		}else {
			("mqttClient is null");
		}
	}
	
//	发布消息
	public void publishMessage(String pubTopic,String message,int qos) {
		if(null != mqttClient&& ()) {
			("发布消息   "+());
			("id:"+());
			MqttMessage mqttMessage = new MqttMessage();
			(qos);
			(());
			
			MqttTopic topic = (pubTopic);
			
			if(null != topic) {
				try {
					MqttDeliveryToken publish = (mqttMessage);
					if(!()) {
						("消息发布成功");
					}
				} catch (MqttException e) {
					// TODO Auto-generated catch block
					();
				}
			}
			
		}else {
			reConnect();
		}
		
	}
//	重新连接
	public void reConnect() {
		if(null != mqttClient) {
			if(!()) {
				if(null != mqttConnectOptions) {
					try {
						(mqttConnectOptions);
					} catch (MqttException e) {
						// TODO Auto-generated catch block
						();
					}
				}else {
					("mqttConnectOptions is null");
				}
			}else {
				("mqttClient is null or connect");
			}
		}else {
			init("123");
		}
		
	}
//	订阅主题
	public void subTopic(String topic) {
		if(null != mqttClient&& ()) {
			try {
				(topic, 1);
			} catch (MqttException e) {
				// TODO Auto-generated catch block
				();
			}
		}else {
			("mqttClient is error");
		}
	}
	
	
//	清空主题
	public void cleanTopic(String topic) {
		if(null != mqttClient&& !()) {
			try {
				(topic);
			} catch (MqttException e) {
				// TODO Auto-generated catch block
				();
			}
		}else {
			("mqttClient is error");
		}
	}

}

细心的朋友可能发现我将回调函数注释掉了，其实对于发送消息来讲，注释掉并没有太大影响，有些人可能会说需要在回调函数中编写错误重连代码，但时我觉得不是一个很好的选择，所有已经将重连机制改在发布消息之前，在上面代码中也有体现，不过如果是接收消息客户端则不能注释掉，原因是接收消息要用到回调函数，下面在写一下接收消息的客户端：

2接收消息的客户端：

package ;

import .;
import .;
import .;
import .;
import .;

import ;

public class MyMqttRecieveMessage {

	private static int QoS = 1;
	private static String Host = "tcp://127.0.0.1:1883";
	private static MemoryPersistence memoryPersistence = null;
	private static MqttConnectOptions mqttConnectOptions = null;
	private static MqttClient mqttClient  = null;
	
	public static void init(String clientId) {
		mqttConnectOptions = new MqttConnectOptions();
		memoryPersistence = new MemoryPersistence();
		ilogger ilogger = new ilogger("MyMqttRecieveMessage", "init");
		if(null != memoryPersistence && null != clientId && null != Host) {
			try {
				 mqttClient = new MqttClient(Host, clientId, memoryPersistence);
			} catch (MqttException e) {
				// TODO Auto-generated catch block
				();
			}
		}else {
			("memoryPersistence clientId Host 有空值");
		}
		
		if(null != mqttConnectOptions) {
			(true);
			(30);
			(45);
			if(null != mqttClient && !()) {
				(new MqttRecieveCallback());
				try {
					();
				} catch (MqttException e) {
					// TODO Auto-generated catch block
					();
				} 
			}else {
				("mqttClient is error");
			}
		}else {
			("mqttConnectOptions is null");
		}
	}
	
	
    public static void recieve(String topic) {
        int[] Qos = {QoS};
        String[] topics = {topic};
        ilogger ilogger = new ilogger("MyMqttRecieveMessage", "subTopic");
        if(null != mqttClient && ()) {
        	if(null!=topics && null!=Qos && >0 && >0) {
        		try {
        			(topics, Qos);
        		} catch (MqttException e) {
        			// TODO Auto-generated catch block
        			();
        		}
        	}else {
        		("there is error");
        	}
        }else {
        	init("123444");
        	recieve(topic);
        }
    }
 

}

接收消息客户端回调函数：

package ;

import .;
import .;
import .;

import ;
import ;

public class MqttRecieveCallback implements MqttCallback{

	@Override
	public void connectionLost(Throwable cause) {
		
	}

	@Override
	public void messageArrived(String topic, MqttMessage message) throws Exception {
		 ("Client 接收消息主题 : " + topic);
	     ("Client 接收消息Qos : " + ());
	     ("Client 接收消息内容 : " + new String(()));
	    
	    
	}

	@Override
	public void deliveryComplete(IMqttDeliveryToken token) {
		
	}

}

切记接收消息时，以上两个函数缺一不可！！！

下一篇讲一下mqtt客户端具体用法，附带测试方法

相关文章