Centos mosquitto mqtt支持 websocket javaclient jsclient

 
 一、mosquitto broker篇 

  1.依赖环境安装： 

 
 yum install gcc-c++ 

 
 yum install openssl-develyum install c-ares-develyum install libuuid-devle 

  注：某些系统可能自带这些依赖环境，直接过滤 

  2.websocket支持 

 
 git clone  
 https://git.oschina.net/woniu201/libwebsockets.git 

 
 安装libwebsockets需要cmake命令，需要2.8以上版本，yum安装的可能是2.6的版本，需要手动下载2.8以上的版本进行安装，请自行百度或谷歌下载，本文档使用cmake-3.2.3-Linux-x86_64 
 ，安装后直接设置环境变量即可。 

 
 1）tar -zxvf cmake-3.2.3-Linux-x86_64.tar.gz 

 
 2）解压后添加环境变量，刷新环境变量 

 
 vi /etc/profile 

 
 注：需要卸载老版本的cmake ，yum remove cmake 

 
 3）查看版本 

  cmake --version 

  4）安装websocket 

  tar -zxvf libwebsockets-v1.5-stable.tar.gz 

  cd libwebsockets/ 

  mkdir build 

  cd build/ 

  cmake .. 

  make install 

  3.mosquitto安装 

  wget wget http://mosquitto.org/files/source/mosquitto-1.4.11.tar.gz 

  tar zxvf mosquitto-1.4.11.tar.gz 

  cd mosquitto-1.4.11 

  vi config.mk 修改如下三项值 

 
 #关闭tls验证 

 
 WITH_TLS:=no 

 
 WITH_TLS_PSK:=no 

 
 ##开启websocket支持 

 
 WITH_WEBSOCKETS:=yes 

  make 

  make install 

  修改mosquitto.conf配置文件，添加websocket支持 

  cd /etc/mosquitto/ 

  cp mosquitto.conf.example mosquitto.conf 

  vi mosquitto.conf 

  修改如下参数： 

 
 listener 81 

 
 protocol websockets 

  5.启动mosquitto 

  mosquitto -c /etc/mosquitto/mosquitto.conf 

  如出现如上错误 

  ln -s /usr/local/lib/libwebsockets.so.5 /usr/lib64/libwebsockets.so.5 

 
 二、java篇 

  javaclient端使用的 eclipse.paho 包，Pom配置如下： 

  <dependency> 

  <groupId>org.eclipse.paho</groupId> 

  <artifactId>org.eclipse.paho.client.mqttv3</artifactId> 

  <version>1.1.0</version> 

  </dependency> 

  <!-- 另外大牛写的mqtt客户端 --> 

  <dependency> 

  <groupId>org.fusesource.mqtt-client</groupId> 

  <artifactId>mqtt-client</artifactId> 

  <version>1.14</version> 

  </dependency> 

 
 1.client 代码： 

  import java.util.Scanner; 

  import org.eclipse.paho.client.mqttv3.MqttClient; 

  import org.eclipse.paho.client.mqttv3.MqttConnectOptions; 

  import org.eclipse.paho.client.mqttv3.MqttException; 

  import org.eclipse.paho.client.mqttv3.MqttMessage; 

  import org.eclipse.paho.client.mqttv3.MqttTopic; 

  import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; 

/**

  * [简要描述]:<br/> 

  * [详细描述]:<br/> 

*

  * @author xiaolinlin 

  * @version 1.0,2017年6月8日 

  * @since smile V100R001C00 

*/

  public class MqttTestClient 

{

  public static final String HOST = "tcp://ip:1881"; 

  public static final String TOPIC = "mqtt/topic"; 

  private static final String clientid = "client124"; 

  private MqttClient client; 

  private MqttConnectOptions options; 

  private String userName = "admin"; 

  private String passWord = "admin123"; 

