消息中间件之ActiveMQ初步认识及第一个程序（二）

1 ActiveMQ 简介

ActiveMQ是Apache出品，最流行的，能力强劲的开源消息总线。

ActiveMQ是一个完全支持JMS1.1和J2EE 1．4规范的JMS Provider现，尽管JMS规范出台己经是很久的事情了，但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位，可以说ActiveMQ在业界应用最广泛，当然如果想要有更强大的性能和海量数据处理能力，ActiveMQ还需要不断的升级版本，80%以上的业务我们使用ActiveMQ以及足够满足需求，当然后续如天猫、淘宝网这种大型的电商网站，尤其是双11这种特殊时间，ActiveMQ需要进行很复杂的优化源码以及架构设计才能完成，后面我还会介绍更强大的分布式消息中间件，RocketMQ，可以说ActiveMQ是核心，是基础，所以我们必须要掌握好。

1 ActiveMQ Windows 安装说明

由于这个是windows的安装这里我不做太多说明。

下载链接（目前是最新版）：http://activemq.apache.org/activemq-5152-release.html

安装完后的目录：



这时我们启动ActiveMQ,进入bin目录一个32位的一个64位的，我的机子是64位的，打开64位的文件夹找的activemq.bat双击启动。

默认的范围问路径是：http://localhost:8161 注意这里端口是8161这个可以自己改，这时你在浏览器访问时，会弹出一个登陆的界面，默认用户名密码都是

admin

。

登陆进去如图：



[b]1.1 如何修改端口？[/b]

打开conf目录找到jetty.xml文件，打开文件，大概在110行左右。

<bean id="jettyPort" class="org.apache.activemq.web.WebConsolePort" init-method="start">
<!-- the default port number for the web console -->
<property name="host" value="0.0.0.0"/>
<!--改成一个自己电脑不占用的端口（其实改不改都无所谓，自己知道有这回事就好）-->
<property name="port" value="8161"/>
</bean>

[b]1.2 如何修改用户名跟密码？[/b]

打开conf目录找到jetty-realm.properties文件，打开文件，看到底部，改成自己用户名密码即可（其实改不改都无所谓，自己知道有这回事就好）。

记得如果你改了，一定要重新启动 ！！！

2 ActiveMQ Hello World

我们首先写一个简单的HelloWorld示例，让大家感受下ActiveMQ ,我们的要实现接受者和发送者两部分代码的编写。

步骤如下(这个步骤都是固定的)：Sendert/Receivert

第一步：建立ConnectionFactory工厂对象，需要填入用户名、密码、以及要连接的地址，均使用默认即可，默认端囗为“tcp://localhost:61616”

第二步，通过ConnectionFactory工厂对象我们创建一个Connection连接，并且调用Connection的start方法开启连接，Connection默认是关闭的。

第三步，过Connection对象创建Session会话（上下文环境对象），用于接收消息，参数配置1为是否启用是事务，参数配置2为签收模式，一般我们设置自动签收。

第四步：通过Session创建Destination对象,指的是一个客户端用来指定生产消息目标和消费消息来源的对象，在PTP模式中，Destination被称作Queue即队列：在Pub/Sub模式，Destination被称作Topic及主题。在程序中可以使用多个Queue和Topic。

第五步：我们需要通过Session对象创建消息的发送和接收对象（生产者和消费者） MessageProducer/MessageConsumer。

第六步：我们可以使用MessageProducer的setDeIiveryMode方法为其设置持久化特性和非持久化特性（DeliveryMode），消费端这部分不用写，我们稍后详细介绍。

第七步：最后我们使用JMS范的TextMessage形式创建数据（通过Session对象），并用MessageProducer的send方法发送数据同理客户端使用receive方法进行接收数据。最后不要忘记关闭Connection连接

生产端：

