消息队列系列之ActiveMQ（JMS、集群配置）

1、ActiveMQ的下载与启动

到http://activemq.apache.org/activemq-5152-release.html下载ActiveMQ

windows版本的启动：

运行bin文件夹中的win32(32位系统)/win64(64位系统)下的：

activemq.bat（直接启动，不能关闭命令行窗口，否则会关闭）

InstallService.bat（以服务方式启动，可以在windows的服务中找到并设置开机自启动）

启动成功后，在浏览器中访问localhost:8161可以进入如下界面，点击进入管理界面，输入默认用户名admin密码admin进入





今后常用的Queue(消息队列)和Topics（话题）将在此处查看。

linux版本的启动

进入安装目录的bin目录下，使用./activemq start 启动，启动成功后可以远程访问activemq管理页面。例如，此linux机器的IP为192.169.1.102,则访问http://192.168.1.102:8161即可





2、消息中间件的相关概念

ActiveMQ两种基本模式

消息队列模式中，消息生产者生产的消息存在消息队列中，消息消费者从消息队列中消费消息，一条消息被一个消费者消费后，其余消费者将消费不到该消息；

发布订阅模式中，消息生产者生产的消息存在话题中，消息消费者从话题中订阅消息，一条消息可以被多个消费者消费。

使用消息中间件的好处：通过消息中间件解耦服务调用

3、JMS规范



4、普通Java程序使用JMS集成ActiveMQ

记得添加jar包:



如果使用maven记得在pom.xml中添加相应依赖:

<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.15.2</version>
</dependency>

4.1 消息队列模式

生产者

package cn.edu.shu.ces.chenjie.jms.queue;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;

public class AppProducer {
private static final String url = "tcp://192.168.1.102:61616";
private static final String queueName = "queue-test";
private static Logger Log = LoggerFactory.getLogger(AppProducer.class);
public static void main(String[] args) throws JMSException {
//1、创建ConnectionFactory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
//2、创建Connection
Connection connection = connectionFactory.createConnection();
//3、启动连接
connection.start();
//4、创建会话
//b:使用事物    应答模式
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5、创建一个目标
Destination destination = session.createQueue(queueName);
//6、创建一个生产者
MessageProducer producer = session.createProducer(destination);
for(int i = 0; i < 100; i++){
//7、创建消息
TextMessage textMessage = session.createTextMessage("test :" + i);
//8、发送消息
producer.send(textMessage);
Log.info("发送消息" + textMessage + "");
}
//9、关闭连接
connection.close();
}
}

消费者

package cn.edu.shu.ces.chenjie.jms.queue;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;

public class AppConsumer {
private static final String url = "tcp://192.168.1.102:61616";
private static final String queueName = "queue-test";
private static Logger Log = LoggerFactory.getLogger(AppProducer.class);
public static void main(String[] args) throws JMSException {
//1、创建ConnectionFactory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
//2、创建Connection
Connection connection = connectionFactory.createConnection();
//3、启动连接
connection.start();
//4、创建会话
//b:使用事物    应答模式
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5、创建一个目标
Destination destination = session.createQueue(queueName);
//6、创建一个生产者
MessageConsumer consumer = session.createConsumer(destination);

//7、创建一个监听器
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
Log.info("接收到消息:" + textMessage);
}
});
//8、关闭连接
//connection.close();
}
}

启动生产者



启动消费者



4.2 话题-订阅模式

生产者

package cn.edu.shu.ces.chenjie.jms.topic;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;

public class AppProducer {
private static final String url = "tcp://192.168.1.102:61616";
private static final String topicName = "topic-test";
private static Logger Log = LoggerFactory.getLogger(AppProducer.class);
public static void main(String[] args) throws JMSException {
//1、创建ConnectionFactory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
//2、创建Connection
Connection connection = connectionFactory.createConnection();
//3、启动连接
connection.start();
//4、创建会话
//b:使用事物    应答模式
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5、创建一个目标
Destination destination = session.createTopic(topicName);
//6、创建一个生产者
MessageProducer producer = session.createProducer(destination);
for(int i = 0; i < 100; i++){
//7、创建消息
TextMessage textMessage = session.createTextMessage("test :" + i);
//8、发送消息
producer.send(textMessage);
Log.info("发送消息" + textMessage + "");
}
//9、关闭连接
connection.close();
}
}

