NodeJs|基于nodejs和activeMQ的消息推送 NodeJs|ActiveMQ

好久没来写博客了，感觉自己堕落了，哈哈，LZ糊里糊涂的又换了家单位，空余时间研究了一下nodejs，顺势搞了这个demo，今天来聊聊基于nodejs和activeMQ的消息推送，内容不算复杂，新手也能一看即会。
首先介绍下一点背景，为什么搞这个东西，LZ上家公司是做监控项目的，很多告警都要实时推送到web端，以前的技术架构是flex+corba+mq(自己封装的)，早期B/S架构的实时推送无非就两种，一个是基于插件的长连接，另外一个就是轮训或者Comet，前者受限于flash技术的衰败肯定会逐渐退出历史舞台，后者的低效率在数据量较大的情况下有随时崩溃的可能。随着html5的web socket技术的产生，很多前端socket技术，例如socket.io逐步得到重用，配合nodejs使用，一个前后端socket连接就可以很轻松地解决问题。
本篇使用技术socket.io+nodeJs+stomp+activeMQ，具体的技术简介就不贴了，各位可以在网络上轻松找到，这里直接代码开路。首先是页面index.htm

#messages { padding: 0px; list-style-type: none; } #messages li { padding: 2px 0px; border-bottom: 1px solid #ccc; }

【NodeJs|基于nodejs和activeMQ的消息推送】代码中的/socket.io/socket.io.js，不是真实存在的，是nodejs加载的socket.io模块，这个插件需要在nodejs中进行添加，命令如下：

npm install socket.io

接下来是nodejs端代码，运行该代码前，要首先启动activeMQ，保证activeMQ，监听了stomp协议和61613端口，LZ在这方面吃了亏，浪费了好长时间，配置文件目录：apache-activemq-5.8.0\conf\activemq.xml

var Stomp = require('stomp-client'); var destination = '/topic/myTopic'; var client = new Stomp('127.0.0.1', 61613, 'user', 'pass'); var fs = require('fs'), http = require('http'), sio = require('socket.io'); var server = http.createServer(function(req, res) { res.writeHead(200, { 'Content-type': 'text/html' }); res.end(fs.readFileSync('./index.htm')); }); server.listen(8888, function() { console.log('Server listening at http://localhost:8888/'); }); // Attach the socket.io server io = sio.listen(server); var tempSockets = []; io.sockets.on('connection', function(socket) { console.log('socket connection success!!!'); tempSockets.push(socket); socket.on('disconnect',function(){ }); }); client.connect(function(sessionId) { client.subscribe(destination, function(body, headers) { console.log('From MQ:', body); //tempSocket.broadcast.emit('message', body); tempSockets.forEach(function(socket) { if(socket.connected) { socket.send(body); } }) }); var i = 0; function publish(){ setTimeout(function(){ client.publish(destination, 'Hello World!'+(i++)); //publish(); },1000); } publish(); });

文中的stomp-client模块也需要nodejs添加

npm install stomp-client

上面代码中，client.publish(destination, 'Hello World!'+(i++))，是nodejs作为提供者向activeMQ发送消息，实际应用中是java端的提供者，代码如下：

package com.feng.activemq; import javax.jms.Connection; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; /** * @author songfeng * @version 1.0. */ public class SendMessage { private static final String url = "tcp://localhost:61616"; private static final String QUEUE_NAME = "choice.queue"; private static final String TOPIC_NAME = "myTopic"; protected String expectedBody = "world!"; public void sendMessage() throws JMSException { Connection connection = null; MessageProducer producer = null; Session session = null; try { ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url); connection = (Connection) connectionFactory.createConnection(); connection.start(); session = (Session) connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //Destination destination = session.createQueue(QUEUE_NAME); Destination destination = session.createTopic(TOPIC_NAME); producer = session.createProducer(destination); for(int i = 1 ; i <= 20; i++) { TextMessage message = session.createTextMessage(expectedBody+i); message.setStringProperty("headname", "remoteB"); producer.send(message); Thread.sleep(1000l); } } catch (Exception e) { e.printStackTrace(); } finally { producer.close(); session.close(); connection.close(); } } public static void main(String[] args) { SendMessage sendMsg = new SendMessage(); try { sendMsg.sendMessage(); } catch (Exception ex) { System.out.println(ex.toString()); } } }

是不是很简单，重要的是研究的过程，哈哈，继续撸码。。。