  private void start() throws MqttException 

{

try

{

  // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存 

  client = new MqttClient(HOST,clientid,new MemoryPersistence()); 

  // MQTT的连接设置 

  options = new MqttConnectOptions(); 

  // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接 

  options.setCleanSession(true); 

  // 设置连接的用户名 

  // options.setUserName(userName); 

  // 设置连接的密码 

  // options.setPassword(passWord.tocharArray()); 

  // 设置超时时间 单位为秒 

  options.setConnectionTimeout(10); 

  // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制 

  options.setKeepAliveInterval(20); 

  // 设置回调 

  MqttTopic topic = client.getTopic(TOPIC); 

  // setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息 

  options.setWill(topic,"close".getBytes(),2,true); 

  client.connect(options); 

  client.setCallback(new ClientCallback(client,options)); 

  // 订阅消息 

  int[] Qos = {1}; 

  String[] topic1 = {TOPIC}; 

  client.subscribe(topic1,Qos); 

  Scanner in = new Scanner(System.in); 

  String msg = ""; 

  MqttMessage message = null; 

  while (true) 

{

  msg = in.nextLine(); 

  if (msg.contains("colse")) 

{

  message = new MqttMessage(msg.getBytes()); 

  break; 

}

  message = new MqttMessage(msg.getBytes()); 

  topic.publish(message); 

}

  in.close(); 

}

  catch (Exception e) 

{

  e.printStackTrace(); 

}

  finally { 

  client.disconnect(); 

}

}

  public static void main(String[] args) throws MqttException 

{

  MqttTestClient client = new MqttTestClient(); 

  client.start(); 

}

}

 
 // callback 类 

  import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; 

  import org.eclipse.paho.client.mqttv3.MqttCallback; 

  import org.eclipse.paho.client.mqttv3.MqttClient; 

  import org.eclipse.paho.client.mqttv3.MqttConnectOptions; 

  import org.eclipse.paho.client.mqttv3.MqttException; 

  import org.eclipse.paho.client.mqttv3.MqttMessage; 

  import org.eclipse.paho.client.mqttv3.MqttSecurityException; 

/**

  * [简要描述]:<br/> 

  * [详细描述]:<br/> 

*

  * @author xiaolinlin 

  * @version 1.0,2017年6月8日 

  * @since smile V100R001C00 

*/

  public class ClientCallback implements MqttCallback 

{

  private MqttClient client; 

  private MqttConnectOptions options; 

  public ClientCallback(MqttClient client,MqttConnectOptions options) { 

  this.client = client; 

  this.options = options; 

}

  @Override 

  public void connectionLost(Throwable cause) 

{

  // 连接丢失后，一般在这里面进行重连 

  System.out.println("连接断开，可以做重连"); 

try

{

  client.connect(options); 

}

  catch (MqttSecurityException e) 

{

  // TODO Auto-generated catch block 

  e.printStackTrace(); 

}

  catch (MqttException e) 

{

  // TODO Auto-generated catch block 

  e.printStackTrace(); 

}

  // while(!sampleClient.isConnected()){ 

  // try { 

  // Thread.sleep(1000); 

  // sampleClient.connect(connOpts); 

  // //客户端每次上线都必须上传自己所有涉及的订阅关系，否则可能会导致消息接收延迟 

  // sampleClient.subscribe(topicFilters,qos); 

  // } catch (Exception e) { 

  // e.printStackTrace(); 

  // } 

  // } 

}

  @Override 

  public void messageArrived(String topic,MqttMessage message) throws Exception 

{

  // subscribe后得到的消息会执行到这里面 

  System.out.println("接收消息主题 : " + topic); 

  System.out.println("接收消息Qos : " + message.getQos()); 

  String msg = new String(message.getPayload()); 

  System.out.println("接收消息到服务端内容 : " + msg); 

  if(msg.contains("close")) { 

  client.close(); 

}

}

  @Override 

  public void deliveryComplete(IMqttDeliveryToken token) 

{

  System.out.println("deliveryComplete---------" + token.isComplete()); 

}

}

 
 2：server端 

  import java.io.UnsupportedEncodingException; 

  import java.util.Scanner; 

  import org.eclipse.paho.client.mqttv3.MqttCallback; 

  import org.eclipse.paho.client.mqttv3.MqttClient; 

  import org.eclipse.paho.client.mqttv3.MqttClientPersistence; 

  import org.eclipse.paho.client.mqttv3.MqttConnectOptions; 

  import org.eclipse.paho.client.mqttv3.MqttDeliveryToken; 

  import org.eclipse.paho.client.mqttv3.MqttException; 

  import org.eclipse.paho.client.mqttv3.MqttMessage; 

  import org.eclipse.paho.client.mqttv3.MqttPersistenceException; 

  import org.eclipse.paho.client.mqttv3.MqttTopic; 

  import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; 