消费者

package cn.edu.shu.ces.chenjie.jms.topic;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;

public class AppConsumer {
private static final String url = "tcp://192.168.1.102:61616";
private static final String topicName = "topic-test";
private static Logger Log = LoggerFactory.getLogger(AppProducer.class);
public static void main(String[] args) throws JMSException {
//1、创建ConnectionFactory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
//2、创建Connection
Connection connection = connectionFactory.createConnection();
//3、启动连接
connection.start();
//4、创建会话
//b:使用事物    应答模式
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5、创建一个目标
Destination destination = session.createTopic(topicName);
//6、创建一个生产者
MessageConsumer consumer = session.createConsumer(destination);

//7、创建一个监听器
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
Log.info("接收到消息:" + textMessage);
}
});
//8、关闭连接
//connection.close();
}
}

【先】启动消费者订阅话题，【再】启动生产者产生消息



5、JavaEE(SpringJMS)中集成ActiveMQ

依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion>

<groupId>cn.edu.shu.ces.chenjie</groupId>
<artifactId>ActiveMQ-Test</artifactId>
<version>1.0-SNAPSHOT</version>

<properties>
<spring.version>4.2.5.RELEASE</spring.version>
</properties>

<dependencies>
<!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-all -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.15.2</version>
<exclusions>
<exclusion>
<artifactId>spring-context</artifactId>
<groupId>org.springframework</groupId>
</exclusion>
</exclusions>
</dependency>

<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${spring.version}</version>
</dependency>

<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${spring.version}</version>
</dependency>

</dependencies>

</project>

项目结构：



在resources中添加三个配置文件：

common.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> <!-- ActiveMQ 提供的ConnectionFactory-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://192.168.1.102:61616" />
<!--最新版的ActiveMQ想要传递对象消息，需要指定可以序列话的类所在的包-->
<property name="trustedPackages">
<list>
<value>java.lang</value>
<value>javax.security</value>
<value>java.util</value>
<value>org.apache.activemq</value>
<value>cn.edu.shu.ces.chenjie</value>
</list>
</property>
</bean>

<!--Spring JMS 提供的ConnectionFactory-->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="targetConnectionFactory">
</property>
</bean>

<!--一个队列目的地，点对点的-->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="spring-queue"/>
</bean>

<!--一个主题目的地，发布订阅模式-->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="spring-topic"/>
</bean>
</beans>

producer.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:annotation-config/>

<import resource="common.xml"/>

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"/>
</bean>

<bean id="producerService" class="cn.edu.shu.ces.chenjie.jms.spring.ProducerServiceImpl"/>
</beans>

consumer.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> <!--导入公共配置-->
<import resource="common.xml"/>

<!--配置消息监听器-->
<bean id="consumerMessageListener" class="cn.edu.shu.ces.chenjie.jms.spring.ConsumerMessageListener"/>

<!--配置消息容器-->
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<!--<property name="destination" ref="queueDestination"/>-->
<property name="destination" ref="topicDestination"/>
<property name="messageListener" ref="consumerMessageListener"/>
</bean>

</beans>

ProducerService

package cn.edu.shu.ces.chenjie.jms.spring;

public interface ProducerService {
void sendMessage(String message);
void sendLoginMessage(User user);
}

ProducerServiceImpl

package cn.edu.shu.ces.chenjie.jms.spring;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

import javax.annotation.Resource;
import javax.jms.*;

public class ProducerServiceImpl implements ProducerService{
@Autowired
JmsTemplate jmsTemplate;

@Resource(name="topicDestination")//queueDestination
Destination destination;

public void sendMessage(final String message) {
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage(message);
return textMessage;
}
});
System.out.println("发送消息:" + message);
}

public void sendLoginMessage(final User user) {
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
Message loginMessage = session.createObjectMessage(user);
return loginMessage;
}
});
System.out.println("发送用户登录消息:" + user);
}
}

ConsumerMessageListener

package cn.edu.shu.ces.chenjie.jms.spring;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;

