官网：http://activemq.apache.org/

下载windows版本

进入解压后的主文件下的bin下的win64目录,启动activemq.bat文件

访问：http://localhost:8161/ 用户名和密码 都是 admin

 

 

ActiveMQ 使用的是标准生产者和消费者模型

　　有两种数据结构 Queue、Topic

1、 Queue 队列 ，生产者生产了一个消息，只能由一个消费者进行消费

2、 Topic 话题，生产者生产了一个消息，可以由多个消费者进行消费

 

使用Java程序操作activeMQ

引入maven坐标

             
                 
      
       org.apache.activemq
                  
      
       activemq-all
                  
      
       ${activemq.version}
              
     

编写生产者代码

public class ActiveMqProducer {    @Test    public void testActive() throws Exception{        //连接工厂,使用默认的用户名,密码                ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();        //获取一个连接        Connection connection = connectionFactory.createConnection();                //建立会话        Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE );                //创建队列或者话题对象        Queue queue = session.createQueue("HelloWorld");                //创建生产者或者消费者对象        MessageProducer producer = session.createProducer(queue);                for (int i = 0; i < 10; i++) {            //发送消息            producer.send(session.createTextMessage("你好,MQ"+i));        }        //提交事物        session.commit();    }}

 

使用默认tcp连接activeMQ端口61616可以看到

编写MQ消费者代码

　　

public class ActiveMqConsumer {    @Test    public void testActive() throws Exception{        //连接工厂,使用默认的用户名,密码                ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();        //获取一个连接        Connection connection = connectionFactory.createConnection();                //开启连接        connection.start();                //建立会话,是否使用事物,如果使用则要提交        Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE );                //创建队列或者话题对象        Queue queue = session.createQueue("HelloWorld");                //创建生产者或者消费者对象        MessageConsumer consumer = session.createConsumer(queue);                while(true){            TextMessage message = (TextMessage) consumer.receive(1000);            if(message != null) {                System.out.println(message.getText());            }else {                break;            }        }    }}

再次查看发现已经被消费

 

使用监听器,监听消息的内容,进行消费

public class ActiveMqConsumerListener {    @Test    public void testActive() throws Exception{        //连接工厂,使用默认的用户名,密码                ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();        //获取一个连接        Connection connection = connectionFactory.createConnection();        connection.start();        //建立会话        Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE );                //创建队列或者话题对象        Queue queue = session.createQueue("HelloWorld");                //创建生产者或者消费者对象        MessageConsumer consumer = session.createConsumer(queue);                //设置监听器        consumer.setMessageListener(new MessageListener() {                        public void onMessage(Message message) {                TextMessage textMessage = (TextMessage) message;                try {                    System.out.println(textMessage.getText());                } catch (Exception e) {                    e.printStackTrace();                }            }        });        while(true){}    }}

 

 

spring整合ActiveMQ

引入maven坐标

　　　　
             
                 
      
       org.apache.activemq
                  
      
       activemq-all
                  
      
       ${activemq.version}
              
     

                     
                 
      
       org.springframework
                  
      
       spring-jms
                  
      
       ${spring.version}
              
     

 

引入spring整合activeMQ文件

编写spring整合active核心配置

　　　　<!-- 扫描包 -->

　　　　<context:component-scan base-package="com.learn.activemq" />

 query生产者

@Servicepublic class QueueSender {    // 注入jmsTemplate    @Autowired    @Qualifier("jmsQueueTemplate")    private JmsTemplate jmsTemplate;    public void send(String queueName, final String message) {        jmsTemplate.send(queueName, new MessageCreator() {            public Message createMessage(Session session) throws JMSException {                return session.createTextMessage(message);            }        });    }}

topic生产者

 

@Servicepublic class TopicSender {    // 注入jmsTemplate    @Autowired    @Qualifier("jmsTopicTemplate")    private JmsTemplate jmsTemplate;    public void send(String topicName, final String message) {        jmsTemplate.send(topicName, new MessageCreator() {            public Message createMessage(Session session) throws JMSException {                return session.createTextMessage(message);            }        });    }}

query消费者1和消费者2

@Servicepublic class QueueConsumer1 implements MessageListener {    public void onMessage(Message message) {        TextMessage textMessage = (TextMessage) message;        try {            System.out                    .println("消费者QueueConsumer1获取消息:" + textMessage.getText());        } catch (JMSException e) {            e.printStackTrace();        }    }}

@Servicepublic class QueueConsumer2 implements MessageListener {    public void onMessage(Message message) {        TextMessage textMessage = (TextMessage) message;        try {            System.out                    .println("消费者QueueConsumer2获取消息:" + textMessage.getText());        } catch (JMSException e) {            e.printStackTrace();        }    }}

topic消费者1和消费者2

@Servicepublic class TopicConsumer1 implements MessageListener {    public void onMessage(Message message) {        TextMessage textMessage = (TextMessage) message;        try {            System.out                    .println("消费者TopicConsumer1获取消息:" + textMessage.getText());        } catch (JMSException e) {            e.printStackTrace();        }    }}

@Servicepublic class TopicConsumer2 implements MessageListener {    public void onMessage(Message message) {        TextMessage textMessage = (TextMessage) message;        try {            System.out                    .println("消费者TopicConsumer2获取消息:" + textMessage.getText());        } catch (JMSException e) {            e.printStackTrace();        }    }}

 

配置 listener 监听器，在 applicationContext-mq-consumer.xml

生产者测试代码

@RunWith(SpringJUnit4ClassRunner.class)@ContextConfiguration(locations = "classpath:applicationContext-mq.xml")public class ProducerTest {    @Autowired    private QueueSender queueSender;    @Autowired    private TopicSender topicSender;    @Test    public void testSendMessage() {        queueSender.send("spring_queue", "queue");        topicSender.send("spring_topic", "topic");    }}

消费者测试代码

@RunWith(SpringJUnit4ClassRunner.class)@ContextConfiguration(locations = "classpath:applicationContext-mq-consumer.xml")public class ConsumerTest {    @Test    public void testConsumerMessage() {        while (true) {            // junit退出，防止进程死掉        }    }}