/**

  * [简要描述]:<br/> 

  * [详细描述]:<br/> 

*

  * @author xiaolinlin 

  * @version 1.0,2017年6月8日 

  * @since smile V100R001C00 

*/

  public class MqttServer 

{

  public static void main(String[] args) throws InterruptedException,MqttException 

{

  String serverURI = "tcp://ip:1881"; 

  String clientId = "server001"; 

  MqttClient client = null; 

try

{

  // MemoryPersistence设置clientid的保存形式，默认为以内存保存 

  MqttClientPersistence persistence = new MemoryPersistence(); 

  // 异步 待研究 

  // MqttAsyncClient client = new MqttAsyncClient(serverURI,

  // clientId,persistence ); 

  client = new MqttClient(serverURI,clientId,persistence); 

  MqttConnectOptions options = new MqttConnectOptions(); 

  // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接 

  options.setCleanSession(false); 

  // options.setUserName("admin"); 

  // options.setPassword("admin123".tocharArray()); 

  // 超时时间 单位S 

  options.setConnectionTimeout(5); 

  // 设置会话心跳时间 单位S 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制 

  options.setKeepAliveInterval(10); 

  // 自动重连 

  options.setAutomaticReconnect(true); 

  MqttCallback callback = new ReciveCallback(client,options); 

  // 建立链接 

  client.connect(options); 

  client.setCallback(callback); 

  String topicStr = "mqtt/topic"; 

  MqttTopic topic = client.getTopic(topicStr); 

  String msg = ""; 

  Scanner in = new Scanner(System.in); 

  while(true) { 

  msg = in.nextLine(); 

  if(msg.contains("close")) { 

  pushMsg(topic,msg); 

  break; 

}

  pushMsg(topic,msg); 

}

  in.close(); 

}

  catch (MqttException e) 

{

  // TODO Auto-generated catch block 

  e.printStackTrace(); 

}

  finally { 

  client.close(); 

}

}

  public static void pushMsg(MqttTopic topic,String msg) 

{

  MqttDeliveryToken token; 

try

{

  MqttMessage message = new MqttMessage(); 

  message.setQos(1); 

  // 是否保留 

  message.setRetained(true); 

  byte[] payload = msg.getBytes("utf-8"); 

  // 实际消息内容 

  message.setPayload(payload); 

  token = topic.publish(message); 

  // 等待完成 

  token.waitForCompletion(); 

  // 打印完成状态 

  System.out.println(token.isComplete() + "========"); 

}

  catch (MqttPersistenceException e) 

{

  // TODO Auto-generated catch block 

  e.printStackTrace(); 

}

  catch (MqttException e) 

{

  // TODO Auto-generated catch block 

  e.printStackTrace(); 

}

  catch (UnsupportedEncodingException e) 

{

  // TODO Auto-generated catch block 

  e.printStackTrace(); 

}

}

}

 
 call back类 

  import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; 

  import org.eclipse.paho.client.mqttv3.MqttCallback; 

  import org.eclipse.paho.client.mqttv3.MqttClient; 

  import org.eclipse.paho.client.mqttv3.MqttConnectOptions; 

  import org.eclipse.paho.client.mqttv3.MqttException; 

  import org.eclipse.paho.client.mqttv3.MqttMessage; 

  import org.eclipse.paho.client.mqttv3.MqttSecurityException; 

/**

  * [简要描述]:<br/> 

  * [详细描述]:<br/> 

*

  * @author xiaolinlin 

  * @version 1.0,2017年6月8日 

  * @since smile V100R001C00 

*/

  public class ReciveCallback implements MqttCallback 

{

  public ReciveCallback(MqttClient client,MqttConnectOptions options) { 

  this.client = client; 

  this.options = options; 

}

  private MqttClient client; 

  private MqttConnectOptions options; 

  @Override 

  public void connectionLost(Throwable cause) 

{

  // 连接丢失后，一般在这里面进行重连 

  System.out.println("连接断开，可以做重连"); 

try

{

  client.connect(options); 

}

  catch (MqttSecurityException e) 

{

  // TODO Auto-generated catch block 

  e.printStackTrace(); 

}

  catch (MqttException e) 

{

  // TODO Auto-generated catch block 

  e.printStackTrace(); 

}

}

  @Override 

