分类导航

CentOS 发布时间：2022-04-02 发布网站：大佬教程 code.js-code.com

大佬教程收集整理的这篇文章主要介绍了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/woniu

 
 一、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 

  @L_439_6@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 @R_489_10495@ng HOST = "tcp://ip:1881"; 

  public static final @R_489_10495@ng TOPIC = "mqtt/topic"; 

  private static final @R_489_10495@ng clientid = "client124"; 

  private MqttClient client; 

  private MqttConnectOptions options; 

  private @R_489_10495@ng userName = "admin"; 

  private @R_489_10495@ng 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}; 

  @R_489_10495@ng[] topic1 = {TOPIC}; 

  client.subscribe(topic1,Qos); 

  ScAnner in = new ScAnner(system.in); 

  @R_489_10495@ng 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(@R_489_10495@ng[] 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(@R_489_10495@ng topic,Mqttmessage messagE) throws Exception 

{

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

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

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

  @R_489_10495@ng 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(@R_489_10495@ng[] args) throws InterruptedException,MqttException 

{

  @R_489_10495@ng serverURI = "tcp://ip:1881"; 

  @R_489_10495@ng 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); 

  @R_489_10495@ng topicStr = "mqtt/topic"; 

  MqttTopic topic = client.getTopic(topicStr); 

  @R_489_10495@ng 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,@R_489_10495@ng 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(@R_489_10495@ng topic,Mqttmessage messagE) throws Exception 

{

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

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

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

  @R_489_10495@ng 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.payload@R_489_10495@ng; 

  $('#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 中的。 

 
 四、效果篇 

  附一些需要讨论的问题： 

@H_404_1180@ 1.鉴权的考虑

@H_404_1180@ 2.点对点消息的考虑

大佬总结

以上是大佬教程为你收集整理的Centos mosquitto mqtt支持 websocket javaclient jsclient全部内容，希望文章能够帮你解决Centos mosquitto mqtt支持 websocket javaclient jsclient所遇到的程序开发问题。

如果觉得大佬教程网站内容还不错，欢迎将大佬教程推荐给程序员好友。

本图文内容来源于网友网络收集整理提供，作为学习参考使用，版权属于原作者。
如您有任何意见或建议可联系处理。小编QQ：384754419，请注明来意。

上一篇: CentOS 7 部署中文字体环境下一篇:CentOS 7中一些参数的设定

猜你在找的CentOS相关文章

Linux（CentOS 5.5）安装Apache 2.2.3 2022-05-09
ContOS7切换国内源 2019-10-31
在阿里云Centos下LNMP环境搭建 2019-10-31
CentOS7.x的DNS服务的基础配置 2019-10-31
安装VMwareTools时出错：What is the location of the directory of C header files t... 2019-10-31
CentOS的启动方式和语言设置 2019-10-31
centos，fedora 如何开机自动联网？？？ 2019-10-31
CentOS 5.5使用yum安装LAMP环境（sohu源） 2019-10-31
CentOS 6.0安装图解教程 2019-10-31
Apache与PHP的整合(编译安装)，不涉及MySQL数据库的整合 2019-10-31