大数据实战之环境搭建(十)

1. Html5, 云计算，移动互联网，大数据你知道多少，若是不知道多少，请抓紧时间学习。

今天要说的是消息队列ActiveMQ，这个和cassandra同样也是apache的产品，开源的，支持不少客户端，好比java，C#，ruby，python等。ActiveMQ 是一个彻底支持JMS和J2EE规范的 JMS Provider实现。在介绍jms以前，先要介绍一下它的规范，俗话说无规矩不成方圆，就像序列化和反序列化同样，你们要遵循必定的规则才能交互。

JMS的基本构件有

(1).链接工厂， 是客户端用来建立链接的对象，ActiveMQConnectionFactory类。

(2). 链接，JMS Connection封装了客户端和JMS提供者之间的一个虚拟链接，ActiveMQConnectio类。

(3). 会话，JMS Session是生产者和消费者之间Produce message和Consume message的上下文。

会话用于建立消息的生产者，消费者和消息。这个上下文保持了会话的事务性，即发送消息和接受消息被组合到了一个事务中。

(4). 目的地，这个目的地代表了生产消息的目标和消费消息的目标，Destination对象。

JMS规范中定义了两种消息传递域，一种是点对点，一种是发布/订阅。点对点的特色是一个消息只能有一个消费者，消费者和生产者之间没有时间上的相关性，不管消费者在生产者发送消息的时候是不是出于运行状态，他均可以提取到消息。订阅发布模式的特色是一个生产者能够有多个消费者，这种模式存在时间上的相关性，消费者只能消费自从他订阅以后的消息，以前的消息是不能消费的。可是JMS规范容许客户建立持久订阅，即便是在生产者在消费者发送消息的时候处于一个未激活的状态，等他激活之后，依然能够消费未激活以前生产者发送的消息

(5).生产者，是由会话建立的一个对象，它主要职责是将消息发送到目的地

(6).消费者，是由会话建立的一个对象，它主要是接收生产者发送到目的地的消息

消费者能够同步消费和异步消费。

OK，概念就先讲到这里，咱们看一下环境

首先从apache 网站上下载ActiveMQ linux版本

在这里我下载的版本是5.8.0。OK咱们把它放到FTP Server上，在CentOS上拷贝至opt文件夹。

ok咱们用命令解压

解压完成后，直接运行启动脚本

[root@bogon opt]# ls
apache-activemq-5.8.0  apache-activemq-5.8.0-bin.tar.gz
[root@bogon opt]# cd apache-activemq-5.8.0
[root@bogon apache-activemq-5.8.0]# ls
activemq-all-5.8.0.jar  docs     NOTICE           webapps-demo
bin                     example  README.txt       WebConsole-README.txt
conf                    lib      user-guide.html
data                    LICENSE  webapps
[root@bogon apache-activemq-5.8.0]# cd bin
[root@bogon bin]# ls
activemq        activemq.jar  linux-x86-32  macosx
activemq-admin  diag          linux-x86-64  wrapper.jar
[root@bogon bin]# sh activemq

OK，开始启动，到这一步启动成功

咱们看一下它的管理界面，默认端口是8161，在conf目录下的jetty.xml中

<property name="connectors">
<list>
<bean id="Connector" class="org.eclipse.jetty.server.nio.SelectChannelConnector">
<property name="port" value="8161" />
</bean>
<!--
Enable this connector if you wish to use https with web console
-->
<!--
<bean id="SecureConnector" class="org.eclipse.jetty.server.ssl.SslSelectChannelConnector">
<property name="port" value="8162" />
<property name="keystore" value="file:${activemq.conf}/broker.ks" />
<property name="password" value="password" />
</bean>
-->
</list>
</property>

OK，咱们在浏览器中输入Http://localhost:8161/admin，提示输入用户名和密码，用户名:admin，密码admin，

这个用户名和密码能够在conf/jetty-realm.properties文件中去修改。

OK，管理界面以下

由于哥一直作.net开发，因此仍是以.net来作个demo，来作个消息的发送可接收。首先须要下载.net客户端。去网站上找到下载.net 客户端的地方

在这里我下载的是1.5.3版本，由于1.6.0版本链接有问题。下载下来以后，咱们使用.net 4.0版本

OK，新建一个solution，建立两个WinForm的工程，以下

注意要引用上面的两个dll，这两个dll分别在build和lib下

咱们看一下代码，没什么好解释的，先看生产者代码

using Apache.NMS;
using Apache.NMS.ActiveMQ;
using Apache.NMS.ActiveMQ.Commands;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Windows.Forms;
namespace ActiveMQSender
{
public partial class FrmMsgSender : Form
{
IConnectionFactory factory;
IConnection connection;
ISession session;
IDestination destination;
IMessageProducer prod;
ITextMessage streamMessage;
~FrmMsgSender()
{
session.Close();
connection.Close();
}
public FrmMsgSender()
{
InitializeComponent();
this.Init();
}
private void Init()
{
factory = new ConnectionFactory("tcp://192.168.192.128:61616/");
connection = factory.CreateConnection("admin", "admin");
session = connection.CreateSession();
destination = new ActiveMQTopic("Bruce Test");
prod = session.CreateProducer(destination);
streamMessage = prod.CreateTextMessage();
}
private void btnSend_Click(object sender, EventArgs e)
{
streamMessage.Text=txtSendMsg.Text.Trim();
prod.Send(streamMessage,MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue);
}
}
}

再看消费者代码

using Apache.NMS;
using Apache.NMS.ActiveMQ;
using Apache.NMS.ActiveMQ.Commands;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Windows.Forms;
namespace FrmMsgReceiver
{
public partial class FrmMsgReceiver : Form
{
IConnectionFactory factory;
IConnection connection;
ISession session;
IDestination destination;
IMessageConsumer consumer;
~FrmMsgReceiver()
{
session.Close();
connection.Close();
}
public FrmMsgReceiver()
{
InitializeComponent();
}
private void Init()
{
factory = new ConnectionFactory("tcp://192.168.192.128:61616/");
connection = factory.CreateConnection("admin", "admin");
session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
destination = new ActiveMQTopic("Bruce Test");
consumer = session.CreateConsumer(destination);
}
private void FrmMsgReceiver1_Load(object sender, EventArgs e)
{
this.Init();
IMessage streamMessage = consumer.Receive();
this.txtReceiveMsg.Text = (streamMessage as ActiveMQStreamMessage).ReadString();
}
}
}

OK，就是这么简单，我也不写运行过程了，启动VMPlayer后，机器的内存已经使用了90%了，可怜的dell 1420。

若是你们有兴趣的话，看一下这篇文章

http://blog.csdn.net/bodybo/article/details/5647968。