  public void messageArrived(String topic,MqttMessage message) throws Exception 

{

  // subscribe后得到的消息会执行到这里面 

  System.out.println("接收消息主题 : " + topic); 

  System.out.println("接收消息Qos : " + message.getQos()); 

  String msg = new String(message.getPayload()); 

  System.out.println("接收消息客户端内容 : " + new String(message.getPayload())); 

  if(msg.contains("close")) { 

  client.close(); 

}

}

  @Override 

  public void deliveryComplete(IMqttDeliveryToken token) 

{

  System.out.println("deliveryComplete---------" + token.isComplete()); 

}

}

  三、js篇 

  HTML代码： 

  <!DOCTYPE html> 

  <html> 

  <head> 

  <Meta charset="utf-8"> 

  <Meta http-equiv="X-UA-Compatible" content="IE=edge"> 

  <Meta http-equiv="Content-Type" content="text/html; charset=UTF-8"> 

  <title>聊天</title> 

  <link rel="stylesheet" href=""> 

  <script language="JavaScript" src="jquery.js"></script> 

  <script language="JavaScript" src="mqttws31-min.js"></script> 

  <script type="text/javascript"> 

  var topic = "mqtt/topic"; 

  var hostname = "120.76.216.235"; 

  var port = 81; 

  var reconnectTimeout = 2000; 

  var mqtt; 

  var options; 

  var username,password; 

  function MQTTconnect() { 

  var clientId = "ClientID_0022"; 

  //var username = 'LTAItKYvBCx06Dbz'; 

  mqtt = new Paho.MQTT.Client(hostname,port,clientId); 

  options = { 

  timeout: 3,

  // useSSL: useTLS,

  cleanSession: true,

  onSuccess: onConnect,//if true(default) the client and server persistent state is deleted on successful connect. 

  onFailure: function (message) { 

  $('#status').val("Connection Failed: " + message.errorMessage + "Retrying"); 

  setTimeout(MQTTconnect,reconnectTimeout); 

}

};

  mqtt.onConnectionLost = onConnectionLost; 

  mqtt.onMessageArrived = onMessageArrived; 

  if (username) { 

  options.userName = username; 

  options.password = password; 

}

  console.log("username=" + username + " password=" + password); 

  mqtt.connect(options); 

}

  function onConnect() { 

  $('#status').val('Connected to ' + hostname + ':' + port); 

  // Connection succeeded; subscribe to our topic 

  mqtt.subscribe(topic,{qos: 0}); 

  $('#topic').val(topic); 

}

  function onConnectionLost(response) { 

  setTimeout(MQTTconnect,reconnectTimeout); 

  $('#status').val("connection lost: " + response.errorMessage + ". Reconnecting"); 

};

  function onMessageArrived(message) { 

  var topic = message.destinationName; 

  var payload = message.payloadString; 

  $('#ws').prepend('<li>' + topic + ' = ' + payload + '</li>'); 

};

  function button_onclick(){ 

  //这里写你要执行的语句 

  var tp = $('#sendtopic').val(); 

  var val = $('#textsend').val(); 

  if(tp==''||val=='') 

{

  alert("no aaa"); 

  return; 

}

  var message = new Paho.MQTT.Message(val); 

  message.destinationName = tp; 

  message.qos=0; 

  mqtt.send(message); 

  // mqtt.publish(tp,val); 

  $('#ws').prepend('<li>' + tp + ' = ' + val+ '</li>'); 

};

  $(document).ready(function() { 

  MQTTconnect(); 

});

  </script> 

  </head> 

  <body> 

  <h1>Mosquitto Websockets</h1> 

  <div> 

  <div>Subscribed to <input type='text' id='topic' disabled /> 

  Status: <input type='text' id='status' size="80" disabled /> 

  </br> 

  publish to <input type='text' id='sendtopic' /> text <input type='text' id='textsend' size="80"/> 

  Status: <input type='button' value="send" id='btn' onclick="javascript:button_onclick()"/></div> 

  <ul id='ws' style="font-family: 'Courier New',Courier,monospace;"></ul> 

  </div> 

  </body> 

  </html> 

  引用的js自行下载，其中mqttws31-min.js 也是属于eclipse.paho 中的。 

 
 四、效果篇 

  附一些需要讨论的问题： 

  1.鉴权的考虑 

  2.点对点消息的考虑 

  3.性能考虑 

Centos mosquitto mqtt支持 websocket javaclient jsclient

猜你在找的CentOS相关文章