本文共 13005 字，大约阅读时间需要 43 分钟。

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中添加相应依赖:

               
    
     org.apache.activemq
                
    
     activemq-core
                
    
     5.15.2
       
   

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

依赖：

   
       
    
     4.0.0
        
    
     cn.edu.shu.ces.chenjie
        
    
     ActiveMQ-Test
        
    
     1.0-SNAPSHOT
        
            
     
      4.2.5.RELEASE
         
        
            
             
                 
      
       org.apache.activemq
                  
      
       activemq-core
                  
      
       5.15.2
                  
                      
                           
        
         spring-context
                            
        
         org.springframework
                        
                   
              
             
                 
      
       org.springframework
                  
      
       spring-context
                  
      
       ${spring.version}
              
             
                 
      
       org.springframework
                  
      
       spring-jms
                  
      
       ${spring.version}
              
         
    
   

项目结构：

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

common.xml

   
       
        
            
             
             
                 
                      
       
        java.lang
                       
       
        javax.security
                       
       
        java.util
                       
       
        org.apache.activemq
                       
       
        cn.edu.shu.ces.chenjie
                   
              
         
        
        
            
             
         
        
        
            
         
        
        
            
         
    
   

producer.xml

consumer.xml

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

 

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

修改持久化目录

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

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

c的activemq.xml配置与b一样

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

依次启动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";