public class Sender {
public static void main(String[] args) throws JMSException {

//1 建立ConnectionFactory工厂对象，需要填入用户名、密码、以及要连接的地址，均使用默认即可，默认端囗为“tcp://localhost:61616”
// 由于这里我们还没有设置 用户名 密码 这里使用默认，在下一节我会介绍这个用户名密码如何设置
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnectionFactory.DEFAULT_USER,
ActiveMQConnectionFactory.DEFAULT_PASSWORD,
"tcp://localhost:61616"
);

//2通过ConnectionFactory工厂对象我们创建一个Connection连接，并且调用Connection的start方法开启连接，Connection默认是关闭的。
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();

//3 过Connection对象创建Session会话（上下文环境对象），用于接收消息，参数配置1为是否启用是事务，参数配置2为签收模式，一般我们设置自动签收。
Session session = connection.createSession(Boolean.FALSE , Session.AUTO_ACKNOWLEDGE);

//4通过Session创建Destination对象,指的是一个客户端用来指定生产消息目标和消费消息来源的对象，在PTP模式中，Destination被称作Queue即队列：在Pub/Sub模式，Destination被称作Topic及主题。在程序中可以使用多个Queue和Topic。
Destination destination = (Destination) session.createQueue("queue");

//5我们需要通过Session对象创建消息的发送和接收对象（生产者和消费者） MessageProducer/MessageConsumer。
MessageProducer producer = session.createProducer(destination);

//6我们可以使用MessageProducer的setDeIiveryMode方法为其设置持久化特性和非持久化特性（DeliveryMode），消费端这部分不用写，我们稍后详细介绍。
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

//7最后我们使用JMS范的TextMessage形式创建数据（通过Session对象），并用MessageProducer的send方法发送数据同理客户端使用receive方法进行接收数据。
for (int i = 0; i < 10; i++) {
TextMessage textMessage = session.createTextMessage();
textMessage.setText("I Tell You >>> "+i);
producer.send(textMessage);
}
//最后不要忘记关闭Connection连接
if (connection!=null){
connection.close();
}

}
}

消费端：

public class Receiver {

public static void main(String[] args) throws  Exception{

//1
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnectionFactory.DEFAULT_USER,
ActiveMQConnectionFactory.DEFAULT_PASSWORD,

fa1f
"tcp://localhost:61616"
);

//2
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();

//3
Session session = connection.createSession(Boolean.FALSE , Session.AUTO_ACKNOWLEDGE);

//4
Destination destination = (Destination) session.createQueue("queue");

//5
MessageConsumer producer = session.createConsumer(destination);

//7
while(true){

TextMessage receive = (TextMessage) producer.receive();
if (receive==null) break;
System.out.println("收到的内容 >>>  " +receive.getText());
}

if (connection!=null){
connection.close();
}
}
}

启动生产端，在浏览器访问：http://localhost:8161/admin/queues.jsp 查看我们的数据,如图：



好我们启动消费端，控制台打印如下：



再次访问http://localhost:8161/admin/queues.jsp，如图（注意参数的变化）：

3 ActiveMQ 安全机制

在安装时候我已经教大家如何修改那个ActiveMQ的Web管理界面的密码了，在这不说了。

在这主要说的是ActiveMQ应该设置有安全机制，只有符合验证的用户才能进行发送和获取消息。其实刚刚在Hello Wolrd 第一步中我就给大家提示过用户名跟密码使用的是默认的，现在我教大家如何改成自己的。

在conf目录下找到acivemq.xml,打开文件，在大概126行中加入(在broker结束标签上)：

<plugins>
<simpleAuthenticationPlugin>
<users>
<authenticationUser username="hfbin" password="hfbin" groups="users,admins"/>
</users>
</simpleAuthenticationPlugin>
</plugins>

这个时候你改了这里，那个生产/消费端中第一步就不能使用默认用户名和密码了，更改如下：

ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
"hfbin",
"hfbin",
"tcp://localhost:61616"
);

4 方法使用

[b]4.1 Conection方法使用[/b]

在成功创建正确的ConectionFactory后，下一步将是创建一个连接，它是JMS定义的一个接囗。ConectionFactory负责返回可以与底层消息传递系统进行通信的 Connectionn实现。通常客户端只使用单一连接。根据JMS文档，Connection的目的是“利用JMS提供者封装开放的连接”，以及表示“客户端与提供者服务例程之间的开放TCP/IP套接字”。该文档还指出Connection应该是进行客户端身份验证的地方等等。

当一一个Connection被创建时，它的传输默认是关闭的，必须使用start方法开启。一个Connection可以建立一个或多个的Session。

当一个程序执行完成后，必须关闭之前创建Connection，否则ActiveMQ不能释放资源，关闭一个Connection同样也关闭了Session，MessageProducer和 MessageConsumer。

Connection createConnection();

Connection createConnection(String userName，String password);

[b]4.2 Session方法使用[/b]

一旦从ConectionFactory中获得一个Connection，必须从Connection中创建一个或者多个Session。Session是一个发送或接收消息线程，可以使用Session创建 MessageProducer, MessageConsumer和Message。

Session可以被事务化，也可以不被事务化，通常， 可以通过向Connection上的适当创建方法传递一个布尔参数对此进行设置。

Session createsession(boolean transacted ， int acknowledgeMode);

其中transacted为使用事务标识，acknowledgeMode为签收模式。

结束事务有两种方法：提交或者回滚。当一个事务提交，消息被处理。如果事务中有个步骤操作失败，事务就回滚，这个事务中已执行的动作将被撤销。在发送消息最后也必要使用session.commit()方法表示提交事务。如果你设置为false则不同提交事务，如果是true则使用session.commit()方法提交事务。

