转载

搭建IM服务 so easy

现在很多网站、APP都通过IM服务来实现消息推送及数据即时同步功能,即时通讯组件逐渐成为产品的标配。目前国内有很多成熟稳定的第三方即时通讯服务厂家,比如: 融云 。使用这些专业的服务可以提高开发效率而且服务稳定有保障。

如果自己DIY或者需要在封闭的局域网内使用IM服务,该怎么办呢?下文就简单介绍一下 EasyPM 曾经实践过的自行搭建IM服务过程。

数据同步方式

实现数据同步,有"推"、"拉" 两种思路,具体有以下几种方式:

  1. 使用HTTP轮循方式

    • 说明:定时向HTTP服务端接口(Web Service API)获取最新消息,可结合ajax技术实现页面无刷新效果,这是主动拉取消息的机制。

    • 优点:实现简单、可控性强、部署成本低

    • 缺点:实时性差,增加服务端负载

  2. 使用XMPP协议

    • 说明:XMPP是基于可扩展标记语言(XML)的协议,它用于即时消息(IM)以及在线现场探测。它促进在服务器之间的准即时操作,其前身是Jabber,是一个开源形式组织产生的网络即时通信协议。XMPP目前被IETF国际标准组织完成了标准化工作。

    • 优点:协议成熟、强大、可扩展性强、目前主要应用于众多IM系统

    • 缺点:协议比较复杂、冗余(基于XML)、费流量

  3. 使用MQTT协议

    • 说明:MQTT(Message Queuing Telemetry Transport,消息队列遥测传输)是IBM开发的一个即时通讯协议,该协议支持所有平台,是轻量级的、基于代理的“发布/订阅”模式的消息传输协议

    • 优点:MQTT协议简洁、可扩展性强、流量开销很小、网络传输时间短

    • 缺点:还不够成熟、实现较复杂

EasyPM 作为一个团队协作应用,消息/数据推送功能是不可或缺的。因为MQTT比较轻量且网络开销小等特点,我们选择了支持MQTT协议的Apollo。

Apollo是什么?

Apollo是apache旗下的基金项目,它是以Apache ActiveMQ5.x为基础,采用全新的线程和消息调度架构重新实现的消息中间件,针对多核处理器进行了优化处理,它的速度更快、更可靠、更易于维护。apollo与ActiveQQ一样支持多协议:STOMP、AMQP、MQTT、Openwire、 SSL、WebSockets,本文只介绍MQTT协议的使用。

下载Apollo

进入 Apollo下载页面 ,选择下载合适的版本

如果操作是系统是Windows Vista或更高版本,则需要安装Microsoft Visual C++ 2010 Redistributable:

64位JVM

32位JVM

创建Apollo实例及服务

  1. 创建实例

    进入 E:/apache-apollo-1.7 之下的bin目录,打开cmd窗口,执行命令: apollo create D:/apollo_broker ,命令执行成功后,在D盘下会有 apollo_broker 目录,这便是apollo的服务实例,apollo之旅便从这里开始。

D:/apollo_broker 下有个bin目录,其中有两个文件:

apollo-broker.cmd是通过cmd命令启动apollo服务的

apollo-broker-service.exe,是用于创建window服务的

  1. 命令行启动服务

    D:/apollo_broker/bin 目录下打开cmd窗口,执行 apollo-broker run 命令来启动apollo服务,

启动成功可以在浏览器中查看运行情况,访问地址为 http://127.0.0.1:61680 , 默认用户名/密码: admin/password

  1. 创建windows服务

    找到 cmd.exe 文件,点击鼠标右键,以管理员身份运行,输入创建windows服务命令,如下图:

MQTT协议的应用

MQTT协议有众多客户端实现,相关客户端请参考 apollo官方文档

本文采用eclipse的 paho客户端 实现

web端接收消息介绍

将 javascript客户端 项目下载下来,并在其项目根目录下执行mvn命令,进行编译,生成target目录,其下生成mqttws31.js、mqttws31-min.js两个js文件,将其拷贝到自己项目相关目录下,并在页面中引用,即可实现javascript客户端的消息订阅和发布,demo代码如下:

var client = new Paho.MQTT.Client(location.hostname, 61623,"/", "clientId");  /* 61623是ws连接的默认端口,可以在apollo中间件中进行配置 (关于apollo的配置请参考:  http://activemq.apache.org/apollo/documentation/user-manual.html ) */ // set callback handlers  client.onConnectionLost = onConnectionLost;  client.onMessageArrived = onMessageArrived;  // connect the client  client.connect({userName:'admin',password:'password',onSuccess:onConnect});  // called when the client connects  function onConnect() { // 连接成功后的处理          // Once a connection has been made, make a subscription and send a message.          console.log("onConnect");          client.subscribe("/topic/event"); // 订阅消息的主题          var message = new Paho.MQTT.Message("Hello,this is a test");          message.destinationName = "/topic/event";          client.send(message); // 发送消息  }  // called when the client loses its connection  function onConnectionLost(responseObject) { // 连接丢失后的处理          if (responseObject.errorCode !== 0) {                  console.log("onConnectionLost:"+responseObject.errorMessage);          }  }  // called when a message arrives  function onMessageArrived(message) { // 消息接收成功后的处理          console.log("onMessageArrived:"+message.payloadString);  }

服务端消息发送介绍

paho java客户端 目前只支持J2SE和安卓,提供 源码下载 和maven库。

我们采用maven库,其地址如下:

Official Releases

Nightly Snapshots

maven dependency配置:

<dependency>          <groupId>org.eclipse.paho</groupId>          <artifactId>org.eclipse.paho.client.mqttv3</artifactId>          <version>1.0.1</version>  </dependency>

java实现代码:

        String topic        = "MQTT Examples";          String content      = "Message from MqttPublishSample";         int qos             = 2;         String broker       = "tcp://127.0.0.1:61613";         String clientId     = "JavaSample";         MemoryPersistence persistence = new MemoryPersistence();         try {             MqttClient sampleClient = new MqttClient(broker, clientId, persistence);             MqttConnectOptions connOpts = new MqttConnectOptions();             connOpts.setCleanSession(true);             System.out.println("Connecting to broker: "+broker);             sampleClient.connect(connOpts);             System.out.println("Connected");             System.out.println("Publishing message: "+content);             MqttMessage message = new MqttMessage(content.getBytes());             message.setQos(qos);             sampleClient.publish(topic, message);             System.out.println("Message published");             sampleClient.disconnect();             System.out.println("Disconnected");             System.exit(0);         } catch(MqttException me) {             System.out.println("reason "+me.getReasonCode());             System.out.println("msg "+me.getMessage());             System.out.println("loc "+me.getLocalizedMessage());             System.out.println("cause "+me.getCause());             System.out.println("excep "+me);             me.printStackTrace();         }

小结

至此,MQTT协议已部署完毕,java端可以发布消息,而javascript端则可以订阅并接收到java端发布的信息。在搭建的过程中,可以参考以下资源:

原文  https://segmentfault.com/a/1190000004906137
正文到此结束
Loading...