public class ConsumerMessageListener implements MessageListener{
public void onMessage(Message message) {
System.out.println("received:" + message);
if(message instanceof ObjectMessage){
try {
User user = (User) ((ObjectMessage) message).getObject();
System.out.println("received:" + user);
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}

User

package cn.edu.shu.ces.chenjie.jms.spring;

import java.io.Serializable;

public class User implements Serializable {
private String username;
private String password;

public String getUsername() {
return username;
}

public void setUsername(String username) {
this.username = username;
}

public String getPassword() {
return password;
}

public void setPassword(String password) {
this.password = password;
}

@Override
public String toString() {
return "User{" +
"username='" + username + '\'' +
", password='" + password + '\'' +
'}';
}
}

AppProducer

package cn.edu.shu.ces.chenjie.jms.spring;

import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;

public class AppProducer {
public static void main(String[] args) {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer.xml");
ProducerService service = context.getBean(ProducerService.class);
for(int i = 0; i < 100; i ++){
service.sendMessage("test " + i);
}
User user = new User();
user.setUsername("chenjie");
user.setPassword("chenjie123");
service.sendLoginMessage(user);
context.close();
}
}

AppConsumer

package cn.edu.shu.ces.chenjie.jms.spring;

import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;

public class AppConsumer {
public static void main(String[] args) {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("consumer.xml");
}
}

启动生产者、消费者





JMS规范规定的和ActiveMQ支持的消息类型有很多，例子中使用到了文本消息和对象消息。

两种消息模式需要更改相应的配置文件（配置文件中的注释）

6、ActiveMQ集群的搭建与测试

集群结构：



NodeA不存储消息，NodeB和NodeC互斥共享消息存储，当一方占有消息存储时成为master，另一方成为slave。成为master的一方跟A同步消息，做到了负载均衡。当B/C中的master宕机时，释放消息存储，此时的slave得到消息存储成为master，做到了安全。

配置安排：

由于是示例，因此在用一台机器上的不同端口部署3个结点充当3台机器，并使用本地磁盘文件夹充当共享文件夹。

实际应该是3台机器，使用分布式文件系统当共享文件夹。



将activemq的文件夹拷贝3份，

分别取名为a b c，同时新建一个share文件夹用于消息存储共享文件夹



依次编辑各个文件夹中conf下的actvmq.xml 和jetty.xml



chenjie@pc2:~/activemq/activemq-a/conf$ vim activemq.xml

注释掉了transportConnectors标签下的后面几个配置，新增了一个networkConnectors配置，使A连接到B和C

 

<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<!--
<transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
-->
</transportConnectors>
<networkConnectors>
<networkConnector name="local_network" uri="static:(tcp://127.0.0.1:61617,tcp://127.0.0.1:61618)" />

</networkConnectors>

chenjie@pc2:~/activemq/activemq-b/conf$ vim activemq.xml

修改持久化目录

<persistenceAdapter>            <kahaDB directory="/home/chenjie/activemq/share"/>        </persistenceAdapter>

<transportConnector name="openwire" uri="tcp://0.0.0.0:61617?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<!-- <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
-->
</transportConnectors>

<networkConnectors>
<networkConnector name="network_a" uri="static:(tcp://127.0.0.1:61616)" />
</networkConnectors>


注释掉transportConnector 的后面几个配置，并增加一个networkConnectors，建立跟a的静态连接

c的activemq.xml配置与b一样

chenjie@pc2:~/activemq/activemq-a/conf$ vim jetty.xml，在此修改端口号，将a b c的端口修改为8061 8062 8062

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

依次启动a b c
依次访问：http://192.168.1.102:8161/admin http://192.168.1.102:8162/admin http://192.168.1.102:8163/admin







可以看到6183对应的c无法访问，这是因为6182对应的b抢占了共享消息存储，所以c没有提供服务。

将b杀掉：



再次访问c对应的http://192.168.1.102:8163/admin



因为b已经杀掉了所以b对应的8062显然不能显示了。这也证实了b和c能够起到保障可用性的作用。

使用集群模式时代码中需修改的地方：

生产者：

private static final String url = "failover:(tcp://192.168.1.102:61617,tcp://192.168.1.102:61618)?randomize=true";

消费者：

private static final String url = "failover:(tcp://192.168.1.61616,tcp://192.168.1.102:61617,tcp://192.168.1.102:61618)?randomize=true";