最后必须要使用签收模式有三种形式：

Session.AUTO_ACKNOWLEDGE 当客户端从receive或onMessage成功返回时。Session自动签收客户端的这条消息的收条。（建议使用自动签收模式）

Session.CLIENT_ACKNOWLEDGE客户端通过调用消息（Message）的ackonwledge方法签收消息。在这种情况下，签收发生在Session层面：签收一个已消费的消息会自动的签收这个Session所有已消费消息的收条。

Session.DUPS_OK_ACKNOWLEDGE此选项指示Session不必确保对传送消息的签收。它可能引起消息的重复，但降低了Session的开销，所以只有客户端能容忍重复的消息，才能使用。

[b]4.3 MessageProducer方法使用[/b]

MessageProducer：MessageProducer是一个由Session创建的对象，用来向Destination发送消息。

void send(Message message);

void send(Message message, int deliveryMode, int priority, long timeToLive);

void send(Destination destination, Message message);

void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive);

其中deliveryMode为传输模式，priority为消息优先级，timeToLive为消息过期时间。ActiverMQ支持两种消息传送模式：PERSISTENT和PERSISTENT两种。如果不指定传输模式，那么默认的是持久性消息。如果容忍消息丢失，那么使用非持久性消息可以改善性能和减少存储的开销。

消息优先级从0-9是个级别，0-4是普通消息，5-9是加急消息。如果不指定优先级，默认为4。需要注意的是，JMS并不一定保证按照优先级的顺序提交消息，建议使用事务，这样的准确性更高，我已经测试过，基本上按照优先级消费。

使用优先级必须要在activemq.xml文件中添加如下配置（大概在44行，在policyEntries标签内加入）：

<policyEntry queue="queue1" prioritizedMessages="true" />

这里queue要跟程序中

session.createQueue("queue1")

里面参数一致。

默认情况下，消息永远不会过期，如果消息在指定周期内，则失去意义，那么可以设置过期时间，时间单位为毫秒。

[b]4.4 MessageComsumer方法使用[/b]

MessageConsumer是由一个Session创建的对象，用来从Destination接收消息。

MessageProducer createProducer(Destination destination);

MessageConsumer createConsumer(Destination destination);

MessageConsumer createConsumer(Destination destination, String messageSelector);

MessageConsumer createConsumer(Destination destination, String name, boolean noLocal);

TopicSubscriber createDurableSubscriber(Topic topic, String name);

TopicSubscriber createDurableSubscriber(Topic topic, String name, String messageSelector, boolean noLocal);

其中messageSelector为消息选择器，noLocal标志默认为false，当设置为true时限制消费者只能接收和自己相同的连接（Connection）所发布的消息，此标志只适用于主题，不适用于队列；name标识订阅主题所对应的订阅名称，持久订阅时需要设置此参数。

Public final String SELECTOR = “name = ‘hfbin’”;该选择器检查了传入消息Name属性，并确定了这个属性的值是否等于hfbin，如果等于，则消息被消费，否者该消息会被忽略。

消息的同步和异步接收：

消息的同步接收是指客户端主动去接收消息，客户端可以采用MessageConsumer的 receive方法去接收下一个消息。

Message receive()

Message receive(long timeout)

Message receiveNoWait()

消息的异步接收是指当消息到达时，ActiveMQ主动通知客户端，可以通过注册一个实现MessageListener接囗的的对象到MessageConsumer。MessageListener只有一个必须实现的方法一一onMessage，它只接收一个参数，即Message。在为每个发送Destination的消息实现onMessage时，将调用该方法。

[b]4.5 Message方法使用[/b]

JMS程序的最终目的是生产和消费的消息能被其他程序使用，JMS的Message是一个即简单又不灵活性的基本格式，允许创建不同平台上的符合非JMS程序格式的消息，Message由以下几部分组成：消息头，属性和消息体。

BlobMessage createBlobMessage(File file)

Blobtaessage createBIobMessage(InputStream in)

BlobMessage createBIobMessage(URL url)

BlobMessage createBIobMessage(URL url, boolean deletedByBroker)

BytesMessage createBytesMessage()

MapMessage createMapMessage()

Message createMessage()

ObjectMessage createObjectMessage()

ObjectMessage createObjectMessage(SeriaIizabIe object)

TextMessage createTextMessage()

TextMessage createTextaessage(String text)

我们一般会在接收端通过instanceof方法去区别数据类型。

这里我就不一一举例说明上面的这些方法使用了，想试的自己动手，根据我的Hello World那个程序改就好了。